Use file pool to handle running out of file descriptors.

pkg/barback/lib/src/file_pool.dart

+// 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.file_pool;

[Jennifer Messerly, 2013/10/14 21:00:08]

barback.src.file_pool?

Or did you mean to make this library public (not in "src")? I think we may need
to use this from Polymer too. Siggi would know for sure. In that case, we'd
ideally be using the same singleton instance as barback.

[Siggi Cherem (dart-lang), 2013/10/14 21:14:28]

Good point. Making it visible would simplify what we do. Alternatively, when
creating copies of files I could write:

new Asset.fromPath(path).read().pipe(new File(output).openWrite())

[Bob Nystrom, 2013/10/14 21:26:13]

We don't use ".src" in other library names in Barback. I'm not sure if we should
or not. <shrug>
No, I didn't intend it to be public. Barback doesn't expose any
file-system-level APIs.

My intent here is that if you're getting the file descriptor error because
you're reading from barback file assets, this should just invisibly solve the
problem for you.

[Bob Nystrom, 2013/10/14 21:26:13]

I'm trying to avoid exposing this functionality as a public API in barback.
Barback's job definitely isn't just to be an IO wrapper. :)
This patch may help here even without you touching the API directly. Eventually
those reads will fail because you have too many files open (for read and write).
When some of the previous ones finish, both readers and writers will close.
Barback will see the reads close and try again.

If you'd like to give this a shot and see if it helps, I'd definitely appreciate
it.

At least in the short term, I'm trying to keep this solution an implementation
detail of barback. If we have to make an actual public API so that other code
can reuse this solution, we can investigate that, but that's a lot more work.

[Jennifer Messerly, 2013/10/14 21:30:42]

trivial detail but: regarding library name -- shouldn't it be consistent with
our hoped-for-import name? e.g. "import barback.src.file_pool;" maybe someday?

[Bob Nystrom, 2013/10/14 22:06:58]

Hmm, good point. I'm up for changing all of these library names at some point,
but I'd rather keep them consistent and I don't want to change them all in this
patch.

+
+import 'dart:async';
+import 'dart:collection';
+import 'dart:io';
+
+/// Manages a pool of files that are opened for reading to cope with maximum
+/// file descriptor limits.
+///
+/// If a file cannot be opened because too many files are already open, this
+/// will defer the open until a previously opened file is closed and then try
+/// 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 {
+  // TODO(rnystrom): Should we cap this to some maximum size?
+  final _pendingListens = new Queue<_FileReader>();
+
+  /// 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 to file to be closed and
+  /// then try again.
+  Stream<List<int>> openRead(File file) => new _FileReader(this, file).stream;

[Siggi Cherem (dart-lang), 2013/10/14 21:14:28]

if we make it public, can we also add openWrite?

[Bob Nystrom, 2013/10/14 21:26:13]

See above. Trying to not make it public. :)

+
+  void _retryPendingListen() {
+    if (_pendingListens.isEmpty) return;
+
+    var pending = _pendingListens.removeFirst();
+    pending._listen();
+  }
+}
+
+/// 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;
+
+  /// The underyling file stream.
+  Stream<List<int>> _fileStream;
+
+  /// The controller for the wrapped stream.
+  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;
+  Timer _timer;
+
+  /// When a [listen] call has thrown a "too many files" error, this will be
+  /// the exception object.
+  Object _exception;
+
+  /// When a [listen] call has thrown a "too many files" error, this will be
+  /// the captured stack trace.
+  Object _stackTrace;
+
+  /// The wrapped stream that the file can be read from.
+  Stream<List<int>> get stream {
+    if (_controller != null) return _controller.stream;
+
+    _controller = new StreamController<List<int>>(onListen: _listen,
+        onPause: () {
+      _subscription.pause();
+    }, onResume: () {
+      _subscription.resume();
+    }, onCancel: () {
+      if (_subscription != null) _subscription.cancel();
+      _subscription = null;
+    }, sync: true);
+
+    return _controller.stream;
+  }
+
+  _FileReader(this._pool, this._file);
+
+  /// Starts listening to the underlying file stream.
+  void _listen() {
+    if (_timer != null) {
+      _timer.cancel();
+      _timer = null;
+    }
+
+    _fileStream = _file.openRead();
+    _subscription = _fileStream.listen(_controller.add,
+        onError: _onError, onDone: _onDone, cancelOnError: true);
+  }
+
+  /// Handles an error from the underlying file stream.
+  ///
+  /// "Too many file" errors are caught so that we can retry later. Other
+  /// errors are passed to the wrapped stream and the underlying stream
+  /// subscription is canceled.
+  void _onError(Object exception, Object stackTrace) {
+    assert(_subscription != null);
+    assert(_exception == null);
+
+    // The subscription is canceled after an error.
+    _subscription = null;
+
+    // We only handle "Too many open files errors".
+    if (exception is! FileException || exception.osError.errorCode != 24) {
+      // TODO(bob): stack trace.
+      _controller.addError(exception, stackTrace);
+      return;
+    }
+
+    // TODO(bob): What if already deferred?
+    _exception = exception;
+    _stackTrace = stackTrace;
+
+    // We'll try to defer the listen in the hopes that another file will close
+    // and we can try. If that doesn't happen after a while, give up and just
+    // throw the original error.
+    // TODO(bob): How long?
+    _timer = new Timer(new Duration(seconds: 5), _onTimeout);
+
+    // Tell the pool that this file is waiting.
+    _pool._pendingListens.add(this);
+  }
+
+  /// Handles the underlying file stream finishing.
+  void _onDone() {
+    _subscription = null;
+
+    _controller.close();
+    _pool._retryPendingListen();
+  }
+
+  /// If this file failed to be read because there were too many open files and
+  /// no file was closed in time to retry, this handles giving up.
+  void _onTimeout() {
+    assert(_subscription == null);
+    assert(_exception != null);
+
+    // We failed to open in time, so just fail with the original error.
+    _pool._pendingListens.remove(this);
+    _controller.addError(_exception, _stackTrace);
+    _controller.close();
+
+    _timer = null;
+    _exception = null;
+    _stackTrace = null;
+
+  }
+}