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

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

Issue 15764003: Add Zone support for Timers. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Fix bug and add tests. 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
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> {
11 final Future<T> future; 11 final Future<T> future;
12 bool _isComplete = false; 12 bool _isComplete = false;
13 13
14 _Completer() : future = new _FutureImpl<T>(); 14 _Completer() : future = new _FutureImpl<T>() {
15 _FutureImpl futureImpl = future;
16 futureImpl._zone.expectCallback();
17 }
15 18
16 void _setFutureValue(T value); 19 void _setFutureValue(T value);
17 void _setFutureError(error); 20 void _setFutureError(error);
18 21
19 void complete([T value]) { 22 void complete([T value]) {
20 if (_isComplete) throw new StateError("Future already completed"); 23 if (_isComplete) throw new StateError("Future already completed");
21 _isComplete = true; 24 _isComplete = true;
25 _FutureImpl futureImpl = future;
26 futureImpl._zone.unexpectCallback();
22 _setFutureValue(value); 27 _setFutureValue(value);
23 } 28 }
24 29
25 void completeError(Object error, [Object stackTrace = null]) { 30 void completeError(Object error, [Object stackTrace = null]) {
26 if (_isComplete) throw new StateError("Future already completed"); 31 if (_isComplete) throw new StateError("Future already completed");
27 _isComplete = true; 32 _isComplete = true;
28 if (stackTrace != null) { 33 if (stackTrace != null) {
29 // Force the stack trace onto the error, even if it already had one. 34 // Force the stack trace onto the error, even if it already had one.
30 _attachStackTrace(error, stackTrace); 35 _attachStackTrace(error, stackTrace);
31 } 36 }
32 _setFutureError(error); 37 _FutureImpl futureImpl = future;
38 if (futureImpl._inSameErrorZone(_Zone.current)) {
39 futureImpl._zone.unexpectCallback();
40 _setFutureError(error);
41 } else {
42 _Zone.current.handleUncaughtError(error);
43 }
33 } 44 }
34 45
35 bool get isCompleted => _isComplete; 46 bool get isCompleted => _isComplete;
36 } 47 }
37 48
38 class _AsyncCompleter<T> extends _Completer<T> { 49 class _AsyncCompleter<T> extends _Completer<T> {
39 void _setFutureValue(T value) { 50 void _setFutureValue(T value) {
40 _FutureImpl future = this.future; 51 _FutureImpl future = this.future;
41 runAsync(() { future._setValue(value); }); 52 runAsync(() { future._setValue(value); });
42 } 53 }
(...skipping 24 matching lines...) Expand all
67 * 78 *
68 * Listeners are kept in a linked list. 79 * Listeners are kept in a linked list.
69 */ 80 */
70 abstract class _FutureListener<T> { 81 abstract class _FutureListener<T> {
71 _FutureListener _nextListener; 82 _FutureListener _nextListener;
72 factory _FutureListener.wrap(_FutureImpl future) { 83 factory _FutureListener.wrap(_FutureImpl future) {
73 return new _FutureListenerWrapper(future); 84 return new _FutureListenerWrapper(future);
74 } 85 }
75 void _sendValue(T value); 86 void _sendValue(T value);
76 void _sendError(error); 87 void _sendError(error);
88
89 bool _inSameErrorZone(_Zone otherZone);
77 } 90 }
78 91
79 /** Adapter for a [_FutureImpl] to be a future result listener. */ 92 /** Adapter for a [_FutureImpl] to be a future result listener. */
80 class _FutureListenerWrapper<T> implements _FutureListener<T> { 93 class _FutureListenerWrapper<T> implements _FutureListener<T> {
81 _FutureImpl future; 94 _FutureImpl future;
82 _FutureListener _nextListener; 95 _FutureListener _nextListener;
83 _FutureListenerWrapper(this.future); 96 _FutureListenerWrapper(this.future);
84 _sendValue(T value) { future._setValue(value); } 97 _sendValue(T value) { future._setValue(value); }
85 _sendError(error) { future._setError(error); } 98 _sendError(error) { future._setError(error); }
99 bool _inSameErrorZone(_Zone otherZone) => future._inSameErrorZone(otherZone);
100 }
101
102 /**
103 * This listener is installed at error-zone boundaries. It signals an
104 * uncaught error in the zone of origin when an error is sent from one error
105 * zone to another.
106 *
107 * When a Future is listening to another Future and they have not been
108 * instantiated in the same error-zone then Futures put an instance of this
109 * class between them (see [_FutureImpl._addListener]).
110 *
111 * For example:
112 *
113 * var completer = new Completer();
114 * var future = completer.future.then((x) => x);
115 * catchErrors(() {
116 * var future2 = future.catchError(print);
117 * });
118 * completer.completeError(499);
119 *
120 * In this example `future` and `future2` are in different error-zones. The
121 * error (499) that originates outside `catchErrors` must not reach the
122 * `catchError` future (`future2`) inside `catchErrors`.
123 *
124 * When invoking `catchError` on `future` the Future installs an
125 * [_ErrorZoneBoundaryListener] between itself and the result, `future2`.
126 *
127 * Conceptually _ErrorZoneBoundaryListeners could be implemented as
128 * `catchError`s on the origin future as well.
129 */
130 class _ErrorZoneBoundaryListener implements _FutureListener {
131 _FutureListener _nextListener;
132 final _FutureListener _listener;
133
134 _ErrorZoneBoundaryListener(this._listener);
135
136 bool _inSameErrorZone(_Zone otherZone) {
137 // Should never be called. We use [_inSameErrorZone] to know if we have
138 // to insert an instance of [_ErrorZoneBoundaryListener] (and in the
139 // controller). Once we have inserted one we should never need to use it
140 // anymore.
141 // It would be valid to `return true` instead.
142 throw new UnsupportedError(
Lasse Reichstein Nielsen 2013/05/29 08:04:59 Then just return true. You never know if someone l
floitsch 2013/05/29 15:21:29 Actually it is not necessarily `true`. It depends
143 "A Zone boundary doesn't support the inSameErrorZone test.");
144 }
145
146 void _sendValue(value) {
147 _listener._sendValue(value);
148 }
149
150 void _sendError(error) {
151 // We are not allowed to send an error from one error-zone to another.
152 // This is the whole purpose of this class.
153 _Zone.current.handleUncaughtError(error);
154 }
86 } 155 }
87 156
88 class _FutureImpl<T> implements Future<T> { 157 class _FutureImpl<T> implements Future<T> {
89 static const int _INCOMPLETE = 0; 158 static const int _INCOMPLETE = 0;
90 static const int _VALUE = 1; 159 static const int _VALUE = 1;
91 static const int _ERROR = 2; 160 static const int _ERROR = 2;
92 static const int _UNHANDLED_ERROR = 4; 161 static const int _UNHANDLED_ERROR = 4;
93 162
94 /** Whether the future is complete, and as what. */ 163 /** Whether the future is complete, and as what. */
95 int _state = _INCOMPLETE; 164 int _state = _INCOMPLETE;
96 165
166 final _Zone _zone = _Zone.current.fork();
167
97 bool get _isComplete => _state != _INCOMPLETE; 168 bool get _isComplete => _state != _INCOMPLETE;
98 bool get _hasValue => _state == _VALUE; 169 bool get _hasValue => _state == _VALUE;
99 bool get _hasError => (_state & _ERROR) != 0; 170 bool get _hasError => (_state & _ERROR) != 0;
100 bool get _hasUnhandledError => (_state & _UNHANDLED_ERROR) != 0; 171 bool get _hasUnhandledError => (_state & _UNHANDLED_ERROR) != 0;
101 172
102 void _clearUnhandledError() { 173 void _clearUnhandledError() {
103 // Works because _UNHANDLED_ERROR is highest bit in use. 174 // Works because _UNHANDLED_ERROR is highest bit in use.
104 _state &= ~_UNHANDLED_ERROR; 175 _state &= ~_UNHANDLED_ERROR;
105 } 176 }
106 177
(...skipping 69 matching lines...) Expand 10 before | Expand all | Expand 10 after
176 Future catchError(f(error), { bool test(error) }) { 247 Future catchError(f(error), { bool test(error) }) {
177 return new _CatchErrorFuture(f, test).._subscribeTo(this); 248 return new _CatchErrorFuture(f, test).._subscribeTo(this);
178 } 249 }
179 250
180 Future<T> whenComplete(action()) { 251 Future<T> whenComplete(action()) {
181 return new _WhenFuture<T>(action).._subscribeTo(this); 252 return new _WhenFuture<T>(action).._subscribeTo(this);
182 } 253 }
183 254
184 Stream<T> asStream() => new Stream.fromFuture(this); 255 Stream<T> asStream() => new Stream.fromFuture(this);
185 256
257 bool _inSameErrorZone(_Zone otherZone) {
258 return _zone.inSameErrorZone(otherZone);
259 }
260
186 void _setValue(T value) { 261 void _setValue(T value) {
187 if (_isComplete) throw new StateError("Future already completed"); 262 if (_isComplete) throw new StateError("Future already completed");
188 _FutureListener listeners = _removeListeners(); 263 _FutureListener listeners = _removeListeners();
189 _state = _VALUE; 264 _state = _VALUE;
190 _resultOrListeners = value; 265 _resultOrListeners = value;
191 while (listeners != null) { 266 while (listeners != null) {
192 _FutureListener listener = listeners; 267 _FutureListener listener = listeners;
193 listeners = listener._nextListener; 268 listeners = listener._nextListener;
194 listener._nextListener = null; 269 listener._nextListener = null;
195 listener._sendValue(value); 270 listener._sendValue(value);
(...skipping 19 matching lines...) Expand all
215 290
216 void _scheduleUnhandledError() { 291 void _scheduleUnhandledError() {
217 _state |= _UNHANDLED_ERROR; 292 _state |= _UNHANDLED_ERROR;
218 // Wait for the rest of the current event's duration to see 293 // Wait for the rest of the current event's duration to see
219 // if a subscriber is added to handle the error. 294 // if a subscriber is added to handle the error.
220 runAsync(() { 295 runAsync(() {
221 if (_hasUnhandledError) { 296 if (_hasUnhandledError) {
222 // No error handler has been added since the error was set. 297 // No error handler has been added since the error was set.
223 _clearUnhandledError(); 298 _clearUnhandledError();
224 var error = _resultOrListeners; 299 var error = _resultOrListeners;
225 print("Uncaught Error: ${error}"); 300 _zone.handleUncaughtError(error);
226 var trace = getAttachedStackTrace(error);
227 if (trace != null) {
228 print("Stack Trace:\n$trace\n");
229 }
230 throw error;
231 } 301 }
232 }); 302 });
233 } 303 }
234 304
235 void _addListener(_FutureListener listener) { 305 void _addListener(_FutureListener listener) {
306 assert(listener._nextListener == null);
307 if (!listener._inSameErrorZone(_zone)) {
308 listener = new _ErrorZoneBoundaryListener(listener);
309 }
236 if (_isComplete) { 310 if (_isComplete) {
237 _clearUnhandledError(); 311 _clearUnhandledError();
238 // Handle late listeners asynchronously. 312 // Handle late listeners asynchronously.
239 runAsync(() { 313 runAsync(() {
240 if (_hasValue) { 314 if (_hasValue) {
241 T value = _resultOrListeners; 315 T value = _resultOrListeners;
242 listener._sendValue(value); 316 listener._sendValue(value);
243 } else { 317 } else {
244 assert(_hasError); 318 assert(_hasError);
245 listener._sendError(_resultOrListeners); 319 listener._sendError(_resultOrListeners);
246 } 320 }
247 }); 321 });
248 } else { 322 } else {
249 assert(!_isComplete); 323 assert(!_isComplete);
250 assert(listener._nextListener == null);
251 listener._nextListener = _resultOrListeners; 324 listener._nextListener = _resultOrListeners;
252 _resultOrListeners = listener; 325 _resultOrListeners = listener;
253 } 326 }
254 } 327 }
255 328
256 _FutureListener _removeListeners() { 329 _FutureListener _removeListeners() {
257 // Reverse listeners before returning them, so the resulting list is in 330 // Reverse listeners before returning them, so the resulting list is in
258 // subscription order. 331 // subscription order.
259 assert(!_isComplete); 332 assert(!_isComplete);
260 _FutureListener current = _resultOrListeners; 333 _FutureListener current = _resultOrListeners;
(...skipping 60 matching lines...) Expand 10 before | Expand all | Expand 10 after
321 * 394 *
322 * A transforming future is itself a future and a future listener. 395 * A transforming future is itself a future and a future listener.
323 * Subclasses override [_sendValue]/[_sendError] to intercept 396 * Subclasses override [_sendValue]/[_sendError] to intercept
324 * the results of a previous future. 397 * the results of a previous future.
325 */ 398 */
326 abstract class _TransformFuture<S, T> extends _FutureImpl<T> 399 abstract class _TransformFuture<S, T> extends _FutureImpl<T>
327 implements _FutureListener<S> { 400 implements _FutureListener<S> {
328 // _FutureListener implementation. 401 // _FutureListener implementation.
329 _FutureListener _nextListener; 402 _FutureListener _nextListener;
330 403
331 void _sendValue(S value); 404 _TransformFuture() {
405 _zone.expectCallback();
406 }
332 407
333 void _sendError(error); 408 void _sendValue(S value) {
409 _zone.executeCallback(() => _zonedSendValue(value));
410 }
411
412 void _sendError(error) {
413 _zone.executeCallback(() => _zonedSendError(error));
414 }
334 415
335 void _subscribeTo(_FutureImpl future) { 416 void _subscribeTo(_FutureImpl future) {
336 future._addListener(this); 417 future._addListener(this);
337 } 418 }
419
420 void _zonedSendValue(S value);
421 void _zonedSendError(error);
338 } 422 }
339 423
340 /** The onValue and onError handlers return either a value or a future */ 424 /** The onValue and onError handlers return either a value or a future */
341 typedef dynamic _FutureOnValue<T>(T value); 425 typedef dynamic _FutureOnValue<T>(T value);
342 typedef dynamic _FutureOnError(error); 426 typedef dynamic _FutureOnError(error);
343 /** Test used by [Future.catchError] to handle skip some errors. */ 427 /** Test used by [Future.catchError] to handle skip some errors. */
344 typedef bool _FutureErrorTest(var error); 428 typedef bool _FutureErrorTest(var error);
345 /** Used by [WhenFuture]. */ 429 /** Used by [WhenFuture]. */
346 typedef _FutureAction(); 430 typedef _FutureAction();
347 431
348 /** Future returned by [Future.then] with no [:onError:] parameter. */ 432 /** Future returned by [Future.then] with no [:onError:] parameter. */
349 class _ThenFuture<S, T> extends _TransformFuture<S, T> { 433 class _ThenFuture<S, T> extends _TransformFuture<S, T> {
350 // TODO(ahe): Restore type when feature is implemented in dart2js 434 // TODO(ahe): Restore type when feature is implemented in dart2js
351 // checked mode. 435 // checked mode.
352 final /* _FutureOnValue<S> */ _onValue; 436 final /* _FutureOnValue<S> */ _onValue;
353 437
354 _ThenFuture(this._onValue); 438 _ThenFuture(this._onValue);
355 439
356 _sendValue(S value) { 440 _zonedSendValue(S value) {
357 assert(_onValue != null); 441 assert(_onValue != null);
358 var result; 442 var result;
359 try { 443 try {
360 result = _onValue(value); 444 result = _onValue(value);
361 } catch (e, s) { 445 } catch (e, s) {
362 _setError(_asyncError(e, s)); 446 _setError(_asyncError(e, s));
363 return; 447 return;
364 } 448 }
365 _setOrChainValue(result); 449 _setOrChainValue(result);
366 } 450 }
367 451
368 void _sendError(error) { 452 void _zonedSendError(error) {
369 _setError(error); 453 _setError(error);
370 } 454 }
371 } 455 }
372 456
373 /** Future returned by [Future.catchError]. */ 457 /** Future returned by [Future.catchError]. */
374 class _CatchErrorFuture<T> extends _TransformFuture<T,T> { 458 class _CatchErrorFuture<T> extends _TransformFuture<T,T> {
375 final _FutureErrorTest _test; 459 final _FutureErrorTest _test;
376 final _FutureOnError _onError; 460 final _FutureOnError _onError;
377 461
378 _CatchErrorFuture(this._onError, this._test); 462 _CatchErrorFuture(this._onError, this._test);
379 463
380 _sendValue(T value) { 464 _zonedSendValue(T value) {
381 _setValue(value); 465 _setValue(value);
382 } 466 }
383 467
384 _sendError(error) { 468 _zonedSendError(error) {
385 assert(_onError != null); 469 assert(_onError != null);
386 // if _test is supplied, check if it returns true, otherwise just 470 // if _test is supplied, check if it returns true, otherwise just
387 // forward the error unmodified. 471 // forward the error unmodified.
388 if (_test != null) { 472 if (_test != null) {
389 bool matchesTest; 473 bool matchesTest;
390 try { 474 try {
391 matchesTest = _test(error); 475 matchesTest = _test(error);
392 } catch (e, s) { 476 } catch (e, s) {
393 _setError(_asyncError(e, s)); 477 _setError(_asyncError(e, s));
394 return; 478 return;
(...skipping 16 matching lines...) Expand all
411 } 495 }
412 496
413 /** Future returned by [Future.then] with an [:onError:] parameter. */ 497 /** Future returned by [Future.then] with an [:onError:] parameter. */
414 class _SubscribeFuture<S, T> extends _ThenFuture<S, T> { 498 class _SubscribeFuture<S, T> extends _ThenFuture<S, T> {
415 final _FutureOnError _onError; 499 final _FutureOnError _onError;
416 500
417 _SubscribeFuture(onValue(S value), this._onError) : super(onValue); 501 _SubscribeFuture(onValue(S value), this._onError) : super(onValue);
418 502
419 // The _sendValue method is inherited from ThenFuture. 503 // The _sendValue method is inherited from ThenFuture.
420 504
421 void _sendError(error) { 505 void _zonedSendError(error) {
422 assert(_onError != null); 506 assert(_onError != null);
423 var result; 507 var result;
424 try { 508 try {
425 result = _onError(error); 509 result = _onError(error);
426 } catch (e, s) { 510 } catch (e, s) {
427 _setError(_asyncError(e, s)); 511 _setError(_asyncError(e, s));
428 return; 512 return;
429 } 513 }
430 _setOrChainValue(result); 514 _setOrChainValue(result);
431 } 515 }
432 } 516 }
433 517
434 /** Future returned by [Future.whenComplete]. */ 518 /** Future returned by [Future.whenComplete]. */
435 class _WhenFuture<T> extends _TransformFuture<T, T> { 519 class _WhenFuture<T> extends _TransformFuture<T, T> {
436 final _FutureAction _action; 520 final _FutureAction _action;
437 521
438 _WhenFuture(this._action); 522 _WhenFuture(this._action);
439 523
440 void _sendValue(T value) { 524 void _zonedSendValue(T value) {
441 try { 525 try {
442 var result = _action(); 526 var result = _action();
443 if (result is Future) { 527 if (result is Future) {
444 Future resultFuture = result; 528 Future resultFuture = result;
445 resultFuture.then((_) { 529 resultFuture.then((_) {
446 _setValue(value); 530 _setValue(value);
447 }, onError: _setError); 531 }, onError: _setError);
448 return; 532 return;
449 } 533 }
450 } catch (e, s) { 534 } catch (e, s) {
451 _setError(_asyncError(e, s)); 535 _setError(_asyncError(e, s));
452 return; 536 return;
453 } 537 }
454 _setValue(value); 538 _setValue(value);
455 } 539 }
456 540
457 void _sendError(error) { 541 void _zonedSendError(error) {
458 try { 542 try {
459 var result = _action(); 543 var result = _action();
460 if (result is Future) { 544 if (result is Future) {
461 Future resultFuture = result; 545 Future resultFuture = result;
462 // TODO(lrn): Find a way to combine [error] into [e]. 546 // TODO(lrn): Find a way to combine [error] into [e].
463 resultFuture.then((_) { 547 resultFuture.then((_) {
464 _setError(error); 548 _setError(error);
465 }, onError: _setError); 549 }, onError: _setError);
466 return; 550 return;
467 } 551 }
468 } catch (e, s) { 552 } catch (e, s) {
469 error = _asyncError(e, s); 553 error = _asyncError(e, s);
470 } 554 }
471 _setError(error); 555 _setError(error);
472 } 556 }
473 } 557 }
OLDNEW
« no previous file with comments | « sdk/lib/async/event_loop.dart ('k') | sdk/lib/async/timer.dart » ('j') | sdk/lib/async/zone.dart » ('J')

Powered by Google App Engine
This is Rietveld 408576698