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

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

Issue 12321010: Increase size for slow-consumer test (overflows memory on X64). (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Done Created 7 years, 10 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/slow_consumer_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 // States shared by single/multi stream implementations. 7 // States shared by single/multi stream implementations.
8 8
9 /// Initial and default state where the stream can receive and send events. 9 /// Initial and default state where the stream can receive and send events.
10 const int _STREAM_OPEN = 0; 10 const int _STREAM_OPEN = 0;
11 /// The stream has received a request to complete, but hasn't done so yet. 11 /// The stream has received a request to complete, but hasn't done so yet.
12 /// No further events can be added to the stream. 12 /// No further events can be added to the stream.
13 const int _STREAM_CLOSED = 1; 13 const int _STREAM_CLOSED = 1;
14 /// The stream has completed and will no longer receive or send events. 14 /// The stream has completed and will no longer receive or send events.
15 /// Also counts as closed. The stream must not be paused when it's completed. 15 /// Also counts as closed. The stream must not be paused when it's completed.
16 /// Always used in conjunction with [_STREAM_CLOSED]. 16 /// Always used in conjunction with [_STREAM_CLOSED].
17 const int _STREAM_COMPLETE = 2; 17 const int _STREAM_COMPLETE = 2;
18 /// Bit that alternates between events, and listeners are updated to the 18 /// Bit that alternates between events, and listeners are updated to the
19 /// current value when they are notified of the event. 19 /// current value when they are notified of the event.
20 const int _STREAM_EVENT_ID = 4; 20 const int _STREAM_EVENT_ID = 4;
21 const int _STREAM_EVENT_ID_SHIFT = 2; 21 const int _STREAM_EVENT_ID_SHIFT = 2;
22 /// Bit set while firing and clear while not. 22 /// Bit set while firing and clear while not.
23 const int _STREAM_FIRING = 8; 23 const int _STREAM_FIRING = 8;
24 /// Bit set while calling a pause-state or subscription-state change callback. 24 /// Bit set while calling a pause-state or subscription-state change callback.
25 const int _STREAM_CALLBACK = 16; 25 const int _STREAM_CALLBACK = 16;
26 /// The count of times a stream has paused is stored in the 26 /// The count of times a stream has paused is stored in the
27 /// state, shifted by this amount. 27 /// state, shifted by this amount.
28 const int _STREAM_PAUSE_COUNT_SHIFT = 8; 28 const int _STREAM_PAUSE_COUNT_SHIFT = 5;
29 29
30 // States for listeners. 30 // States for listeners.
31 31
32 /// The listener is currently not subscribed to its source stream. 32 /// The listener is currently not subscribed to its source stream.
33 const int _LISTENER_UNSUBSCRIBED = 0; 33 const int _LISTENER_UNSUBSCRIBED = 0;
34 /// The listener is actively subscribed to its source stream. 34 /// The listener is actively subscribed to its source stream.
35 const int _LISTENER_SUBSCRIBED = 1; 35 const int _LISTENER_SUBSCRIBED = 1;
36 /// The listener is subscribed until it has been notified of the current event. 36 /// The listener is subscribed until it has been notified of the current event.
37 /// This flag bit is always used in conjuction with [_LISTENER_SUBSCRIBED]. 37 /// This flag bit is always used in conjuction with [_LISTENER_SUBSCRIBED].
38 const int _LISTENER_PENDING_UNSUBSCRIBE = 2; 38 const int _LISTENER_PENDING_UNSUBSCRIBE = 2;
(...skipping 102 matching lines...) Expand 10 before | Expand all | Expand 10 after
141 141
142 /** Whether the stream has been closed (a done event requested). */ 142 /** Whether the stream has been closed (a done event requested). */
143 bool get _isClosed => (_state & _STREAM_CLOSED) != 0; 143 bool get _isClosed => (_state & _STREAM_CLOSED) != 0;
144 144
145 /** Whether the stream is completed. */ 145 /** Whether the stream is completed. */
146 bool get _isComplete => (_state & _STREAM_COMPLETE) != 0; 146 bool get _isComplete => (_state & _STREAM_COMPLETE) != 0;
147 147
148 /** Whether one or more active subscribers have requested a pause. */ 148 /** Whether one or more active subscribers have requested a pause. */
149 bool get _isPaused => _state >= (1 << _STREAM_PAUSE_COUNT_SHIFT); 149 bool get _isPaused => _state >= (1 << _STREAM_PAUSE_COUNT_SHIFT);
150 150
151 /** Whether we are currently executing a state-chance callback. */
152 bool get _isInCallback => (_state & _STREAM_CALLBACK) != 0;
153
151 /** Check whether the pending event queue is non-empty */ 154 /** Check whether the pending event queue is non-empty */
152 bool get _hasPendingEvent => 155 bool get _hasPendingEvent =>
153 _pendingEvents != null && !_pendingEvents.isEmpty; 156 _pendingEvents != null && !_pendingEvents.isEmpty;
154 157
155 /** Whether we are currently firing an event. */ 158 /** Whether we are currently firing an event. */
156 bool get _isFiring => (_state & _STREAM_FIRING) != 0; 159 bool get _isFiring => (_state & _STREAM_FIRING) != 0;
157 160
158 /** Whether the state bits allow firing. */ 161 /** Whether the state bits allow firing. */
159 bool get _mayFireState { 162 bool get _mayFireState {
160 // The state disallows firing if: 163 // The state disallows firing if:
(...skipping 63 matching lines...) Expand 10 before | Expand all | Expand 10 after
224 // This sets the _STREAM_FIRING bit and toggles the _STREAM_EVENT_ID 227 // This sets the _STREAM_FIRING bit and toggles the _STREAM_EVENT_ID
225 // bit. All current subscribers will now have a _LISTENER_EVENT_ID 228 // bit. All current subscribers will now have a _LISTENER_EVENT_ID
226 // that doesn't match _STREAM_EVENT_ID, and they will receive the 229 // that doesn't match _STREAM_EVENT_ID, and they will receive the
227 // event being fired. 230 // event being fired.
228 _state ^= _STREAM_FIRING | _STREAM_EVENT_ID; 231 _state ^= _STREAM_FIRING | _STREAM_EVENT_ID;
229 } 232 }
230 233
231 void _endFiring() { 234 void _endFiring() {
232 assert(_isFiring); 235 assert(_isFiring);
233 _state ^= _STREAM_FIRING; 236 _state ^= _STREAM_FIRING;
234
235 if (!_hasSubscribers) { 237 if (!_hasSubscribers) {
236 _callOnSubscriptionStateChange(); 238 _callOnSubscriptionStateChange();
237 } else if (_isPaused) { 239 } else if (_isPaused) {
238 _callOnPauseStateChange(); 240 _callOnPauseStateChange();
239 } 241 }
240 } 242 }
241 243
242 /** 244 /**
243 * Record that a listener wants a pause from events. 245 * Record that a listener wants a pause from events.
244 * 246 *
(...skipping 80 matching lines...) Expand 10 before | Expand all | Expand 10 after
325 * invalid state. 327 * invalid state.
326 * 328 *
327 * This method must not be called while [isFiring] is true. 329 * This method must not be called while [isFiring] is true.
328 */ 330 */
329 void _forEachSubscriber(void action(_StreamSubscriptionImpl<T> subscription)); 331 void _forEachSubscriber(void action(_StreamSubscriptionImpl<T> subscription));
330 332
331 /** Calls [_onPauseStateChange] while setting callback bit. */ 333 /** Calls [_onPauseStateChange] while setting callback bit. */
332 void _callOnPauseStateChange() { 334 void _callOnPauseStateChange() {
333 // After calling [_close], all pauses are handled internally by the Stream. 335 // After calling [_close], all pauses are handled internally by the Stream.
334 if (_isClosed) return; 336 if (_isClosed) return;
335 _state |= _STREAM_CALLBACK; 337 if (!_isInCallback) {
336 _onPauseStateChange(); 338 _state |= _STREAM_CALLBACK;
337 _state ^= _STREAM_CALLBACK; 339 _onPauseStateChange();
340 _state ^= _STREAM_CALLBACK;
341 } else {
342 _onPauseStateChange();
343 }
338 } 344 }
339 345
340 /** Calls [_onSubscriptionStateChange] while setting callback bit. */ 346 /** Calls [_onSubscriptionStateChange] while setting callback bit. */
341 void _callOnSubscriptionStateChange() { 347 void _callOnSubscriptionStateChange() {
342 _state |= _STREAM_CALLBACK; 348 if (!_isInCallback) {
343 _onSubscriptionStateChange(); 349 _state |= _STREAM_CALLBACK;
344 _state ^= _STREAM_CALLBACK; 350 _onSubscriptionStateChange();
351 _state ^= _STREAM_CALLBACK;
352 } else {
353 _onSubscriptionStateChange();
354 }
345 } 355 }
346 356
347 /** 357 /**
348 * Called when the first subscriber requests a pause or the last a resume. 358 * Called when the first subscriber requests a pause or the last a resume.
349 * 359 *
350 * Read [isPaused] to see the new state. 360 * Read [isPaused] to see the new state.
351 */ 361 */
352 void _onPauseStateChange() {} 362 void _onPauseStateChange() {}
353 363
354 /** 364 /**
(...skipping 109 matching lines...) Expand 10 before | Expand all | Expand 10 after
464 */ 474 */
465 class _SingleStreamImpl<T> extends _StreamImpl<T> { 475 class _SingleStreamImpl<T> extends _StreamImpl<T> {
466 _StreamListener _subscriber = null; 476 _StreamListener _subscriber = null;
467 477
468 // A single-stream is considered paused when it has no subscriber. 478 // A single-stream is considered paused when it has no subscriber.
469 // Exception is when it's complete (which only matters for pause-state-change 479 // Exception is when it's complete (which only matters for pause-state-change
470 // callbacks), where it's not considered paused. 480 // callbacks), where it's not considered paused.
471 bool get _isPaused => (!_hasSubscribers && !_isComplete) || super._isPaused; 481 bool get _isPaused => (!_hasSubscribers && !_isComplete) || super._isPaused;
472 482
473 483
474 // A single-stream is considered paused when it has no subscriber. 484 bool get _canFireEvent {
475 bool get _canFireEvent => 485 // A single-stream is considered paused when it has no subscriber and
476 _mayFireState && !_hasPendingEvent && _hasSubscribers; 486 // isn't complete, so it can't fire if there is no subscriber.
487 // It won't try to fire when completed, so no need to check that.
488 return _mayFireState && !_hasPendingEvent && _hasSubscribers;
489 }
477 490
478 491
479 /** Whether there is currently a subscriber on this [Stream]. */ 492 /** Whether there is currently a subscriber on this [Stream]. */
480 bool get _hasSubscribers => _subscriber != null; 493 bool get _hasSubscribers => _subscriber != null;
481 494
482 // ------------------------------------------------------------------- 495 // -------------------------------------------------------------------
483 // Internal implementation. 496 // Internal implementation.
484 497
485 /** 498 /**
486 * Create the new subscription object. 499 * Create the new subscription object.
(...skipping 714 matching lines...) Expand 10 before | Expand all | Expand 10 after
1201 onError: this._signalError, 1214 onError: this._signalError,
1202 onDone: this._close); 1215 onDone: this._close);
1203 } else { 1216 } else {
1204 // TODO(lrn): Check why this can happen. 1217 // TODO(lrn): Check why this can happen.
1205 if (_subscription == null) return; 1218 if (_subscription == null) return;
1206 _subscription.cancel(); 1219 _subscription.cancel();
1207 _subscription = null; 1220 _subscription = null;
1208 } 1221 }
1209 } 1222 }
1210 } 1223 }
OLDNEW
« no previous file with comments | « no previous file | tests/lib/async/slow_consumer_test.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698