Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(273)

Side by Side Diff: sdk/lib/async/future_impl.dart

Issue 15942010: Make _FutureImpl._chain do path shortening on listeners. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Now complete with error if there are cyclic dependencies. Created 7 years, 6 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch | Annotate | Revision Log
« no previous file with comments | « no previous file | tests/lib/async/future_test.dart » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file
2 // for details. All rights reserved. Use of this source code is governed by a 2 // for details. All rights reserved. Use of this source code is governed by a
3 // BSD-style license that can be found in the LICENSE file. 3 // BSD-style license that can be found in the LICENSE file.
4 4
5 part of dart.async; 5 part of dart.async;
6 6
7 deprecatedFutureValue(_FutureImpl future) => 7 deprecatedFutureValue(_FutureImpl future) =>
8 future._isComplete ? future._resultOrListeners : null; 8 future._isComplete ? future._resultOrListeners : null;
9 9
10 abstract class _Completer<T> implements Completer<T> { 10 abstract class _Completer<T> implements Completer<T> {
(...skipping 68 matching lines...) Expand 10 before | Expand all | Expand 10 after
79 /** Adapter for a [_FutureImpl] to be a future result listener. */ 79 /** Adapter for a [_FutureImpl] to be a future result listener. */
80 class _FutureListenerWrapper<T> implements _FutureListener<T> { 80 class _FutureListenerWrapper<T> implements _FutureListener<T> {
81 _FutureImpl future; 81 _FutureImpl future;
82 _FutureListener _nextListener; 82 _FutureListener _nextListener;
83 _FutureListenerWrapper(this.future); 83 _FutureListenerWrapper(this.future);
84 _sendValue(T value) { future._setValue(value); } 84 _sendValue(T value) { future._setValue(value); }
85 _sendError(error) { future._setError(error); } 85 _sendError(error) { future._setError(error); }
86 } 86 }
87 87
88 class _FutureImpl<T> implements Future<T> { 88 class _FutureImpl<T> implements Future<T> {
89 // State of the future. The state determines the interpretation of the
90 // [resultOrListeners] field.
91 // TODO(lrn): rename field since it can also contain a chained future.
92
93 /// Initial state, waiting for a result. In this state, the
94 /// [resultOrListeners] field holds a single-linked list of
95 /// [FutureListener] listeners.
89 static const int _INCOMPLETE = 0; 96 static const int _INCOMPLETE = 0;
90 static const int _VALUE = 1; 97 /// The future has been chained to another future. The result of that
91 static const int _ERROR = 2; 98 /// other future becomes the result of this future as well.
92 static const int _UNHANDLED_ERROR = 4; 99 /// In this state, the [resultOrListeners] field holds the future that
100 /// will give the result to this future. Both existing and new listeners are
101 /// forwarded directly to the other future.
102 static const int _CHAINED = 1;
103 /// The future has been chained to another future, but there hasn't been
104 /// any listeners added to this future yet. If it is completed with an
105 /// error, the error will be considered unhandled.
106 static const int _CHAINED_UNLISTENED = 3;
107 /// The future has been completed with a value result.
108 static const int _VALUE = 4;
109 /// The future has been completed with an error result.
110 static const int _ERROR = 6;
111 /// Extra bit set when the future has been completed with an error result.
112 /// but no listener has been scheduled to receive the error.
113 /// If the bit is still set when a [runAsync] call triggers, the error will
114 /// be reported to the top-level handler.
115 /// Assigning a listener before that time will clear the bit.
116 static const int _UNHANDLED_ERROR = 8;
93 117
94 /** Whether the future is complete, and as what. */ 118 /** Whether the future is complete, and as what. */
95 int _state = _INCOMPLETE; 119 int _state = _INCOMPLETE;
96 120
97 bool get _isComplete => _state != _INCOMPLETE; 121 bool get _isChained => (_state & _CHAINED) != 0;
122 bool get _hasChainedListener => _state == _CHAINED;
123 bool get _isComplete => _state >= _VALUE;
98 bool get _hasValue => _state == _VALUE; 124 bool get _hasValue => _state == _VALUE;
99 bool get _hasError => (_state & _ERROR) != 0; 125 bool get _hasError => _state >= _ERROR;
100 bool get _hasUnhandledError => (_state & _UNHANDLED_ERROR) != 0; 126 bool get _hasUnhandledError => _state >= _UNHANDLED_ERROR;
101 127
102 void _clearUnhandledError() { 128 void _clearUnhandledError() {
103 // Works because _UNHANDLED_ERROR is highest bit in use.
104 _state &= ~_UNHANDLED_ERROR; 129 _state &= ~_UNHANDLED_ERROR;
105 } 130 }
106 131
107 /** 132 /**
108 * Either the result, or a list of listeners until the future completes. 133 * Either the result, a list of listeners or another future.
109 * 134 *
110 * The result of the future is either a value or an error. 135 * The result of the future is either a value or an error.
111 * A result is only stored when the future has completed. 136 * A result is only stored when the future has completed.
112 * 137 *
113 * The listeners is an internally linked list of [_FutureListener]s. 138 * The listeners is an internally linked list of [_FutureListener]s.
114 * Listeners are only remembered while the future is not yet complete. 139 * Listeners are only remembered while the future is not yet complete,
140 * and it is not chained to another future.
115 * 141 *
116 * Since the result and the listeners cannot occur at the same time, 142 * The future is another future that his future is chained to. This future
117 * we can use the same field for both. 143 * is waiting for the other future to complete, and when it does, this future
144 * will complete with the same result.
145 * All listeners are forwarded to the other future.
146 *
147 * The cases are disjoint (incomplete and unchained, incomplete and
148 * chained, or completed with value or error), so the field only needs to hold
149 * one value at a time.
118 */ 150 */
119 var _resultOrListeners; 151 var _resultOrListeners;
120 152
121 _FutureImpl(); 153 _FutureImpl();
122 154
123 _FutureImpl.immediate(T value) { 155 _FutureImpl.immediate(T value) {
124 _state = _VALUE; 156 _state = _VALUE;
125 _resultOrListeners = value; 157 _resultOrListeners = value;
126 } 158 }
127 159
(...skipping 50 matching lines...) Expand 10 before | Expand all | Expand 10 after
178 } 210 }
179 211
180 Future<T> whenComplete(action()) { 212 Future<T> whenComplete(action()) {
181 return new _WhenFuture<T>(action).._subscribeTo(this); 213 return new _WhenFuture<T>(action).._subscribeTo(this);
182 } 214 }
183 215
184 Stream<T> asStream() => new Stream.fromFuture(this); 216 Stream<T> asStream() => new Stream.fromFuture(this);
185 217
186 void _setValue(T value) { 218 void _setValue(T value) {
187 if (_isComplete) throw new StateError("Future already completed"); 219 if (_isComplete) throw new StateError("Future already completed");
188 _FutureListener listeners = _removeListeners(); 220 _FutureListener listeners = _isChained ? null : _removeListeners();
189 _state = _VALUE; 221 _state = _VALUE;
190 _resultOrListeners = value; 222 _resultOrListeners = value;
191 while (listeners != null) { 223 while (listeners != null) {
192 _FutureListener listener = listeners; 224 _FutureListener listener = listeners;
193 listeners = listener._nextListener; 225 listeners = listener._nextListener;
194 listener._nextListener = null; 226 listener._nextListener = null;
195 listener._sendValue(value); 227 listener._sendValue(value);
196 } 228 }
197 } 229 }
198 230
199 void _setError(error) { 231 void _setError(error) {
200 if (_isComplete) throw new StateError("Future already completed"); 232 if (_isComplete) throw new StateError("Future already completed");
201 _FutureListener listeners = _removeListeners(); 233
234 _FutureListener listeners;
235 bool hasListeners;
236 if (_isChained) {
237 listeners = null;
238 hasListeners = (_state == _CHAINED); // and not _CHAINED_UNLISTENED.
239 } else {
240 listeners = _removeListeners();
241 hasListeners = (listeners != null);
242 }
243
202 _state = _ERROR; 244 _state = _ERROR;
203 _resultOrListeners = error; 245 _resultOrListeners = error;
204 if (listeners == null) { 246
247 if (!hasListeners) {
205 _scheduleUnhandledError(); 248 _scheduleUnhandledError();
206 return; 249 return;
207 } 250 }
208 do { 251 while (listeners != null) {
209 _FutureListener listener = listeners; 252 _FutureListener listener = listeners;
210 listeners = listener._nextListener; 253 listeners = listener._nextListener;
211 listener._nextListener = null; 254 listener._nextListener = null;
212 listener._sendError(error); 255 listener._sendError(error);
213 } while (listeners != null); 256 }
214 } 257 }
215 258
216 void _scheduleUnhandledError() { 259 void _scheduleUnhandledError() {
217 _state |= _UNHANDLED_ERROR; 260 assert(_state == _ERROR);
261 _state = _ERROR | _UNHANDLED_ERROR;
218 // Wait for the rest of the current event's duration to see 262 // Wait for the rest of the current event's duration to see
219 // if a subscriber is added to handle the error. 263 // if a subscriber is added to handle the error.
220 runAsync(() { 264 runAsync(() {
221 if (_hasUnhandledError) { 265 if (_hasUnhandledError) {
222 // No error handler has been added since the error was set. 266 // No error handler has been added since the error was set.
223 _clearUnhandledError(); 267 _clearUnhandledError();
268 // TODO(floitsch): Hook this into unhandled error handling.
224 var error = _resultOrListeners; 269 var error = _resultOrListeners;
225 print("Uncaught Error: ${error}"); 270 print("Uncaught Error: ${error}");
226 var trace = getAttachedStackTrace(error); 271 var trace = getAttachedStackTrace(error);
227 if (trace != null) { 272 if (trace != null) {
228 print("Stack Trace:\n$trace\n"); 273 print("Stack Trace:\n$trace\n");
229 } 274 }
230 throw error; 275 throw error;
231 } 276 }
232 }); 277 });
233 } 278 }
234 279
235 void _addListener(_FutureListener listener) { 280 void _addListener(_FutureListener listener) {
281 if (_isChained) {
282 _state = _CHAINED; // In case it was _CHAINED_UNLISTENED.
283 _FutureImpl resultSource = _chainSource;
284 resultSource._addListener(listener);
285 return;
286 }
236 if (_isComplete) { 287 if (_isComplete) {
237 _clearUnhandledError(); 288 _clearUnhandledError();
238 // Handle late listeners asynchronously. 289 // Handle late listeners asynchronously.
239 runAsync(() { 290 runAsync(() {
240 if (_hasValue) { 291 if (_hasValue) {
241 T value = _resultOrListeners; 292 T value = _resultOrListeners;
242 listener._sendValue(value); 293 listener._sendValue(value);
243 } else { 294 } else {
244 assert(_hasError); 295 assert(_hasError);
245 listener._sendError(_resultOrListeners); 296 listener._sendError(_resultOrListeners);
(...skipping 25 matching lines...) Expand all
271 322
272 /** 323 /**
273 * Make another [_FutureImpl] receive the result of this one. 324 * Make another [_FutureImpl] receive the result of this one.
274 * 325 *
275 * If this future is already complete, the [future] is notified 326 * If this future is already complete, the [future] is notified
276 * immediately. This function is only called during event resolution 327 * immediately. This function is only called during event resolution
277 * where it's acceptable to send an event. 328 * where it's acceptable to send an event.
278 */ 329 */
279 void _chain(_FutureImpl future) { 330 void _chain(_FutureImpl future) {
280 if (!_isComplete) { 331 if (!_isComplete) {
281 _addListener(future._asListener()); 332 future._chainFromFuture(this);
282 } else if (_hasValue) { 333 } else if (_hasValue) {
283 future._setValue(_resultOrListeners); 334 future._setValue(_resultOrListeners);
284 } else { 335 } else {
285 assert(_hasError); 336 assert(_hasError);
286 _clearUnhandledError(); 337 _clearUnhandledError();
287 future._setError(_resultOrListeners); 338 future._setError(_resultOrListeners);
288 } 339 }
289 } 340 }
290 341
291 /** 342 /**
343 * Returns the future that this future is chained to.
344 *
345 * If that future is itself chained to something else,
346 * get the [_chainSource] of that future instead, and make this
347 * future chain directly to the earliest source.
348 */
349 _FutureImpl get _chainSource {
350 assert(_isChained);
351 _FutureImpl future = _resultOrListeners;
352 if (future._isChained) {
353 future = _resultOrListeners = future._chainSource;
354 }
355 return future;
356 }
357
358 /**
359 * Make this incomplete future end up with the same result as [resultSource].
360 *
361 * This is done by moving all listeners to [resultSource] and forwarding all
362 * future [_addListener] calls to [resultSource] directly.
363 */
364 void _chainFromFuture(_FutureImpl resultSource) {
365 assert(!_isComplete);
366 assert(!_isChained);
367 if (resultSource._isChained) {
368 resultSource = resultSource._chainSource;
369 }
370 assert(!resultSource._isChained);
371 if (identical(this, resultSource)) {
372 // The only unchained future in a future dependency tree (as defined
373 // by the chain-relations) is the "root" that every other future depends
374 // on. The future we are adding is unchained, so if it is already in the
375 // tree, it must be the root, so that's the only one we need to check
376 // against to detect a cycle.
377 _setError(new StateError("Cyclic future dependency."));
378 return;
379 }
380 _FutureListener cursor = _removeListeners();
381 bool hadListeners = cursor != null;
382 while (cursor != null) {
383 _FutureListener listener = cursor;
384 cursor = cursor._nextListener;
385 listener._nextListener = null;
386 resultSource._addListener(listener);
387 }
388 // Listen with this future as well, so that when the other future completes,
389 // this future will be completed as well.
390 resultSource._addListener(this._asListener());
391 _resultOrListeners = resultSource;
392 _state = hadListeners ? _CHAINED : _CHAINED_UNLISTENED;
393 }
394
395 /**
292 * Helper function to handle the result of transforming an incoming event. 396 * Helper function to handle the result of transforming an incoming event.
293 * 397 *
294 * If the result is itself a [Future], this future is linked to that 398 * If the result is itself a [Future], this future is linked to that
295 * future's output. If not, this future is completed with the result. 399 * future's output. If not, this future is completed with the result.
296 */ 400 */
297 void _setOrChainValue(var result) { 401 void _setOrChainValue(var result) {
402 assert(!_isChained);
403 assert(!_isComplete);
298 if (result is Future) { 404 if (result is Future) {
299 // Result should be a Future<T>. 405 // Result should be a Future<T>.
300 if (result is _FutureImpl) { 406 if (result is _FutureImpl) {
301 _FutureImpl chainFuture = result; 407 _FutureImpl chainFuture = result;
302 chainFuture._chain(this); 408 chainFuture._chain(this);
303 return; 409 return;
304 } else { 410 } else {
305 Future future = result; 411 Future future = result;
306 future.then(_setValue, 412 future.then(_setValue,
307 onError: _setError); 413 onError: _setError);
(...skipping 156 matching lines...) Expand 10 before | Expand all | Expand 10 after
464 _setError(error); 570 _setError(error);
465 }, onError: _setError); 571 }, onError: _setError);
466 return; 572 return;
467 } 573 }
468 } catch (e, s) { 574 } catch (e, s) {
469 error = _asyncError(e, s); 575 error = _asyncError(e, s);
470 } 576 }
471 _setError(error); 577 _setError(error);
472 } 578 }
473 } 579 }
OLDNEW
« no previous file with comments | « no previous file | tests/lib/async/future_test.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698