Only run at most 10 transformers at once in barback.

BUG=14320
[email protected]

Review URL: https://codereview.chromium.org//36213002

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@29060 260f80e4-7a28-3924-810f-c04153c831b5

Author: nex3

pkg/barback/lib/src/file_pool.dart

@@ -9,8 +9,7 @@ import 'dart:collection';
 import 'dart:convert';
 import 'dart:io';
 
-import 'package:stack_trace/stack_trace.dart';
-
+import 'pool.dart';
 import 'utils.dart';
 
 /// Manages a pool of files that are opened for reading to cope with maximum
@@ -21,42 +20,26 @@ import 'utils.dart';
 /// again. If this doesn't succeed after a certain amount of time, the open
 /// will fail and the original "too many files" exception will be thrown.
 class FilePool {
-  /// [_FileReader]s whose last [listen] call failed and that are waiting for
-  /// another file to close so they can be retried.
-  final _pendingListens = new Queue<_FileReader>();
-
-  /// The timeout timer.
-  ///
-  /// This timer is set as soon as the file limit is reached and is reset every
-  /// time a file finishes being read or a new file is opened. If it fires, that
-  /// indicates that the caller became deadlocked, likely due to files waiting
-  /// for additional files to be read before they could be closed.
-  Timer _timer;
-
-  /// The number of files currently open in the pool.
-  int _openFiles = 0;
-
-  /// The maximum number of file descriptors that the pool will allocate.
+  /// The underlying pool.
   ///
-  /// This is based on empirical tests that indicate that beyond 32, additional
-  /// file reads don't provide substantial additional throughput.
-  final int _maxOpenFiles = 32;
+  /// The maximum number of allocated descriptors is based on empirical tests
+  /// that indicate that beyond 32, additional file reads don't provide
+  /// substantial additional throughput.
+  final Pool _pool = new Pool(32, timeout: new Duration(seconds: 60));
 
   /// Opens [file] for reading.
   ///
   /// When the returned stream is listened to, if there are too many files
   /// open, this will wait for a previously opened file to be closed and then
   /// try again.
   Stream<List<int>> openRead(File file) {
-    var reader = new _FileReader(this, file);
-    if (_openFiles < _maxOpenFiles) {
-      _openFiles++;
-      reader.start();
-    } else {
-      _pendingListens.add(reader);
-      _heartbeat();
-    }
-    return reader.stream;
+    return futureStream(_pool.checkOut().then((resource) {
+      return file.openRead().transform(new StreamTransformer.fromHandlers(
+          handleDone: (sink) {
+        sink.close();
+        resource.release();
+      }));
+    }));
   }
 
   /// Reads [file] as a string using [encoding].
@@ -79,107 +62,4 @@ class FilePool {
 
     return completer.future;
   }
-
-  /// If there are any file reads that are waiting for available descriptors,
-  /// this will allow the oldest one to start reading.
-  void _startPendingListen() {
-    if (_pendingListens.isEmpty) {
-      _openFiles--;
-      if (_timer != null) {
-        _timer.cancel();
-        _timer = null;
-      }
-      return;
-    }
-
-    _heartbeat();
-    var pending = _pendingListens.removeFirst();
-    pending.start();
-  }
-
-  /// Indicates that some external action has occurred and the timer should be
-  /// restarted.
-  void _heartbeat() {
-    if (_timer != null) _timer.cancel();
-    _timer = new Timer(new Duration(seconds: 60), _onTimeout);
-  }
-
-  /// Handles [_timer] timing out by causing all pending file readers to emit
-  /// exceptions.
-  void _onTimeout() {
-    for (var reader in _pendingListens) {
-      reader.timeout();
-    }
-    _pendingListens.clear();
-    _timer = null;
-  }
-}
-
-/// Wraps a raw file reading stream in a stream that handles "too many files"
-/// errors.
-///
-/// This also notifies the pool when the underlying file stream is closed so
-/// that it can try to open a waiting file.
-class _FileReader {
-  final FilePool _pool;
-  final File _file;
-
-  /// Whether the caller has paused this reader's stream.
-  bool _isPaused = false;
-
-  /// The underyling file stream.
-  Stream<List<int>> _fileStream;
-
-  /// The controller for the stream wrapper.
-  StreamController<List<int>> _controller;
-
-  /// The current subscription to the underlying file stream.
-  ///
-  /// This will only be non-null while the wrapped stream is being listened to.
-  StreamSubscription _subscription;
-
-  /// The wrapped stream that the file can be read from.
-  Stream<List<int>> get stream => _controller.stream;
-
-  _FileReader(this._pool, this._file) {
-    _controller = new StreamController<List<int>>(onPause: () {
-      _isPaused = true;
-      if (_subscription != null) _subscription.pause();
-    }, onResume: () {
-      _isPaused = false;
-      if (_subscription != null) _subscription.resume();
-    }, onCancel: () {
-      if (_subscription != null) _subscription.cancel();
-      _subscription = null;
-    }, sync: true);
-  }
-
-  /// Starts listening to the underlying file stream.
-  void start() {
-    _fileStream = _file.openRead();
-    _subscription = _fileStream.listen(_controller.add,
-        onError: _onError, onDone: _onDone, cancelOnError: true);
-    if (_isPaused) _subscription.pause();
-  }
-
-  /// Emits a timeout exception.
-  void timeout() {
-    assert(_subscription == null);
-    _controller.addError("FilePool deadlock: all file descriptors have been in "
-        "use for too long.", new Trace.current().vmTrace);
-    _controller.close();
-  }
-
-  /// Forwards an error from the underlying file stream.
-  void _onError(Object exception, StackTrace stackTrace) {
-    _controller.addError(exception, stackTrace);
-    _onDone();
-  }
-
-  /// Handles the underlying file stream finishing.
-  void _onDone() {
-    _subscription = null;
-    _controller.close();
-    _pool._startPendingListen();
-  }
 }

pkg/barback/lib/src/package_graph.dart

@@ -14,6 +14,7 @@ import 'barback_logger.dart';
 import 'build_result.dart';
 import 'errors.dart';
 import 'package_provider.dart';
+import 'pool.dart';
 import 'transformer.dart';
 import 'utils.dart';
 
@@ -65,6 +66,13 @@ class PackageGraph {
   /// [Future] returned by [getAllAssets].
   var _lastUnexpectedError;
 
+  // TODO(nweiz): Allow transformers to declare themselves as "lightweight" or
+  // "heavyweight" and adjust their restrictions appropriately. Simple
+  // transformers may be very efficient to run in parallel, whereas dart2js uses
+  // a lot of memory and should be run more sequentially.
+  /// A pool that controls how many transformers may be applied at once.
+  final Pool transformPool = new Pool(10);
+
   /// Creates a new [PackageGraph] that will transform assets in all packages
   /// made available by [provider].
   PackageGraph(this.provider, {BarbackLogger logger})

pkg/barback/lib/src/pool.dart

@@ -0,0 +1,133 @@
+import 'dart:async';
+import 'dart:collection';
+
+import 'package:stack_trace/stack_trace.dart';
+
+/// Manages an abstract pool of resources with a limit on how many may be in use
+/// at once.
+///
+/// When a resource is needed, the user should call [checkOut]. When the
+/// returned future completes with a [PoolResource], the resource may be
+/// allocated. Once the resource has been released, the user should call
+/// [PoolResource.release]. The pool will ensure that only a certain number of
+/// [PoolResource]s may be checked out at once.
+class Pool {
+  /// Completers for checkouts beyond the first [_maxCheckedOutResources].
+  ///
+  /// When an item is released, the next element of [_pendingResources] will be
+  /// completed.
+  final _pendingResources = new Queue<Completer<PoolResource>>();
+
+  /// The maximum number of resources that may be checked out at once.
+  final int _maxCheckedOutResources;
+
+  /// The number of resources that are currently checked out.
+  int _checkedOutResources = 0;
+
+  /// The timeout timer.
+  ///
+  /// If [_timeout] isn't null, this timer is set as soon as the resource limit
+  /// is reached and is reset every time an resource is released or a new
+  /// resource is requested. If it fires, that indicates that the caller became
+  /// deadlocked, likely due to files waiting for additional files to be read
+  /// before they could be closed.
+  Timer _timer;
+
+  /// The amount of time to wait before timing out the pending resources.
+  Duration _timeout;
+
+  /// Creates a new pool with the given limit on how many resources may be
+  /// checked out at once.
+  ///
+  /// If [timeout] is passed, then if that much time passes without any activity
+  /// all pending [checkOut] futures will throw an exception. This is indented
+  /// to avoid deadlocks.
+  Pool(this._maxCheckedOutResources, {Duration timeout})
+      : _timeout = timeout;
+
+  /// Check out a [PoolResource].
+  ///
+  /// If the maximum number of resources is already checked out, this will delay
+  /// until one of them is released.
+  Future<PoolResource> checkOut() {
+    if (_checkedOutResources < _maxCheckedOutResources) {
+      _checkedOutResources++;
+      return new Future.value(new PoolResource._(this));
+    } else {
+      var completer = new Completer<PoolResource>();
+      _pendingResources.add(completer);
+      _heartbeat();
+      return completer.future;
+    }
+  }
+
+  /// Checks out a resource for the duration of [callback], which may return a
+  /// Future.
+  ///
+  /// The return value of [callback] is piped to the returned Future.
+  Future withResource(callback()) {
+    return checkOut().then((resource) =>
+        new Future.sync(callback).whenComplete(resource.release));
+  }
+
+  /// If there are any pending checkouts, this will fire the oldest one.
+  void _onResourceReleased() {
+    if (_pendingResources.isEmpty) {
+      _checkedOutResources--;
+      if (_timer != null) {
+        _timer.cancel();
+        _timer = null;
+      }
+      return;
+    }
+
+    _heartbeat();
+    var pending = _pendingResources.removeFirst();
+    pending.complete(new PoolResource._(this));
+  }
+
+  /// Indicates that some external action has occurred and the timer should be
+  /// restarted.
+  void _heartbeat() {
+    if (_timer != null) _timer.cancel();
+    if (_timeout == null) {
+      _timer = null;
+    } else {
+      _timer = new Timer(_timeout, _onTimeout);
+    }
+  }
+
+  /// Handles [_timer] timing out by causing all pending resource completers to
+  /// emit exceptions.
+  void _onTimeout() {
+    for (var completer in _pendingResources) {
+      completer.completeException("Pool deadlock: all resources have been "
+          "checked out for too long.", new Trace.current().vmTrace);
+    }
+    _pendingResources.clear();
+    _timer = null;
+  }
+}
+
+/// A member of a [Pool].
+///
+/// A [PoolResource] is a token that indicates that a resource is allocated.
+/// When the associated resource is released, the user should call [release].
+class PoolResource {
+  final Pool _pool;
+
+  /// Whether [this] has been released yet.
+  bool _released = false;
+
+  PoolResource._(this._pool);
+
+  /// Tells the parent [Pool] that the resource associated with this resource is
+  /// no longer allocated, and that a new [PoolResource] may be checked out.
+  void release() {
+    if (_released) {
+      throw new StateError("A PoolResource may only be released once.");
+    }
+    _released = true;
+    _pool._onResourceReleased();
+  }
+}

pkg/barback/lib/src/transform_node.dart

@@ -127,7 +127,10 @@ class TransformNode {
     _inputSubscriptions.clear();
 
     _isDirty = false;
-    return transformer.apply(transform).catchError((error) {
+
+    return phase.cascade.graph.transformPool
+        .withResource(() => transformer.apply(transform))
+        .catchError((error) {
       // If the transform became dirty while processing, ignore any errors from
       // it.
       if (_isDirty) return;

pkg/barback/test/package_graph/many_parallel_transformers_test.dart

@@ -0,0 +1,33 @@
+// Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+library barback.test.package_graph.transform_test;
+
+import 'package:barback/src/utils.dart';
+import 'package:scheduled_test/scheduled_test.dart';
+
+import '../utils.dart';
+
+main() {
+  initConfig();
+
+  test("handles many parallel transformers", () {
+    var files = new List.generate(100, (i) => "app|$i.txt");
+    var rewrite = new RewriteTransformer("txt", "out");
+    initGraph(files, {"app": [[rewrite]]});
+
+    // Pause and resume apply to simulate parallel long-running transformers.
+    rewrite.pauseApply();
+    updateSources(files);
+    schedule(pumpEventQueue);
+    rewrite.resumeApply();
+
+    for (var i = 0; i < 100; i++) {
+      expectAsset("app|$i.out", "$i.out");
+    }
+    buildShouldSucceed();
+
+    expect(rewrite.maxParallelRuns, completion(equals(10)));
+  });
+}

pkg/barback/test/transformer/mock.dart

@@ -27,6 +27,9 @@ abstract class MockTransformer extends Transformer {
   Future<int> get numRuns => schedule(() => _numRuns);
   var _numRuns = 0;
 
+  Future<int> get maxParallelRuns => schedule(() => _maxParallelRuns);
+  var _maxParallelRuns = 0;
+
   /// The number of currently running transforms.
   int _runningTransforms = 0;
 
@@ -176,6 +179,9 @@ abstract class MockTransformer extends Transformer {
     _numRuns++;
     if (_runningTransforms == 0) _started.complete();
     _runningTransforms++;
+    if (_runningTransforms > _maxParallelRuns) {
+      _maxParallelRuns = _runningTransforms;
+    }
     return newFuture(() => doApply(transform)).then((_) {
       if (_apply != null) return _apply.future;
     }).whenComplete(() {