Rewrite Futures.

sdk/lib/async/future_impl.dart

Index: sdk/lib/async/future_impl.dart
diff --git a/sdk/lib/async/future_impl.dart b/sdk/lib/async/future_impl.dart
index 4c846c67c6b5e7753ce1022d5dcc7a8c5eb1f206..302bd3206dea948b96bf70b5c4d0cf20aaecf45b 100644
--- a/sdk/lib/async/future_impl.dart
+++ b/sdk/lib/async/future_impl.dart
@@ -4,155 +4,49 @@
 
 part of dart.async;
 
-abstract class _Completer<T> implements Completer<T> {
-  final Future<T> future;
-  bool _isComplete = false;
-
-  _Completer() : future = new _FutureImpl<T>() {
-    _FutureImpl futureImpl = future;
-    futureImpl._zone.expectCallback();
-  }
+/** The onValue and onError handlers return either a value or a future */
+typedef dynamic _FutureOnValue<T>(T value);
+typedef dynamic _FutureOnError(error);
+/** Test used by [Future.catchError] to handle skip some errors. */
+typedef bool _FutureErrorTest(var error);
+/** Used by [WhenFuture]. */
+typedef _FutureAction();
 
-  void _setFutureValue(T value);
-  void _setFutureError(error);
+abstract class _Completer<T> implements Completer<T> {
+  final _Future<T> future = new _Future<T>();
 
-  void complete([T value]) {
-    if (_isComplete) throw new StateError("Future already completed");
-    _isComplete = true;
-    _FutureImpl futureImpl = future;
-    _setFutureValue(value);
-  }
+  void complete([T value]);
 
-  void completeError(Object error, [Object stackTrace = null]) {
-    if (_isComplete) throw new StateError("Future already completed");
-    _isComplete = true;
-    if (stackTrace != null) {
-      // Force the stack trace onto the error, even if it already had one.
-      _attachStackTrace(error, stackTrace);
-    }
-    _FutureImpl futureImpl = future;
-    _setFutureError(error);
-  }
+  void completeError(Object error, [Object stackTrace = null]);
 
-  bool get isCompleted => _isComplete;
+  // The future's _isComplete doesn't take into account pending completions.
+  // We therefore use _mayComplete.
+  bool get isCompleted => !future._mayComplete;
 }
 
 class _AsyncCompleter<T> extends _Completer<T> {
-  void _setFutureValue(T value) {
-    _FutureImpl future = this.future;
-    future._asyncSetValue(value);
-    // The async-error will schedule another callback, so we can cancel
-    // the expectation without shutting down the zone.
-    future._zone.cancelCallbackExpectation();
-  }
-
-  void _setFutureError(error) {
-    _FutureImpl future = this.future;
-    future._asyncSetError(error);
-    // The async-error will schedule another callback, so we can cancel
-    // the expectation without shutting down the zone.
-    future._zone.cancelCallbackExpectation();
-  }
-}
-
-class _SyncCompleter<T> extends _Completer<T> {
-  void _setFutureValue(T value) {
-    _FutureImpl future = this.future;
-    future._setValue(value);
-    future._zone.cancelCallbackExpectation();
-  }
 
-  void _setFutureError(error) {
-    _FutureImpl future = this.future;
-    future._setError(error);
-    future._zone.cancelCallbackExpectation();
+  void complete([T value]) {
+    future._asyncComplete(value);
   }
-}
 
-/**
- * A listener on a future.
- *
- * When the future completes, the [_sendValue] or [_sendError] method
- * is invoked with the result.
- *
- * Listeners are kept in a linked list.
- */
-abstract class _FutureListener<T> {
-  _FutureListener _nextListener;
-  factory _FutureListener.wrap(_FutureImpl future) {
-    return new _FutureListenerWrapper(future);
+  void completeError(Object error, [Object stackTrace = null]) {
+    future._asyncCompleteError(error, stackTrace);
   }
-  void _sendValue(T value);
-  void _sendError(error);
-
-  bool _inSameErrorZone(_Zone otherZone);
 }
 
-/** Adapter for a [_FutureImpl] to be a future result listener. */
-class _FutureListenerWrapper<T> implements _FutureListener<T> {
-  _FutureImpl future;
-  _FutureListener _nextListener;
-  _FutureListenerWrapper(this.future);
-  _sendValue(T value) { future._setValueUnchecked(value); }
-  _sendError(error) { future._setErrorUnchecked(error); }
-  bool _inSameErrorZone(_Zone otherZone) => future._inSameErrorZone(otherZone);
-}
-
-/**
- * This listener is installed at error-zone boundaries. It signals an
- * uncaught error in the zone of origin when an error is sent from one error
- * zone to another.
- *
- * When a Future is listening to another Future and they have not been
- * instantiated in the same error-zone then Futures put an instance of this
- * class between them (see [_FutureImpl._addListener]).
- *
- * For example:
- *
- *     var completer = new Completer();
- *     var future = completer.future.then((x) => x);
- *     catchErrors(() {
- *       var future2 = future.catchError(print);
- *     });
- *     completer.completeError(499);
- *
- * In this example `future` and `future2` are in different error-zones. The
- * error (499) that originates outside `catchErrors` must not reach the
- * `catchError` future (`future2`) inside `catchErrors`.
- *
- * When invoking `catchError` on `future` the Future installs an
- * [_ErrorZoneBoundaryListener] between itself and the result, `future2`.
- *
- * Conceptually _ErrorZoneBoundaryListeners could be implemented as
- * `catchError`s on the origin future as well.
- */
-class _ErrorZoneBoundaryListener implements _FutureListener {
-  _FutureListener _nextListener;
-  final _FutureListener _listener;
-
-  _ErrorZoneBoundaryListener(this._listener);
-
-  bool _inSameErrorZone(_Zone otherZone) {
-    // Should never be called. We use [_inSameErrorZone] to know if we have
-    // to insert an instance of [_ErrorZoneBoundaryListener] (and in the
-    // controller). Once we have inserted one we should never need to use it
-    // anymore.
-    throw new UnsupportedError(
-        "A Zone boundary doesn't support the inSameErrorZone test.");
-  }
+class _SyncCompleter<T> extends _Completer<T> {
 
-  void _sendValue(value) {
-    _listener._sendValue(value);
+  void complete([T value]) {
+    future._complete(value);
   }
 
-  void _sendError(error) {
-    // We are not allowed to send an error from one error-zone to another.
-    // This is the whole purpose of this class.
-    _Zone.current.handleUncaughtError(error);
+  void completeError(Object error, [Object stackTrace = null]) {
+    future._completeError(error, stackTrace);
   }
 }
 
-class _FutureImpl<T> implements Future<T> {
+class _Future<T> implements Future<T> {
   // State of the future. The state determines the interpretation of the
   // [resultOrListeners] field.
   // TODO(lrn): rename field since it can also contain a chained future.
@@ -161,35 +55,40 @@ class _FutureImpl<T> implements Future<T> {
   /// [resultOrListeners] field holds a single-linked list of
   /// [FutureListener] listeners.
   static const int _INCOMPLETE = 0;
-  /// Pending completion. Set when completed using [_asyncSetValue] or
-  /// [_asyncSetError]. It is an error to try to complete it again.
+  /// Pending completion. Set when completed using [_asyncComplete] or
+  /// [_asyncCompleteError]. It is an error to try to complete it again.
   static const int _PENDING_COMPLETE = 1;
   /// The future has been chained to another future. The result of that
   /// other future becomes the result of this future as well.
-  /// In this state, the [resultOrListeners] field holds the future that
-  /// will give the result to this future. Both existing and new listeners are
-  /// forwarded directly to the other future.
+  /// In this state, no callback should be executed anymore.
+  // TODO(floitsch): we don't really need a special "_CHAINED" state. We could
+  // just use the PENDING_COMPLETE state instead.
   static const int _CHAINED = 2;
-  /// The future has been chained to another future, but there hasn't been
-  /// any listeners added to this future yet. If it is completed with an
-  /// error, the error will be considered unhandled.
-  static const int _CHAINED_UNLISTENED = 6;
   /// The future has been completed with a value result.
-  static const int _VALUE = 8;
+  static const int _VALUE = 4;
   /// The future has been completed with an error result.
-  static const int _ERROR = 12;
+  static const int _ERROR = 8;
 
   /** Whether the future is complete, and as what. */
   int _state = _INCOMPLETE;
 
   final _Zone _zone = _Zone.current.fork();
 
-  bool get _isChained => (_state & _CHAINED) != 0;
-  bool get _hasChainedListener => _state == _CHAINED;
-  bool get _isComplete => _state >= _VALUE;
   bool get _mayComplete => _state == _INCOMPLETE;
+  bool get _isChained => _state == _CHAINED;
+  bool get _isComplete => _state >= _VALUE;
   bool get _hasValue => _state == _VALUE;
-  bool get _hasError => _state >= _ERROR;
+  bool get _hasError => _state == _ERROR;
+
+  set _isChained(bool value) {
+    if (value) {
+      assert(_mayComplete);
+      _state = _CHAINED;
+    } else {
+      assert(_isChained);
+      _state = _INCOMPLETE;
+    }
+  }
 
   /**
    * Either the result, a list of listeners or another future.
@@ -212,175 +111,130 @@ class _FutureImpl<T> implements Future<T> {
    */
   var _resultOrListeners;
 
-  _FutureImpl();
+  _Future _nextListener;

[Lasse Reichstein Nielsen, 2013/09/10 12:03:53]

Extra field, no documentation? What is this?

[Lasse Reichstein Nielsen, 2013/09/10 13:52:12]

If a single listener is the common case, how about having the list externally,
so _resultOrListeners contains either a single listener, or a (dart2js-like
Link<_Future> structure).
Might need a status bit to tell the difference.

[floitsch, 2013/09/10 17:19:14]

The same as before (line 81). Will add documentation.

[floitsch, 2013/09/10 17:19:14]

Completely agree. Adding a TODO.

 
-  _FutureImpl.immediate(T value) {
-    _state = _VALUE;
-    _resultOrListeners = value;
+  // TODO(floitsch): we only need two closure fields to store the callbacks.
+  // If we store the type of a closure in the state field (where there are
+  // still bits left), we can just store two closures instead of using 4
+  // fields of which 2 are always null.
+  final _FutureOnValue _onValueCallback;
+  final _FutureErrorTest _errorTestCallback;
+  final _FutureOnError _onErrorCallback;
+  final _FutureAction _whenCompleteActionCallback;

[Lasse Reichstein Nielsen, 2013/09/10 12:03:53]

Four more fields? Even two?
How about just one field, and a container for the case when there is more? Or
just have subclasses for the four cases instead?

[Lasse Reichstein Nielsen, 2013/09/10 13:52:12]

I am concerned about having the extra overhead on all futures.
I think we can do better, but this is ok as a first implementation.

[floitsch, 2013/09/10 17:19:14]

See TODO above.
Will optimize later.

[floitsch, 2013/09/10 17:19:14]

Agreed. We will have lots of time to optimize...

+
+  _FutureOnValue get _onValue => _isChained ? null : _onValueCallback;
+  _FutureErrorTest get _errorTest => _isChained ? null : _errorTestCallback;
+  _FutureOnError get _onError => _isChained ? null : _onErrorCallback;
+  _FutureAction get _whenCompleteAction
+      => _isChained ? null : _whenCompleteActionCallback;
+
+  _Future()
+      : _onValueCallback = null, _errorTestCallback = null,

[Lasse Reichstein Nielsen, 2013/09/10 12:03:53]

Default is null, you don't have to write all these.

[Lasse Reichstein Nielsen, 2013/09/10 13:52:12]

My bad.

[floitsch, 2013/09/10 17:19:14]

That is not correct. Final fields need to be initialized. Otherwise there is a
warning.

+        _onErrorCallback = null, _whenCompleteActionCallback = null;
+
+  _Future.immediate(T value)
+        : _onValueCallback = null, _errorTestCallback = null,
+          _onErrorCallback = null, _whenCompleteActionCallback = null {
+    _asyncComplete(value);
   }
 
-  _FutureImpl.immediateError(var error, [Object stackTrace]) {
-    if (stackTrace != null) {
-      // Force stack trace onto error, even if it had already one.
-      _attachStackTrace(error, stackTrace);
-    }
-    _asyncSetError(error);
+  _Future.immediateError(var error, [Object stackTrace])
+      : _onValueCallback = null, _errorTestCallback = null,
+        _onErrorCallback = null, _whenCompleteActionCallback = null {
+    _asyncCompleteError(error, stackTrace);
   }
 
-  factory _FutureImpl.wait(Iterable<Future> futures) {
-    Completer completer;
-    // List collecting values from the futures.
-    // Set to null if an error occurs.
-    List values;
-    void handleError(error) {
-      if (values != null) {
-        values = null;
-        completer.completeError(error);
-      }
-    }
-    // As each future completes, put its value into the corresponding
-    // position in the list of values.
-    int remaining = 0;
-    for (Future future in futures) {
-      int pos = remaining++;
-      future.catchError(handleError).then((Object value) {
-        if (values == null) return null;
-        values[pos] = value;
-        remaining--;
-        if (remaining == 0) {
-          completer.complete(values);
-        }
-      });
-    }
-    if (remaining == 0) {
-      return new Future.value(const []);
-    }
-    values = new List(remaining);
-    completer = new Completer<List>();
-    return completer.future;
+  _Future._then(this._onValueCallback, this._onErrorCallback)
+      : _errorTestCallback = null, _whenCompleteActionCallback = null {
+    _zone.expectCallback();
+  }
+
+  _Future._catchError(this._onErrorCallback, this._errorTestCallback)
+    : _onValueCallback = null, _whenCompleteActionCallback = null {
+    _zone.expectCallback();
+  }
+
+  _Future._whenComplete(this._whenCompleteActionCallback)
+      : _onValueCallback = null, _errorTestCallback = null,
+        _onErrorCallback = null {
+    _zone.expectCallback();
   }
 
   Future then(f(T value), { onError(error) }) {
-    if (onError == null) {
-      return new _ThenFuture(f).._subscribeTo(this);
-    }
-    return new _SubscribeFuture(f, onError).._subscribeTo(this);
+    _Future result;
+    result = new _Future._then(f, onError);
+    _addListener(result);
+    return result;
   }
 
   Future catchError(f(error), { bool test(error) }) {
-    return new _CatchErrorFuture(f, test).._subscribeTo(this);
+    _Future result = new _Future._catchError(f, test);
+    _addListener(result);
+    return result;
   }
 
   Future<T> whenComplete(action()) {
-    return new _WhenFuture<T>(action).._subscribeTo(this);
+    _Future result = new _Future<T>._whenComplete(action);
+    _addListener(result);
+    return result;
   }
 
   Stream<T> asStream() => new Stream.fromFuture(this);
 
-  bool _inSameErrorZone(_Zone otherZone) {
-    return _zone.inSameErrorZone(otherZone);
-  }
-
-  void _setValue(T value) {
+  void _markPendingCompletion() {
     if (!_mayComplete) throw new StateError("Future already completed");
-    _setValueUnchecked(value);
+    _state = _PENDING_COMPLETE;
   }
 
-  void _setValueUnchecked(T value) {
-    _FutureListener listeners = _isChained ? null : _removeListeners();
-    _state = _VALUE;
-    _resultOrListeners = value;
-    while (listeners != null) {
-      _FutureListener listener = listeners;
-      listeners = listener._nextListener;
-      listener._nextListener = null;
-      listener._sendValue(value);
-    }
+  void _clearPendingCompletion() {
+    assert(_state == _PENDING_COMPLETE);
+    _state = _INCOMPLETE;
   }
 
-  void _setError(Object error) {
-    if (!_mayComplete) throw new StateError("Future already completed");
-    _setErrorUnchecked(error);
+  T get _value {
+    assert(_isComplete && _hasValue);
+    return _resultOrListeners;
   }
 
-  void _setErrorUnchecked(Object error) {
-    _FutureListener listeners;
-    bool hasListeners;
-    if (_isChained) {
-      listeners = null;
-      hasListeners = (_state == _CHAINED);  // and not _CHAINED_UNLISTENED.
-    } else {
-      listeners = _removeListeners();
-      hasListeners = (listeners != null);
-    }
-
-    _state = _ERROR;
-    _resultOrListeners = error;
-
-    if (!hasListeners) {
-      // TODO(floitsch): Hook this into unhandled error handling.
-      var error = _resultOrListeners;
-      _zone.handleUncaughtError(error);
-      return;
-    }
-    while (listeners != null) {
-      _FutureListener listener = listeners;
-      listeners = listener._nextListener;
-      listener._nextListener = null;
-      listener._sendError(error);
-    }
+  Object get _error {
+    assert(_isComplete && _hasError);
+    return _resultOrListeners;
   }
 
-  void _asyncSetValue(T value) {
-    if (!_mayComplete) throw new StateError("Future already completed");
-    _state = _PENDING_COMPLETE;
-    runAsync(() { _setValueUnchecked(value); });
+  void _setValue(T value) {
+    assert(!_isComplete);  // But may have a completion pending.
+    _state = _VALUE;
+    _resultOrListeners = value;
   }
 
-  void _asyncSetError(Object error) {
-    if (!_mayComplete) throw new StateError("Future already completed");
-    _state = _PENDING_COMPLETE;
-    runAsync(() { _setErrorUnchecked(error); });
+  void _setError(Object error) {
+    assert(!_isComplete);  // But may have a completion pending.
+    _state = _ERROR;
+    _resultOrListeners = error;
   }
 
-  void _addListener(_FutureListener listener) {
+  void _addListener(_Future listener) {
     assert(listener._nextListener == null);
-    if (!listener._inSameErrorZone(_zone)) {
-      listener = new _ErrorZoneBoundaryListener(listener);
-    }
-    if (_isChained) {
-      _state = _CHAINED;  // In case it was _CHAINED_UNLISTENED.
-      _FutureImpl resultSource = _chainSource;
-      resultSource._addListener(listener);
-      return;
-    }
     if (_isComplete) {
       // Handle late listeners asynchronously.
       runAsync(() {
-        if (_hasValue) {
-          T value = _resultOrListeners;
-          listener._sendValue(value);
-        } else {
-          assert(_hasError);
-          listener._sendError(_resultOrListeners);
-        }
+        _propagateToSuccessors(this, listener);
       });
     } else {
-      assert(!_isComplete);
       listener._nextListener = _resultOrListeners;
       _resultOrListeners = listener;

[Lasse Reichstein Nielsen, 2013/09/10 12:03:53]

Ah, now I get it (where "now" is reading some code below).
The futures that are listeners of a future form a single linked list, and the
_nextListener is the link.

Far too clever abbreviation:
 _resultOrListeners = listener.._nextListener = _resultOrListeners;

[floitsch, 2013/09/10 17:19:14]

too clever :)

     }
   }
 
-  _FutureListener _removeListeners() {
+  _Future _removeListeners() {
     // Reverse listeners before returning them, so the resulting list is in
     // subscription order.
     assert(!_isComplete);
-    _FutureListener current = _resultOrListeners;
+    _Future current = _resultOrListeners;
     _resultOrListeners = null;
-    _FutureListener prev = null;
+    _Future prev = null;
     while (current != null) {
-      _FutureListener next = current._nextListener;
+      _Future next = current._nextListener;
       current._nextListener = prev;
       prev = current;
       current = next;
@@ -388,270 +242,263 @@ class _FutureImpl<T> implements Future<T> {
     return prev;
   }
 
-  /**
-   * Make another [_FutureImpl] receive the result of this one.
-   *
-   * If this future is already complete, the [future] is notified
-   * immediately. This function is only called during event resolution
-   * where it's acceptable to send an event.
-   */
-  void _chain(_FutureImpl future) {
-    if (!_isComplete) {
-      future._chainFromFuture(this);
-    } else if (_hasValue) {
-      future._setValue(_resultOrListeners);
-    } else {
-      assert(_hasError);
-      future._setError(_resultOrListeners);
-    }
-  }
-
-  /**
-   * Returns the future that this future is chained to.
-   *
-   * If that future is itself chained to something else,
-   * get the [_chainSource] of that future instead, and make this
-   * future chain directly to the earliest source.
-   */
-  _FutureImpl get _chainSource {
-    assert(_isChained);
-    _FutureImpl future = _resultOrListeners;
-    if (future._isChained) {
-      future = _resultOrListeners = future._chainSource;
-    }
-    return future;
-  }
+  static void _chainFutures(Future source, _Future target) {
+    assert(!target._isComplete);
 
-  /**
-   * Make this incomplete future end up with the same result as [resultSource].
-   *
-   * This is done by moving all listeners to [resultSource] and forwarding all
-   * future [_addListener] calls to [resultSource] directly.
-   */
-  void _chainFromFuture(_FutureImpl resultSource) {
-    assert(!_isComplete);
-    assert(!_isChained);
-    if (resultSource._isChained) {
-      resultSource = resultSource._chainSource;
-    }
-    assert(!resultSource._isChained);
-    if (identical(this, resultSource)) {
-      // The only unchained future in a future dependency tree (as defined
-      // by the chain-relations) is the "root" that every other future depends
-      // on. The future we are adding is unchained, so if it is already in the
-      // tree, it must be the root, so that's the only one we need to check
-      // against to detect a cycle.
-      _setError(new StateError("Cyclic future dependency."));
-      return;
-    }
-    _FutureListener cursor = _removeListeners();
-    bool hadListeners = cursor != null;
-    while (cursor != null) {
-      _FutureListener listener = cursor;
-      cursor = cursor._nextListener;
-      listener._nextListener = null;
-      resultSource._addListener(listener);
-    }
-    // Listen with this future as well, so that when the other future completes,
-    // this future will be completed as well.
-    resultSource._addListener(this._asListener());
-    _resultOrListeners = resultSource;
-    _state = hadListeners ? _CHAINED : _CHAINED_UNLISTENED;
-  }
-
-  /**
-   * Helper function to handle the result of transforming an incoming event.
-   *
-   * If the result is itself a [Future], this future is linked to that
-   * future's output. If not, this future is completed with the result.
-   */
-  void _setOrChainValue(var result) {
-    assert(!_isChained);
-    assert(!_isComplete);
-    if (result is Future) {
-      // Result should be a Future<T>.
-      if (result is _FutureImpl) {
-        _FutureImpl chainFuture = result;
-        chainFuture._chain(this);
-        return;
+    // Mark the target as chained (and as such half-completed).
+    target._isChained = true;
+    if (source is _Future) {
+      _Future internalFuture = source;
+      if (internalFuture._isComplete) {
+        _propagateToSuccessors(internalFuture, target);
       } else {
-        Future future = result;
-        future.then(_setValue,
-                    onError: _setError);
-        return;
+        internalFuture._addListener(target);
       }
     } else {
-      // Result must be of type T.
-      _setValue(result);
+      source.then((value) {
+          // Clear the is-chained bit, so that we can use the standard
+          // _complete method.
+          target._isChained = false;
+          target._complete(value);
+        },
+        onError: (error) {
+          // Clear the is-chained bit, so that we can use the standard
+          // _completeError method.
+          target._isChained = false;
+          target._completeError(error);
+        });
     }
   }
 
-  _FutureListener _asListener() => new _FutureListener.wrap(this);
-}
-
-/**
- * Transforming future base class.
- *
- * A transforming future is itself a future and a future listener.
- * Subclasses override [_sendValue]/[_sendError] to intercept
- * the results of a previous future.
- */
-abstract class _TransformFuture<S, T> extends _FutureImpl<T>
-                                      implements _FutureListener<S> {
-  // _FutureListener implementation.
-  _FutureListener _nextListener;
-
-  _TransformFuture() {
-    _zone.expectCallback();
-  }
-
-  void _sendValue(S value) {
-    _zone.executeCallback(() => _zonedSendValue(value));
+  void _complete(value) {
+    assert(_onValueCallback == null &&
+           _onErrorCallback == null &&
+           _whenCompleteActionCallback == null &&
+           _errorTestCallback == null);
+    if (!_mayComplete) throw new StateError("Future already completed");
+    if (value is Future) {
+      _chainFutures(value, this);
+      return;
+    }
+    _Future listeners = _removeListeners();
+    _setValue(value);
+    _propagateToSuccessors(this, listeners);
   }
 
-  void _sendError(error) {
-    _zone.executeCallback(() => _zonedSendError(error));
-  }
+  void _completeError(error, [StackTrace stackTrace]) {
+    assert(_onValueCallback == null &&
+        _onErrorCallback == null &&
+        _whenCompleteActionCallback == null &&
+        _errorTestCallback == null);
+    // _isComplete does not trigger for pending completions.
+    if (!_mayComplete) throw new StateError("Future already completed");
+    if (stackTrace != null) {
+      // Force the stack trace onto the error, even if it already had one.
+      _attachStackTrace(error, stackTrace);
+    }
 
-  void _subscribeTo(_FutureImpl future) {
-    future._addListener(this);
+    _Future listeners = _isChained ? null : _removeListeners();
+    _setError(error);
+    _propagateToSuccessors(this, listeners);

[Lasse Reichstein Nielsen, 2013/09/10 12:03:53]

Does it make sense to call this if listeners is null?

[floitsch, 2013/09/10 17:19:14]

Yes. Because that's where we need to trigger an uncaught exception.

   }
 
-  void _zonedSendValue(S value);
-  void _zonedSendError(error);
-}
-
-/** The onValue and onError handlers return either a value or a future */
-typedef dynamic _FutureOnValue<T>(T value);
-typedef dynamic _FutureOnError(error);
-/** Test used by [Future.catchError] to handle skip some errors. */
-typedef bool _FutureErrorTest(var error);
-/** Used by [WhenFuture]. */
-typedef _FutureAction();
-
-/** Future returned by [Future.then] with no [:onError:] parameter. */
-class _ThenFuture<S, T> extends _TransformFuture<S, T> {
-  // TODO(ahe): Restore type when feature is implemented in dart2js
-  // checked mode.
-  final /* _FutureOnValue<S> */ _onValue;
-
-  _ThenFuture(this._onValue);
-
-  _zonedSendValue(S value) {
-    assert(_onValue != null);
-    var result;
-    try {
-      result = _onValue(value);
-    } catch (e, s) {
-      _setError(_asyncError(e, s));
+  void _asyncComplete(value) {
+    assert(_onValueCallback == null &&
+        _onErrorCallback == null &&
+        _whenCompleteActionCallback == null &&
+        _errorTestCallback == null);
+    if (!_mayComplete) throw new StateError("Future already completed");
+    // Two corner cases if the value is a future:
+    //   1. the future is already completed and an error.
+    //   2. the future is not yet completed but might become an error.
+    // The first case means that we must not immediately complete the Future,
+    // as our code would immediately start propagating the error without
+    // giving the time to install error-handlers.
+    // However the second case requires us to deal with the value immediately.
+    // Otherwise the value could complete with an error and report an
+    // unhandled error, even though we know we are already going to listen to
+    // it.
+    if (value is Future &&
+        (value is! _Future || !(value as _Future)._isComplete)) {
+      // Case 2 from above. We need to register.
+      // Note that we are still completing asynchronously: either we register
+      // through .then (in which case the completing is asynchronous), or we
+      // have a _Future which isn't complete yet.
+      _complete(value);
       return;
     }
-    _setOrChainValue(result);
-  }
 
-  void _zonedSendError(error) {
-    _setError(error);
+    _markPendingCompletion();
+    runAsync(() {
+      _clearPendingCompletion();
+      _complete(value);
+    });
   }
-}
-
-/** Future returned by [Future.catchError]. */
-class _CatchErrorFuture<T> extends _TransformFuture<T,T> {
-  final _FutureErrorTest _test;
-  final _FutureOnError _onError;
-
-  _CatchErrorFuture(this._onError, this._test);
 
-  _zonedSendValue(T value) {
-    _setValue(value);
+  _asyncCompleteError(error, [StackTrace stackTrace]) {

[kevmoo-old, 2013/09/09 19:27:45]

void?

[floitsch, 2013/09/10 17:19:14]

Done.

+    assert(_onValueCallback == null &&
+        _onErrorCallback == null &&

[Lasse Reichstein Nielsen, 2013/09/10 12:03:53]

indent to after '('.
Or, preferably, make this four asserts. That way it's easier to see what is
wrong if one of them fails.

[floitsch, 2013/09/10 17:19:14]

Done.

+        _whenCompleteActionCallback == null &&
+        _errorTestCallback == null);
+    if (!_mayComplete) throw new StateError("Future already completed");
+    _markPendingCompletion();
+    runAsync(() {
+      _clearPendingCompletion();
+      _completeError(error, stackTrace);
+    });
   }
 
-  _zonedSendError(error) {
-    assert(_onError != null);
-    // if _test is supplied, check if it returns true, otherwise just
-    // forward the error unmodified.
-    if (_test != null) {
-      bool matchesTest;
-      try {
-        matchesTest = _test(error);
-      } catch (e, s) {
-        _setError(_asyncError(e, s));
+  /**
+   * Propagates the value/error of [source] to its [listeners], executing the
+   * listeners' callbacks.
+  *
+   * If [runCallback] is true (which should be the default) it executes
+   * the registered action of listeners. If it is `false` then the callback is
+   * skipped. This is used to complete futures with chained futures.
+   */
+  static void _propagateToSuccessors(_Future source, _Future listeners) {
+    while (true) {
+      if (!source._isComplete) return;  // Chained future.
+      bool hasError = source._hasError;
+      if (hasError && listeners == null) {
+        source._zone.handleUncaughtError(source._error);
         return;
       }
-      if (!matchesTest) {
-        _setError(error);
+      if (listeners == null) return;
+      _Future listener;
+      _Future remainingListeners = listeners;
+      do {
+        // We handle each listener individually, using the stack as a queue.
+        // Usually there is only one listener, so this should not be a problem.
+        listener = remainingListeners;  // Just an alias.
+        remainingListeners = remainingListeners._nextListener;
+        // Cut the connection to the remaining listeners. This way we can
+        // call _propagateToSuccessors without fearing that it may have an
+        // impact on other listeners.
+        listener._nextListener = null;
+        if (remainingListeners != null) {
+          _propagateToSuccessors(source, listener);

[Lasse Reichstein Nielsen, 2013/09/10 12:03:53]

This seems like an awkward way to do a loop.
Drop the self-recursion and have a helper function that does the actual work,
and a loop calling that for each listener.
Feel free to call the helper function directly, or through a different function
if you want the asserts outside the loop, if you know you have only a single
listener.

[floitsch, 2013/09/11 09:29:30]

created helper function.

+        }
+      } while (remainingListeners != null);
+
+      if (hasError && !source._zone.inSameErrorZone(listener._zone)) {
+        // Don't cross zone boundaries with errors.
+        source._zone.handleUncaughtError(source._error);
         return;
       }
-    }
-    // Act on the error, and use the result as this future's result.
-    var result;
-    try {
-      result = _onError(error);
-    } catch (e, s) {
-      _setError(_asyncError(e, s));
-      return;
-    }
-    _setOrChainValue(result);
-  }
-}
-
-/** Future returned by [Future.then] with an [:onError:] parameter. */
-class _SubscribeFuture<S, T> extends _ThenFuture<S, T> {
-  final _FutureOnError _onError;
-
-  _SubscribeFuture(onValue(S value), this._onError) : super(onValue);
-
-  // The _sendValue method is inherited from ThenFuture.
-
-  void _zonedSendError(error) {
-    assert(_onError != null);
-    var result;
-    try {
-      result = _onError(error);
-    } catch (e, s) {
-      _setError(_asyncError(e, s));
-      return;
-    }
-    _setOrChainValue(result);
-  }
-}
-
-/** Future returned by [Future.whenComplete]. */
-class _WhenFuture<T> extends _TransformFuture<T, T> {
-  final _FutureAction _action;
-
-  _WhenFuture(this._action);
-
-  void _zonedSendValue(T value) {
-    try {
-      var result = _action();
-      if (result is Future) {
-        Future resultFuture = result;
-        resultFuture.then((_) {
-          _setValue(value);
-        }, onError: _setError);
+      if (!identical(_Zone.current, listener._zone)) {
+        // Run the propagation in the listener's zone to avoid
+        // zone transitions. The idea is that many chained futures will
+        // be in the same zone.
+        listener._zone.executePeriodicCallback(() {
+          _propagateToSuccessors(source, listeners);

[Lasse Reichstein Nielsen, 2013/09/10 12:03:53]

listeners -> listener 
(same thing, easier to read).

[floitsch, 2013/09/10 17:19:14]

Done.

+        });
         return;
       }
-    } catch (e, s) {
-      _setError(_asyncError(e, s));
-      return;
-    }
-    _setValue(value);
-  }
-
-  void _zonedSendError(error) {
-    try {
-      var result = _action();
-      if (result is Future) {
-        Future resultFuture = result;
-        // TODO(lrn): Find a way to combine [error] into [e].
-        resultFuture.then((_) {
-          _setError(error);
-        }, onError: _setError);
+      // Normal listener.

[Lasse Reichstein Nielsen, 2013/09/10 12:03:53]

What does "normal" mean? 
Seems to be "in the same zone as the current"?

[floitsch, 2013/09/10 17:19:14]

Replaced with "// Do actual propagation".

+      // TODO(floitsch): Do we need to go through the zone even if we
+      // don't have a callback to execute?
+      bool listenerHasValue;
+      var listenerValueOrError;
+      // Set to true if a whenComplete needs to wait for a future.
+      // The whenComplete action will resume the propagation by itself.
+      bool isPropagationAborted = false;
+      // Even though we are already in the right zone (due to the optimization
+      // above), we still need to go through the zone. The overhead of
+      // executeCallback is however smaller when it is already in the correct
+      // zone.
+      // TODO(floitsch): only run callbacks in the zone, not the whole
+      // handling code.
+      listener._zone.executeCallback(() {
+        // TODO(floitsch): mark the listener as pending completion. Currently
+        // we can't do this, since the markPendingCompletion verifies that
+        // the future is not already marked (or chained).
+        try {
+          if (!hasError) {
+            var value = source._value;
+            if (listener._onValue != null) {
+              listenerValueOrError = listener._onValue(value);
+              listenerHasValue = true;
+            } else {
+              // Copy over the value from the source.
+              listenerValueOrError = value;
+              listenerHasValue = true;
+            }
+          } else {
+            Object error = source._error;
+            _FutureErrorTest test = listener._errorTest;
+            bool matchesTest = true;
+            if (test != null) {
+              matchesTest = test(error);
+            }
+            if (matchesTest && listener._onError != null) {
+              listenerValueOrError = listener._onError(error);
+              listenerHasValue = true;
+            } else {
+              // Copy over the error from the source.
+              listenerValueOrError = error;
+              listenerHasValue = false;
+            }
+          }
+
+          if (listener._whenCompleteAction != null) {
+            var completeResult = listener._whenCompleteAction();
+            if (completeResult is Future) {
+              listener._isChained = true;
+              completeResult.then((ignored) {
+                // Try again, but this time don't run the whenComplete callback.
+                _propagateToSuccessors(source, listener);
+              }, onError: (error) {
+                // When there is an error, we have to make the error the new
+                // result of the current listener.
+                if (completeResult is! _Future) {
+                  // This should be a rare case.
+                  completeResult = new _Future();

[Lasse Reichstein Nielsen, 2013/09/10 12:03:53]

Is there an .immediateError constructor?

[floitsch, 2013/09/10 17:19:14]

There is, but even though it is called "immediateError" it is completed
asynchronously.

+                  completeResult._setError(error);
+                }
+                _propagateToSuccessors(completeResult, listener);
+              });
+              isPropagationAborted = true;
+              // We will reenter the listener's zone.
+              listener._zone.expectCallback();
+            }
+          }
+        } catch (e, s) {
+          // Set the exception as error.
+          listenerValueOrError = _asyncError(e, s);
+          listenerHasValue = false;
+        }
+        if (listenerHasValue && listenerValueOrError is Future) {
+          // We are going to reenter the zone to finish what we started.
+          listener._zone.expectCallback();
+        }
+      });
+      if (isPropagationAborted) return;
+      // If the listener's value is a future we need to chain it.
+      if (listenerHasValue && listenerValueOrError is Future) {
+        Future chainSource = listenerValueOrError;
+        // Shortcut if the chain-source is already completed. Just continue the
+        // loop.
+        if (chainSource is _Future && (chainSource as _Future)._isComplete) {
+          // propagate the value (simulating a tail call).
+          listener._isChained = true;
+          source = chainSource;
+          listeners = listener;
+          continue;
+        }
+        _chainFutures(chainSource, listener);
         return;
       }
-    } catch (e, s) {
-      error = _asyncError(e, s);
+
+      if (listenerHasValue) {
+        listeners = listener._removeListeners();
+        listener._setValue(listenerValueOrError);
+      } else {
+        listeners = listener._removeListeners();
+        listener._setError(listenerValueOrError);
+      }
+      // Prepare for next round.
+      source = listener;
     }
-    _setError(error);
   }
 }