Move barback to a more event-based model.

pkg/barback/lib/src/phase.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.phase;
 
 import 'dart:async';
 
 import 'asset.dart';
 import 'asset_cascade.dart';
 import 'asset_id.dart';
 import 'asset_node.dart';
 import 'asset_set.dart';
 import 'errors.dart';
 import 'transform_node.dart';
 import 'transformer.dart';
+import 'utils.dart';
 
 /// One phase in the ordered series of transformations in an [AssetCascade].
 ///
 /// Each phase can access outputs from previous phases and can in turn pass
 /// outputs to later phases. Phases are processed strictly serially. All
 /// transforms in a phase will be complete before moving on to the next phase.
 /// Within a single phase, all transforms will be run in parallel.
 ///
 /// Building can be interrupted between phases. For example, a source is added
 /// which starts the background process. Sometime during, say, phase 2 (which
@@ skipping 11 matching lines @@
   ///
   /// Their outputs will be available to the next phase.
   final List<Transformer> _transformers;
 
   /// The inputs that are available for transforms in this phase to consume.
   ///
   /// For the first phase, these will be the source assets. For all other
   /// phases, they will be the outputs from the previous phase.
   final inputs = new Map<AssetId, AssetNode>();
 
-  /// The transforms currently applicable to assets in [inputs].
+  /// The transforms currently applicable to assets in [inputs], indexed by
+  /// the ids of their primary inputs.
   ///
   /// These are the transforms that have been "wired up": they represent a
   /// repeatable transformation of a single concrete set of inputs. "dart2js"
   /// is a transformer. "dart2js on web/main.dart" is a transform.
-  final _transforms = new Set<TransformNode>();
+  final _transforms = new Map<AssetId, Set<TransformNode>>();
 
-  /// The nodes that are new in this phase since the last time [process] was
-  /// called.
+  /// Futures that will complete once the transformers that can consume a given
+  /// asset are determined.
   ///
-  /// When we process, we'll check these to see if we can hang new transforms
-  /// off them.
-  final _newInputs = new Set<AssetNode>();
+  /// Whenever an asset is added or modified, we need to asynchronously
+  /// determine which transformers can use it as their primary input. We can't
+  /// start processing until we know which transformers to run, and this allows
+  /// us to wait until we do.
+  var _adjustTransformersFutures = new Map<AssetId, Future>();

[Bob Nystrom, 2013/07/31 20:05:41]

"adjust" is a bit confusing to me. How about "possible"?

[nweiz, 2013/07/31 22:47:53]

The name comes from the fact that these are futures running
[_adjustTransformers]. I feel like "possible" implies that these are futures
that may return transformers, which is incorrect; they don't return anything.
All that matters is their side effects and when they end.

+
+  /// New asset nodes that were added while [_adjustTransformers] was still
+  /// being run on an old version of that asset.
+  var _pendingNewInputs = new Map<AssetId, AssetNode>();
+
+  /// The ids of assets that are emitted by transforms in this phase.
+  ///
+  /// This is used to detect collisions where multiple transforms emit the same
+  /// output.
+  final _outputs = new Set<AssetId>();
 
   /// The phase after this one.
   ///
   /// Outputs from this phase will be passed to it.
   final Phase _next;
 
   Phase(this.cascade, this._index, this._transformers, this._next);
 
-  /// Updates the phase's inputs with [updated] and removes [removed].
-  ///
-  /// This marks any affected [transforms] as dirty or discards them if their
-  /// inputs are removed.
-  void updateInputs(AssetSet updated, Set<AssetId> removed) {
-    // Remove any nodes that are no longer being output. Handle removals first
-    // in case there are assets that were removed by one transform but updated
-    // by another. In that case, the update should win.
-    for (var id in removed) {
-      var node = inputs.remove(id);
-
-      // Every transform that was using it is dirty now.
-      if (node != null) {
-        node.consumers.forEach((consumer) => consumer.dirty());
-      }
+  /// Adds a new asset as an input for this phase.
+  ///
+  /// [node] doesn't have to be [AssetState.AVAILABLE]. Once it is, the phase
+  /// will automatically begin determining which transforms can consume it as a
+  /// primary input. The transforms themselves won't be applied until [process]
+  /// is called, however.
+  ///
+  /// This should only be used for brand-new assets or assets that have been
+  /// removed and re-created. The phase will automatically handle updated assets
+  /// using the [AssetNode.onStateChange] stream.
+  void addInput(AssetNode node) {
+    // We remove [node.id] from [inputs] as soon as the node is removed rather
+    // than at the same time [node.id] is removed from [_transforms] so we don't
+    // have to wait on [_adjustTransformers]. It's important that [inputs] is
+    // always up-to-date so that the [AssetCascade] can look there for available
+    // assets.
+    inputs[node.id] = node;
+    node.whenRemoved.then((_) => inputs.remove(node.id));
+
+    if (_adjustTransformersFutures.containsKey(node.id)) {

[Bob Nystrom, 2013/07/31 20:05:41]

Switch the order of cases here and do if (!_adjusted...) then return early from
there so you don't have to indent everything.

[nweiz, 2013/07/31 22:47:53]

Done.

+      // If an input is added while the same input is still being processed,
+      // that means that the asset was removed and recreated while
+      // [_adjustTransformers] was being run on the old value. We have to wait
+      // until that finishes, then run it again on whatever the newest version
+      // of that asset is.

[Bob Nystrom, 2013/07/31 20:05:41]

This comment is confused compared to the code. Is the idea that we are *already*
waiting (i.e. there is a future for this node ID in _adjustTransformersFutures)
so we don't have to wait again and can return here?

[nweiz, 2013/07/31 22:47:53]

The comment is notionally attached to the _pendingNewInputs line. I'll
reorganize to clarify.

+      var containedKey = _pendingNewInputs.containsKey(node.id);
+      _pendingNewInputs[node.id] = node;
+      if (containedKey) return;
+
+      _adjustTransformersFutures[node.id].then((_) {
+        assert(!_adjustTransformersFutures.containsKey(node.id));
+        assert(_pendingNewInputs.containsKey(node.id));
+        _transforms[node.id] = new Set<TransformNode>();
+        _adjustTransformers(_pendingNewInputs.remove(node.id));
+      }, onError: (_) {
+        // If there was a programmatic error while processing the old input,
+        // we don't want to just ignore it; it may have left the system in an
+        // inconsistent state. We also don't want to top-level it, so we
+        // ignore it here but don't start processing the new input. That way
+        // when [process] is called, the error will be piped through its
+        // return value.
+      }).catchError((e) {
+        // If our code above has a programmatic error, ensure it will be piped
+        // through [process] by putting it into [_adjustTransformersFutures].
+        _adjustTransformersFutures[node.id] = new Future.error(e);
+      });
+    } else {
+      _transforms[node.id] = new Set<TransformNode>();
+      _adjustTransformers(node);
     }
-
-    // Update and new or modified assets.
-    for (var asset in updated) {
-      var node = inputs[asset.id];
-      if (node == null) {
-        // It's a new node. Add it and remember it so we can see if any new
-        // transforms will consume it.
-        node = new AssetNode(asset);
-        inputs[asset.id] = node;
-        _newInputs.add(node);
-      } else {
-        node.updateAsset(asset);
-      }
-    }
+  }
+
+  /// Returns the input for this phase with the given [id], but only if that
+  /// input is known not to be consumed as a transformer's primary input.
+  ///
+  /// If the input is unavailable, or if the phase hasn't determined whether or
+  /// not any transformers will consume it as a primary input, null will be
+  /// returned instead. This means that the return value is guaranteed to always
+  /// be [AssetState.AVAILABLE].
+  AssetNode getUnconsumedInput(AssetId id) {
+    if (!inputs.containsKey(id)) return null;

[Bob Nystrom, 2013/07/31 20:05:41]

How about some blank lines above the comments in here?

[nweiz, 2013/07/31 22:47:53]

Done.

+    // If the asset has inputs but no _transforms set, that means that 

[Bob Nystrom, 2013/07/31 20:05:41]

means what?

[nweiz, 2013/07/31 22:47:53]

This can't actually happen any more; removed the clause.

+    if (!_transforms.containsKey(id)) return null;

[Bob Nystrom, 2013/07/31 20:05:41]

I don't understand this. If there's no transformer for the id, doesn't that mean
that it *is* potentially available?

[nweiz, 2013/07/31 22:47:53]

There's a difference between no transformer being available and the set of
transforms itself being null. But as I mentioned above, that can never happen
when the input exists, so I've removed this clause.

+    // If the asset has transforms, it's not unconsumed.
+    if (!_transforms[id].isEmpty) return null;
+    // If we're working on figuring out if the asset has transforms, we can't
+    // prove that it's unconsumed.
+    if (_adjustTransformersFutures.containsKey(id)) return null;
+    // The asset should be available. If it were removed, it wouldn't be in
+    // _inputs, and if it were dirty, it'd be in _adjustTransformersFutures.
+    assert(inputs[id].state.isAvailable);
+    return inputs[id];
+  }
+
+  /// Asynchronously determines which transformers can consume [node] as a
+  /// primary input and creates transforms for them.
+  ///
+  /// This ensures that if [node] is modified or removed during or after the
+  /// time it takes to adjust its transformers, they're appropriately
+  /// re-adjusted. Its progress can be tracked in [_adjustTransformersFutures].
+  void _adjustTransformers(AssetNode node) {

[Bob Nystrom, 2013/07/31 20:05:41]

This method name isn't very clear. How about _findNewTransforms or something
like that?

[nweiz, 2013/07/31 22:47:53]

It doesn't just find new transforms. It also checks whether existing transforms
still apply to modified inputs. The best way to think about it is that it makes
sure that all the transforms for a given input are wired up correctly, whatever
that takes. I think "_adjustTransforms" is as good a name as any for that.

+    // Once the input is available, hook up transformers for it. If it changes
+    // while that's happening, try again.
+    _adjustTransformersFutures[node.id] = node.tryUntilStable((asset) {
+      var oldTransformers = _transforms[node.id]
+          .map((transform) => transform.transformer).toSet();
+
+      return _removeStaleTransforms(asset)
+          .then((_) => _addNewTransforms(node, oldTransformers));
+    }).then((_) {
+      // Now all the transforms are set up correctly and the asset is available
+      // for the time being. Set up handlers for when the asset changes in the
+      // future.
+      node.onStateChange.first.then((state) {

[Bob Nystrom, 2013/07/31 20:05:41]

Do we need to handle the .first future having an error?

[nweiz, 2013/07/31 22:47:53]

Done. It now pipes the error so that it will be emitted via [process].

+        if (state.isRemoved) {
+          _transforms.remove(node.id);
+        } else {
+          _adjustTransformers(node);
+        }
+      });
+    }).catchError((error) {
+      if (error is! AssetNotFoundException || error.id != node.id) throw error;
+
+      // If the asset is removed, [tryUntilStable] will throw an
+      // [AssetNotFoundException]. In that case, just remove all transforms for
+      // the node.
+      _transforms.remove(node.id);
+    }).whenComplete(() {
+      _adjustTransformersFutures.remove(node.id);
+    });
+
+    // Don't top-level errors coming from the input processing. Any errors will
+    // eventually be piped through [process]'s returned Future.
+    _adjustTransformersFutures[node.id].catchError((_) {});
+  }
+
+  // Remove any old transforms that used to have [asset] as a primary asset but
+  // no longer apply to its new contents.
+  Future _removeStaleTransforms(Asset asset) {
+    return Future.wait(_transforms[asset.id].map((transform) {
+      // TODO(rnystrom): Catch all errors from isPrimary() and redirect to
+      // results.
+      return transform.transformer.isPrimary(asset).then((isPrimary) {
+        if (isPrimary) return;
+        _transforms[asset.id].remove(transform);
+        transform.remove();
+      });
+    }));
+  }
+
+  // Add new transforms for transformers that consider [asset] to be a primary

[Bob Nystrom, 2013/07/31 20:05:41]

"[asset]" -> "[node]'s asset"

[nweiz, 2013/07/31 22:47:53]

Done.

[nweiz, 2013/07/31 22:47:53]

Done.

+  // input.
+  //
+  // [oldTransformers] is the set of transformers that had [node] as a primary
+  // input prior to this. They don't need to be checked, since they were removed
+  // or preserved in [_removeStaleTransforms].
+  Future _addNewTransforms(AssetNode node, Set<Transformer> oldTransformers) {

[Bob Nystrom, 2013/07/31 20:05:41]

"new" -> "fresh" to correspond with "stale" above? Our terminology is a bit all
over the place. It would be nice to have a consistent set of metaphors here,
though it's tricky.

[nweiz, 2013/07/31 22:47:53]

Done.

+    return Future.wait(_transformers.map((transformer) {
+      if (oldTransformers.contains(transformer)) return new Future.value();
+
+      // If the asset is unavailable, the results of this [_adjustTransformers]
+      // run will be discarded, so we can just short-circuit.
+      if (node.asset == null) return new Future.value();
+
+      // We can safely access [node.asset] here even though it might have
+      // changed since (as above) if it has, [_adjustTransformers] will just be
+      // re-run.
+      // TODO(rnystrom): Catch all errors from isPrimary() and redirect to
+      // results.
+      return transformer.isPrimary(node.asset).then((isPrimary) {
+        if (!isPrimary) return;
+        _transforms[node.id].add(new TransformNode(this, transformer, node));
+      });
+    }));
   }
 
   /// Processes this phase.
   ///
-  /// For all new inputs, it tries to see if there are transformers that can
-  /// consume them. Then all applicable transforms are applied.
-  ///
   /// Returns a future that completes when processing is done. If there is
   /// nothing to process, returns `null`.
   Future process() {
-    var future = _processNewInputs();
-    if (future == null) {
-      return _processTransforms();
-    }
-
-    return future.then((_) => _processTransforms());
-  }
-
-  /// Creates new transforms for any new inputs that are applicable.
-  Future _processNewInputs() {
-    if (_newInputs.isEmpty) return null;
-
-    var futures = [];
-    for (var node in _newInputs) {
-      for (var transformer in _transformers) {
-        // TODO(rnystrom): Catch all errors from isPrimary() and redirect
-        // to results.
-        futures.add(transformer.isPrimary(node.asset).then((isPrimary) {
-          if (!isPrimary) return;
-          var transform = new TransformNode(this, transformer, node);
-          node.consumers.add(transform);
-          _transforms.add(transform);
-        }));
-      }
-    }
-
-    _newInputs.clear();
-
-    return Future.wait(futures);
+    if (_adjustTransformersFutures.isEmpty) return _processTransforms();
+    return _waitForInputs().then((_) => _processTransforms());
+  }
+
+  Future _waitForInputs() {
+    if (_adjustTransformersFutures.isEmpty) return new Future.value();
+    return Future.wait(_adjustTransformersFutures.values)
+        .then((_) => _waitForInputs());
   }
 
   /// Applies all currently wired up and dirty transforms.
-  ///
-  /// Passes their outputs to the next phase.
   Future _processTransforms() {
     // Convert this to a list so we can safely modify _transforms while
     // iterating over it.
-    var dirtyTransforms = _transforms.where((transform) => transform.isDirty)
-        .toList();
+    var dirtyTransforms =
+        flatten(_transforms.values.map((transforms) => transforms.toList()))
+        .where((transform) => transform.isDirty).toList();
     if (dirtyTransforms.isEmpty) return null;
 
-    return Future.wait(dirtyTransforms.map((transform) {
-      if (inputs.containsKey(transform.primary.id)) return transform.apply();
-
-      // If the primary input for the transform has been removed, get rid of it
-      // and all its outputs.
-      _transforms.remove(transform);
-      return new Future.value(
-          new TransformOutputs(new AssetSet(), transform.outputs));
-    })).then((transformOutputs) {
-      // Collect all of the outputs. Since the transforms are run in parallel,
-      // we have to be careful here to ensure that the result is deterministic
-      // and not influenced by the order that transforms complete.
-      var updated = new AssetSet();
-      var removed = new Set<AssetId>();
+    return Future.wait(dirtyTransforms.map((transform) => transform.apply()))
+        .then((allNewOutputs) {
+      var newOutputs = allNewOutputs.reduce((set1, set2) => set1.union(set2));
+
       var collisions = new Set<AssetId>();
-
-      // Handle the generated outputs of all transforms first.
-      for (var outputs in transformOutputs) {
-        // Collect the outputs of all transformers together.
-        for (var asset in outputs.updated) {
-          if (updated.containsId(asset.id)) {
-            // Report a collision.
-            collisions.add(asset.id);
-          } else {
-            // TODO(rnystrom): In the case of a collision, the asset that
-            // "wins" is chosen non-deterministically. Do something better.
-            updated.add(asset);
-          }
+      for (var newOutput in newOutputs) {
+        if (_outputs.contains(newOutput.id)) {
+          collisions.add(newOutput.id);
+        } else {
+          _next.addInput(newOutput);
+          _outputs.add(newOutput.id);
+          newOutput.whenRemoved.then((_) => _outputs.remove(newOutput.id));
         }
-
-        // Track any assets no longer output by this transform. We don't
-        // handle the case where *another* transform generates the asset
-        // no longer generated by this one. updateInputs() handles that.
-        removed.addAll(outputs.removed);
       }
 
-      // Report any collisions in deterministic order.
+      // Report collisions in a deterministic order.
       collisions = collisions.toList();
       collisions.sort((a, b) => a.toString().compareTo(b.toString()));
       for (var collision in collisions) {
         cascade.reportError(new AssetCollisionException(collision));
         // TODO(rnystrom): Define what happens after a collision occurs.
       }
-
-      // Pass the outputs to the next phase.
-      _next.updateInputs(updated, removed);
     });
   }
 }