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;
+
+import 'dart:async';
+import 'dart:collection';
+import 'dart:convert';
+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?

[nweiz, 2013/10/16 00:16:01]

I don't think so. Let the "too many files" error be the limit.

[Bob Nystrom, 2013/10/16 00:51:46]

Done.

+  final _pendingListens = new Queue<_FileReader>();

[nweiz, 2013/10/16 00:16:01]

Document this.

[Bob Nystrom, 2013/10/16 00:51:46]

Done.

+
+  /// 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

[nweiz, 2013/10/16 00:16:01]

"opened to file" -> "opened file"

[Bob Nystrom, 2013/10/16 00:51:46]

Done.

+  /// then try again.
+  Stream<List<int>> openRead(File file) => new _FileReader(this, file).stream;
+
+  /// Reads [file] as a string using [encoding].
+  ///
+  /// If there are too many files open and the read fails, this will wait for
+  /// a previously opened file to be closed and then try again.
+  Future<String> readAsString(File file, Encoding encoding) {
+    return _readAsBytes(file).then(encoding.decode);
+  }
+
+  /// Reads [file] as a list of bytes, using [openRead] to retry if there are
+  /// failures.
+  Future<List<int>> _readAsBytes(File file) {
+    var completer = new Completer<List<int>>();
+    var builder = new BytesBuilder();
+
+    openRead(file).listen(builder.add, onDone: () {
+      completer.complete(builder.takeBytes());
+    }, onError: completer.completeError, cancelOnError: true);
+
+    return completer.future;
+  }
+
+  /// Tries to reopen the next pending open if there are any.

[nweiz, 2013/10/16 00:16:01]

"reopen" -> "restart"

Also a little confusing that you call it an "open" here and a "listen" in the
code.

[Bob Nystrom, 2013/10/16 00:51:46]

Fixed. The original implementation retried on open(), not listen().

+  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.

[nweiz, 2013/10/16 00:16:01]

"wrapped" here isn't accurate; the wrapped stream is [_fileStream]. Maybe
"wrapping stream"?

[Bob Nystrom, 2013/10/16 00:51:46]

"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;
+  Timer _timer;

[nweiz, 2013/10/16 00:16:01]

Document this.

[Bob Nystrom, 2013/10/16 00:51:46]

Done.

+
+  /// 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,

[nweiz, 2013/10/16 00:16:01]

Why isn't this being initialized in the constructor? [stream] is always called
immediately after the FileReader is created anyway.

[Bob Nystrom, 2013/10/16 00:51:46]

Done.

+        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);

[nweiz, 2013/10/16 00:16:01]

This doesn't seem right. It's definitely possible for the same FileReader to get
multiple "too many files" errors.

[Bob Nystrom, 2013/10/16 00:51:46]

Changed this to clear _exception in _listen(). Since the subscription is
cancelled on an error, that should ensure that _exception will be null when we
get here.

+
+    // 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) {

[nweiz, 2013/10/16 00:16:01]

We should figure out what error code the exception will produce on Windows and
check for that here as well.

[Bob Nystrom, 2013/10/16 00:51:46]

I did a little checking and I couldn't find much info on this. Once it lands, we
can see how it goes on the bots.

+      _controller.addError(exception, stackTrace);
+      return;
+    }
+
+    _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(rnystrom): How long should this delay be?

[nweiz, 2013/10/16 00:16:01]

The chance of a deadlock here is extremely small, so I'm happy with making the
timeout very long. I don't think this timeout strategy is quite correct, though.
Suppose there are many readers queued up and being slowly dequeued. We're now
under the file limit, so each one that's dequeued is working correctly. But the
dequeuing is slow enough that one of the last readers to be added can still time
out, even though there's no deadlock. This could be solved by putting the timer
on the FilePool, and resetting it whenever a file is successfully opened.

[Bob Nystrom, 2013/10/16 00:51:46]

Good call.

Increased the timeout and added a long TODO about this since I don't have time
to get to it today.

+    _timer = new Timer(new Duration(seconds: 10), _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;

[nweiz, 2013/10/16 00:16:01]

I don't know how I feel about nulling out all the mutable instance variables
whenever we know a class's lifecycle is done. On one hand, it's safe because the
lifecycle is done, and there's some chance it catches a bug. On the other hand,
it's probably unnecessary, since this will get garbage collected anyway.

[Bob Nystrom, 2013/10/16 00:51:46]

Yeah, I'm mainly doing it as an ad-hoc state machine so I can assert on them in
the various methods and sanity check what's going on.

+

[nweiz, 2013/10/16 00:16:01]

Nit: extra newline.

[Bob Nystrom, 2013/10/16 00:51:46]

Done.

+  }
+}