Dart2js async* functions only resume execution of the body when suspended on yield.

sdk/lib/_internal/compiler/js_lib/js_helper.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 _js_helper;
 
 import 'dart:_async_await_error_codes' as async_error_codes;
 
 import 'dart:_js_embedded_names' show
     DEFERRED_LIBRARY_URIS,
@@ skipping 3723 matching lines @@
     return;
   }
 
   if (object is IterationMarker) {
     if (controller.cancelationCompleter != null) {
       _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
           async_error_codes.STREAM_WAS_CANCELED)(null);
       return;
     }
     if (object.state == IterationMarker.YIELD_SINGLE) {
+      controller.isWaitingForYield = true;
       controller.add(object.value);
       // If the controller is paused we stop producing more values.
       if (controller.isPaused) {
         return;
       }
       // TODO(sigurdm): We should not suspend here according to the spec.

[Lasse Reichstein Nielsen, 2015/04/08 14:12:14]

Isn't the spec allowing you to do so, but not requiring it (as long as you
"eventually" break a loop of yields)?
I think it's the right thing to do - output the value, allow the listener to
process it, then go back to creating the next value.

[sigurdm, 2015/04/09 11:24:08]

Yes I think so - this is an outdated comment. Removed.

       scheduleMicrotask(() {
+        controller.isWaitingForYield = false;

[Lasse Reichstein Nielsen, 2015/04/08 14:12:15]

I think this line should be moved outside of the scheduleMicrotask. At the time
you do the scheduleMicrotask, you have already decided that you are not paused,
and execution will continue. Alternatively, as written below, do the isPaused
check in the scheduleMicrotask callback.

[sigurdm, 2015/04/09 11:24:08]

Done.

         _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
                                 async_error_codes.SUCCESS)
             (null);
       });

[Lasse Reichstein Nielsen, 2015/04/08 14:12:14]

I think I would prefer waiting until the scheduleMicrotask callback to check if
the output stream is paused. That allows listeners on the output stream to pause
in response to the just output value, and have it immediately stop working.

It also avoids the race condition of the subscription not being paused in the
controller.isPaused text above, but becoming paused *and* resumed again before
the scheduleMicrotask callback is run. I think that would still cause multiple
resumes below.

[sigurdm, 2015/04/09 11:24:08]

Done.

       return;
     } else if (object.state == IterationMarker.YIELD_STAR) {
       Stream stream = object.value;
       controller.isAdding = true;
       // Errors of [stream] are passed though to the main stream. (see
       // [AsyncStreamController.addStream].

[Lasse Reichstein Nielsen, 2015/04/08 14:12:15]

Missing end-paren :)

[sigurdm, 2015/04/09 11:24:08]

Thanks!

       // TODO(sigurdm): The spec is not very clear here. Clarify with Gilad.
       controller.addStream(stream).then((_) {
         controller.isAdding = false;
         _wrapJsFunctionForAsync(bodyFunctionOrErrorCode,
                                 async_error_codes.SUCCESS)(null);
       });
       return;
     }
   }
 
@@ skipping 18 matching lines @@
 ///
 /// Also has a subSubscription that when not null will provide events for the
 /// stream, and will be paused and resumed along with this controller.
 class AsyncStarStreamController {
   StreamController controller;
   Stream get stream => controller.stream;
   Completer cancelationCompleter = null;
   bool get isCanceled => cancelationCompleter != null;
   bool isAdding = false;
   bool isPaused = false;
+  bool isWaitingForYield = false;

[floitsch, 2015/04/08 14:01:25]

isAtYield ?

Add comment what exactly this means. For example.
/// True, when the async* function just yielded a value, but hasn't
/// resumed execution yet.

[sigurdm, 2015/04/08 14:07:40]

Done.

[Lasse Reichstein Nielsen, 2015/04/08 14:12:14]

Consider documenting these fields, preferably by describing the possible states
that they controller can be in (including any scheduleMicrotask callbacks that
may be scheduled in some states). 
I find that being extremely explicit about the states is the best way to avoid
an accidental inconsistent state. 

[sigurdm, 2015/04/09 11:24:08]

Good advice!
I realized I can leave out `isAdding` it contains no useful information beyond
isSuspended. Similarly i maintained my own isPaused to be able to decide if to
resume after onCancel. That should depend only on isSuspended.

+
   add(event) => controller.add(event);
   addStream(Stream stream) {
     return controller.addStream(stream, cancelOnError: false);
   }
   addError(error, stackTrace) => controller.addError(error, stackTrace);
   close() => controller.close();
 
   AsyncStarStreamController(body) {
     controller = new StreamController(
       onListen: () {
         scheduleMicrotask(() {
-          Function wrapped = _wrapJsFunctionForAsync(body,
-                                                     async_error_codes.SUCCESS);
+          Function wrapped =
+              _wrapJsFunctionForAsync(body, async_error_codes.SUCCESS);
           wrapped(null);
         });
       },
       onPause: () {
         isPaused = true;
       }, onResume: () {
         isPaused = false;
-        if (!isAdding) {
+        if (!isAdding && isWaitingForYield) {

[Lasse Reichstein Nielsen, 2015/04/08 14:12:15]

I don't think this is safe enough. 
If the stream becomes both paused after setting isWaitingForYield to true, but
before the following scheduleMicrotask callback being run, and it is also
resumed again in that time, then this code will run *as well* as the
scheduleMicrotask callback.

[Lasse Reichstein Nielsen, 2015/04/08 14:12:15]

Maybe name it "isWaitingForResume"? You have already executed the yield, and is
now blocked after the yield because the output stream subscription is paused.

Should isWaitingForYield be set to false somewhere?

[sigurdm, 2015/04/09 11:24:08]

It is now isSuspended

           asyncStarHelper(null, body, this);
         }
       }, onCancel: () {
         if (!controller.isClosed) {
           cancelationCompleter = new Completer();
           if (isPaused) asyncStarHelper(null, body, this);

[Lasse Reichstein Nielsen, 2015/04/08 14:12:14]

Are you sure this is safe?
You can have a call to asyncStarHelper already scheduled with scheduleMicrotask,
and then two rapid calls to sub.pause() and sub.cancel() would end up here,
scheduling another continuation.

It's generally not clear how you determine if a call to asyncStarHelper is
already scheduled somehow (schedule microtask/waiting on future), which is what
is really important before scheduling another call.

[sigurdm, 2015/04/09 11:24:08]

True - I need a better mechanism for this.
Now I only ever schedule a new call if isSuspended.

 
           return cancelationCompleter.future;
         }
       });
   }
 }
 
 makeAsyncStarController(body) {
   return new AsyncStarStreamController(body);
 }
@@ skipping 96 matching lines @@
   // This is a function that will return a helper function that does the
   // iteration of the sync*.
   //
   // Each invocation should give a body with fresh state.
   final dynamic /* js function */ _outerHelper;
 
   SyncStarIterable(this._outerHelper);
 
   Iterator get iterator => new SyncStarIterator(JS('', '#()', _outerHelper));
 }