Re-implement directory polling.

pkg/watcher/lib/src/directory_watcher.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 watcher.directory_watcher;
 
 import 'dart:async';
+import 'dart:collection';
 import 'dart:io';
 
 import 'package:crypto/crypto.dart';
 
 import 'stat.dart';
 import 'watch_event.dart';
 
 /// Watches the contents of a directory and emits [WatchEvent]s when something
 /// in the directory has changed.
 class DirectoryWatcher {
   /// The directory whose contents are being monitored.
   final String directory;
 
   /// The broadcast [Stream] of events that have occurred to files in
   /// [directory].
   ///
   /// Changes will only be monitored while this stream has subscribers. Any
   /// file changes that occur during periods when there are no subscribers
   /// will not be reported the next time a subscriber is added.
   Stream<WatchEvent> get events => _events.stream;
   StreamController<WatchEvent> _events;
 
-  _WatchState _state = _WatchState.notWatching;
+  _WatchState _state = _WatchState.UNSUBSCRIBED;
 
   /// A [Future] that completes when the watcher is initialized and watching
   /// for file changes.
   ///
   /// If the watcher is not currently monitoring the directory (because there
   /// are no subscribers to [events]), this returns a future that isn't
   /// complete yet. It will complete when a subscriber starts listening and
   /// the watcher finishes any initialization work it needs to do.
   ///
   /// If the watcher is already monitoring, this returns an already complete
   /// future.
   Future get ready => _ready.future;
   Completer _ready = new Completer();
 
   /// The amount of time the watcher pauses between successive polls of the
   /// directory contents.
   final Duration pollingDelay;
 
   /// The previous status of the files in the directory.
   ///
   /// Used to tell which files have been modified.
   final _statuses = new Map<String, _FileStatus>();
 
+  /// The subscription used while [directory] is being listed.
+  ///
+  /// Will be `null` if a list is not currently happening.
+  StreamSubscription<FileSystemEntity> _listSubscription;
+
+  /// This will be `true` if we are currently asynchronously processing files
+  /// from [_filesToCheck].

[nweiz, 2013/08/01 22:52:39]

"_filesToCheck" -> "_filesToProcess"

[Bob Nystrom, 2013/08/02 17:26:29]

Done.

+  bool _isProcessing = false;
+
+  /// The queue of files waiting to be processed to see if they have been
+  /// modified.
+  ///
+  /// Processing a file is asynchronous, as is listing the directory, so the
+  /// queue exists to let each of those proceed at their own rate. The lister
+  /// will enqueue files as quickly as it can. Meanwhile, files are dequeued
+  /// and processed sequentially.
+  final _filesToProcess = new Queue<String>();

[nweiz, 2013/08/01 22:52:39]

See how this looks with a FutureGroup instead of a queue, as discussed in
person.

[Bob Nystrom, 2013/08/02 17:26:29]

I started going in that direction but it looks like it will be a lot of work for
unknown benefit. How about we stick with the current approach?

[nweiz, 2013/08/02 20:07:52]

SGTM. Can you abstract out the notion of a sequential stream into a class or
method? An API as simple as this should work:

    class SequentialStream<T> {
      SequentialStream(Stream<T> inner, Future callback<T>);
      Future get onDone;
      void cancel();
    }

[Bob Nystrom, 2013/08/02 21:45:11]

Done.

+
+  /// The set of files that have been seen in the current directory listing.
+  ///
+  /// Used to tell which files have been removed: files that are in [_statuses]
+  /// but not in here when a poll completes have been removed.
+  final _polledFiles = new Set<String>();
+
   /// Creates a new [DirectoryWatcher] monitoring [directory].
   ///
   /// If [pollingDelay] is passed, it specifies the amount of time the watcher
   /// will pause between successive polls of the directory contents. Making
   /// this shorter will give more immediate feedback at the expense of doing
   /// more IO and higher CPU usage. Defaults to one second.
   DirectoryWatcher(this.directory, {Duration pollingDelay})
       : pollingDelay = pollingDelay != null ? pollingDelay :
                                               new Duration(seconds: 1) {
-    _events = new StreamController<WatchEvent>.broadcast(onListen: () {
-      _state = _state.listen(this);
-    }, onCancel: () {
-      _state = _state.cancel(this);
+    _events = new StreamController<WatchEvent>.broadcast(
+        onListen: _watch, onCancel: _cancel);
+  }
+
+  /// Scans to see which files were already present before the watcher was
+  /// subscribed to, and then starts watching the directory for changes.
+  void _watch() {
+    assert(_state == _WatchState.UNSUBSCRIBED);
+    _state = _WatchState.SCANNING;
+    _poll();
+  }
+
+  /// Stops watching the directory when there are no more subscribers.
+  void _cancel() {
+    assert(_state != _WatchState.UNSUBSCRIBED);
+    _state = _WatchState.UNSUBSCRIBED;
+
+    // If we're in the middle of listing the directory, stop.
+    if (_listSubscription != null) _listSubscription.cancel();
+
+    // Don't process any remaining files.
+    _filesToProcess.clear();
+    _polledFiles.clear();

[nweiz, 2013/08/01 22:52:39]

Also clear _statuses.

[Bob Nystrom, 2013/08/02 17:26:29]

Done.

+
+    _ready = new Completer();
+  }
+
+  /// Scans the contents of the directory once to see which files have been
+  /// added, removed, and modified.
+  void _poll() {
+    _filesToProcess.clear();
+    _polledFiles.clear();
+
+    var stream = new Directory(directory).list(recursive: true);
+    _listSubscription = stream.listen((entity) {
+      assert(_state != _WatchState.UNSUBSCRIBED);
+
+      if (entity is! File) return;
+      _enqueueFile(entity.path);
+    }, onDone: () {
+      assert(_state != _WatchState.UNSUBSCRIBED);
+      _listSubscription = null;
+
+      // Null tells the queue consumer that we're done listing.
+      _enqueueFile(null);
     });
   }
 
-  /// Starts the asynchronous polling process.
-  ///
-  /// Scans the contents of the directory and compares the results to the
-  /// previous scan. Loops to continue monitoring as long as there are
-  /// subscribers to the [events] stream.
-  Future _watch() {
-    var files = new Set<String>();
+  /// Add [path] to the queue of files whose status needs to be processed.
+  void _enqueueFile(String path) {
+    _filesToProcess.add(path);
 
-    var stream = new Directory(directory).list(recursive: true);
+    // Start up the asynchronous processing if not already running.
+    if (_isProcessing) return;
+    _isProcessing = true;
 
-    return stream.map((entity) {
-      if (entity is! File) return new Future.value();
-      files.add(entity.path);
-      // TODO(rnystrom): These all run as fast as possible and read the
-      // contents of the files. That means there's a pretty big IO hit all at
-      // once. Maybe these should be queued up and rate limited?
-      return _refreshFile(entity.path);
-    }).toList().then((futures) {
-      // Once the listing is done, make sure to wait until each file is also
-      // done.
-      return Future.wait(futures);
-    }).then((_) {
-      var removedFiles = _statuses.keys.toSet().difference(files);
-      for (var removed in removedFiles) {
-        if (_state.shouldNotify) {
-          _events.add(new WatchEvent(ChangeType.REMOVE, removed));
+    refreshNextFile() {

[nweiz, 2013/08/01 22:52:39]

It feels like this warrants being its own method.

[Bob Nystrom, 2013/08/02 17:26:29]

Done.

+      var file = _filesToProcess.removeFirst();
+
+      // `null` is the sentinel which means the directory listing is complete.
+      if (file == null) {
+        _isProcessing = false;
+
+        // Any files that were not seen in the last poll but that we have a
+        // status for must have been removed.
+        var removedFiles = _statuses.keys.toSet().difference(_polledFiles);
+        for (var removed in removedFiles) {
+          if (_state == _WatchState.WATCHING) {
+            _events.add(new WatchEvent(ChangeType.REMOVE, removed));
+          }
+          _statuses.remove(removed);
         }
-        _statuses.remove(removed);
+
+        assert(_state != _WatchState.UNSUBSCRIBED);

[nweiz, 2013/08/01 22:52:39]

This should be at the top of [refreshNextFile].

[Bob Nystrom, 2013/08/02 17:26:29]

Done.

+        if (_state == _WatchState.SCANNING) {
+          _state = _WatchState.WATCHING;
+          print("done scanning, watching");

[nweiz, 2013/08/01 22:52:39]

Left over from debugging?

[Bob Nystrom, 2013/08/02 17:26:29]

Yup. Done.

+          _ready.complete();
+        }
+
+        // Wait.
+        return new Future.delayed(pollingDelay).then((_) {
+          // Stop if we unsubscribed while waiting.
+          if (_state == _WatchState.UNSUBSCRIBED) return;
+
+          // And then poll again.
+          _poll();
+        });
       }
 
-      var previousState = _state;
-      _state = _state.finish(this);
+      return _processFile(file).then((_) {
+        // Stop if we unsubscribed while waiting.
+        if (_state == _WatchState.UNSUBSCRIBED) return;
 
-      // If we were already sending notifications, add a bit of delay before
-      // restarting just so that we don't whale on the file system.
-      // TODO(rnystrom): Tune this and/or make it tunable?
-      if (_state.shouldNotify) {
-        return new Future.delayed(pollingDelay);
-      }
-    }).then((_) {
-      // Make sure we haven't transitioned to a non-watching state during the
-      // delay.
-      if (_state.shouldWatch) _watch();
+        // If we have drained the queue (i.e. we are processing faster than
+        // we are listing files), then stop processing and wait until something
+        // has been enqueued.
+        if (_filesToProcess.isEmpty) {
+          _isProcessing = false;
+          return;
+        }
+
+        return refreshNextFile();
+      });
+    }
+
+    // Pipe all errors to the notification stream.
+    refreshNextFile().catchError((error) {
+      _events.addError(error, getAttachedStackTrace(error));

[nweiz, 2013/08/01 22:52:39]

You don't need to manually forward the stack trace.

[Bob Nystrom, 2013/08/02 17:26:29]

Done.

     });
   }
 
-  /// Compares the current state of the file at [path] to the state it was in
-  /// the last time it was scanned.
-  Future _refreshFile(String path) {
+  Future _processFile(String path) {
     return getModificationTime(path).then((modified) {
+      // Stop if we unsubscribed while waiting.
+      if (_state == _WatchState.UNSUBSCRIBED) return;
+
       var lastStatus = _statuses[path];
 
       // If it's modification time hasn't changed, assume the file is unchanged.

[nweiz, 2013/08/01 22:52:39]

"it's" -> "its"

[Bob Nystrom, 2013/08/02 17:26:29]

Done.

-      if (lastStatus != null && lastStatus.modified == modified) return;
+      if (lastStatus != null && lastStatus.modified == modified) {
+        // The file is still here.
+        _polledFiles.add(path);
+        return;
+      }
 
       return _hashFile(path).then((hash) {
+        // Stop if we unsubscribed while waiting.
+        if (_state == _WatchState.UNSUBSCRIBED) return;
+
         var status = new _FileStatus(modified, hash);
         _statuses[path] = status;
+        _polledFiles.add(path);
 
-        // Only notify if the file contents changed.
-        if (_state.shouldNotify &&
-            (lastStatus == null || !_sameHash(lastStatus.hash, hash))) {
-          var change = lastStatus == null ? ChangeType.ADD : ChangeType.MODIFY;
-          _events.add(new WatchEvent(change, path));
+        // Only notify while in the watching state.
+        if (_state == _WatchState.WATCHING) {

[nweiz, 2013/08/01 22:52:39]

Short-circuit here and below.

[Bob Nystrom, 2013/08/02 17:26:29]

Done.

+          var changed = lastStatus == null || !_sameHash(lastStatus.hash, hash);
+          if (changed) {
+            var type = lastStatus == null ? ChangeType.ADD : ChangeType.MODIFY;
+            _events.add(new WatchEvent(type, path));
+          }
         }
       });
     });
   }
 
   /// Calculates the SHA-1 hash of the file at [path].
   Future<List<int>> _hashFile(String path) {
     return new File(path).readAsBytes().then((bytes) {
       var sha1 = new SHA1();
       sha1.add(bytes);
       return sha1.close();
     });
   }
 
   /// Returns `true` if [a] and [b] are the same hash value, i.e. the same
   /// series of byte values.
   bool _sameHash(List<int> a, List<int> b) {
     // Hashes should always be the same size.
     assert(a.length == b.length);
 
     for (var i = 0; i < a.length; i++) {
       if (a[i] != b[i]) return false;
     }
 
     return true;
   }
 }
 
-/// An "event" that is sent to the [_WatchState] FSM to trigger state
-/// transitions.
-typedef _WatchState _WatchStateEvent(DirectoryWatcher watcher);
+class _WatchState {
+  static const UNSUBSCRIBED = const _WatchState("unsubscribed");
+  static const SCANNING = const _WatchState("scanning");
+  static const WATCHING = const _WatchState("watching");

[nweiz, 2013/08/01 22:52:39]

Add some docs about what each state means.

[Bob Nystrom, 2013/08/02 17:26:29]

Done.

 
-/// The different states that the watcher can be in and the transitions between
-/// them.
-///
-/// This class defines a finite state machine for keeping track of what the
-/// asynchronous file polling is doing. Each instance of this is a state in the
-/// machine and its [listen], [cancel], and [finish] fields define the state
-/// transitions when those events occur.
-class _WatchState {
-  /// The watcher has no subscribers.
-  static final notWatching = new _WatchState(
-      listen: (watcher) {
-    watcher._watch();
-    return _WatchState.scanning;
-  });
+  final String name;
 
-  /// The watcher has subscribers and is scanning for pre-existing files.
-  static final scanning = new _WatchState(
-      cancel: (watcher) {
-    // No longer watching, so create a new incomplete ready future.
-    watcher._ready = new Completer();
-    return _WatchState.cancelling;
-  }, finish: (watcher) {
-    watcher._ready.complete();
-    return _WatchState.watching;
-  }, shouldWatch: true);
-
-  /// The watcher was unsubscribed while polling and we're waiting for the poll
-  /// to finish.
-  static final cancelling = new _WatchState(
-      listen: (_) => _WatchState.scanning,
-      finish: (_) => _WatchState.notWatching);
-
-  /// The watcher has subscribers, we have scanned for pre-existing files and
-  /// now we're polling for changes.
-  static final watching = new _WatchState(
-      cancel: (watcher) {
-    // No longer watching, so create a new incomplete ready future.
-    watcher._ready = new Completer();
-    return _WatchState.cancelling;
-  }, finish: (_) => _WatchState.watching,
-      shouldWatch: true, shouldNotify: true);
-
-  /// Called when the first subscriber to the watcher has been added.
-  final _WatchStateEvent listen;
-
-  /// Called when all subscriptions on the watcher have been cancelled.
-  final _WatchStateEvent cancel;
-
-  /// Called when a poll loop has finished.
-  final _WatchStateEvent finish;
-
-  /// If the directory watcher should be watching the file system while in
-  /// this state.
-  final bool shouldWatch;
-
-  /// If a change event should be sent for a file modification while in this
-  /// state.
-  final bool shouldNotify;
-
-  _WatchState({this.listen, this.cancel, this.finish,
-      this.shouldWatch: false, this.shouldNotify: false});
+  const _WatchState(this.name);
 }
 
 class _FileStatus {
   /// The last time the file was modified.
   DateTime modified;
 
   /// The SHA-1 hash of the contents of the file.
   List<int> hash;
 
   _FileStatus(this.modified, this.hash);
 }