Implement async* functions in VM

runtime/lib/core_patch.dart

Index: runtime/lib/core_patch.dart
===================================================================
--- runtime/lib/core_patch.dart	(revision 44030)
+++ runtime/lib/core_patch.dart	(working copy)
@@ -26,11 +26,102 @@
   String toString() => _enum_names[index];
 }
 
-typedef bool SyncGeneratorCallback(Iterator iterator);
 
+// _AsyncStarStreamController is used by the compiler to implement
+// async* generator functions.
+class _AsyncStarStreamController {
+  StreamController controller;
+  Function asyncStarBody;
+  bool isAdding = false;
+  bool isCancelled = false;
+  bool onListenReceived = false;
+  bool isClosed = false;
+
+  Stream get stream => controller.stream;
+
+  // Adds element to steam, returns true iff the caller should suspend

[Lasse Reichstein Nielsen, 2015/02/26 14:54:52]

suspend -> terminate/return from?

[hausner, 2015/02/26 21:54:07]

Correct. 

+  // execution of the generator.
+  bool add(event) {
+    if (!onListenReceived) _fatal("yield before stream is listened to!");
+    // If stream is cancelled, tell caller to exit the async generator.

[Lasse Reichstein Nielsen, 2015/02/26 14:54:52]

The spec says that you should pause first if the stream is paused, and then
terminate the generator if the stream is cancelled before it is resumed.
That means that the boolean saying whether to exit the generator isn't known
until either the onCancel or the onResume callback, whichever comes first.
That's why I suggest that the asyncStartBody takes a boolean argument, instead
of just having the add method returning a bool (which bight also be fine,
because it quickly handles the case where the subscription is cancelled before
the yield, but might also be spec violating if the yield is specced to always
yield control before continuing - while(true)try{yield 42;}finally(continue;}
would be an an uninterruptible loop if a cancelled subscription made the yield
try to exit immediately).

This implementation will "work" (generate the same stream events), but it will
run non-finally generator code after the subscription is canceled, which
shouldn't happen, and it will run until it hits the *next* yield, which may
cause visible side effects.
So, if you do:

 import "dart:async";
 main() async {
   var c1 = new Completer();
   var c2 = new Completer();
   var c3 = new Completer();
   var stream = () async* { 
     print("*1"); 
     yield 1; 
     c1.complete(); 
     await c2.future; 
     c3.complete();
     print("*2"); 
     yield 2; 
     print("*3");
   } ();
   var sub = stream.listen(print);
   await c1.future;  // prints "*1", "1"
   sub.pause();
   c2.complete();    // prints "*2"
   await c3.future;
   sub.cancel();     // should not print 2 or "*3"
 }
 
you will also print "*3". The current dart2js implementation only prints "*1",
"1" and "*2".

[hausner, 2015/02/26 21:54:07]

I am leaving this as a TODO for later. The body closure already has 3 (optional)
parameters that are used if there is an await in the async* code. I don't want
to make things more complex at this point. 

As a note: instead of passing a parameter to the closure body, I think it would
also be ok to add a getter to the AsyncStarStreamController that returns whether
the stream is cancelled. Then, the compiler could generate this code for a yield
e statement:

  controller.add(e);
  suspend_generator;
  if (controller.isCancelled) return;

Would that work?

[Lasse Reichstein Nielsen, 2015/02/27 07:51:12]

Looks fine. It basically what I did in an attempt to rewrite async* to async
(no-start) code. The suspend_generator will suspend either until a microtask,
or, if paused, until the resume.
(I assume this is pre-transformed pseudo code).

+    if (!controller.hasListener) {
+      return true;
+    }
+    controller.add(event);
+    if (!controller.isPaused) {
+      scheduleMicrotask(asyncStarBody);
+    }
+    return false;
+  }
+
+  // Adds stream, tells caller to suspend execution of the generator
+  // function. The generator will be scheduled again when all of the
+  // elements of the added stream have been consumed.
+  bool addStream(Stream stream) {
+    assert(onListenReceived);
+    // If stream is cancelled, tell caller to exit the async generator.
+    if (!controller.hasListener) return true;
+    isAdding = true;
+    var whenDoneAdding =
+        controller.addStream(stream as Stream, cancelOnError: false);
+    whenDoneAdding.then((_) {
+      isAdding = false;
+      if (!controller.isPaused) {
+        scheduleMicrotask(asyncStarBody);
+      }
+    });
+    return false;
+  }
+
+  addError(error, stackTrace) {
+    if (!onListenReceived) _fatal("yield before stream is listened to!");
+    // If stream is cancelled, tell caller to exit the async generator.
+    if (!controller.hasListener) return true;

[Lasse Reichstein Nielsen, 2015/02/26 14:54:52]

I think an uncaught exception reaching the stream can only happen if control has
already left tbe body of the generator (that's why it's uncaught). A yield
statement cannot add an error to the stream.
That should mean that the return value is irrelevant. Returning here is still
fine, it just doesn't have to be "true".

[hausner, 2015/02/26 21:54:08]

Yes, this method is only called from the implicit try-catch around the async*
body. In that case, control flow is on its way out of the generator. The only
thing that will happen after this call is the invocation of the close() method
from the finally clause.

Made addError() a void function and removed the code to schedule the body again.

+    controller.addError(error, stackTrace);
+    if (!controller.isPaused) {
+      scheduleMicrotask(asyncStarBody);
+      return false;
+    }
+

[Lasse Reichstein Nielsen, 2015/02/26 14:54:52]

It returns null if the controller isn't paused - which as argued above is fine,
but might as well make it a void function.

[hausner, 2015/02/26 21:54:07]

Done.

+  }
+
+  close() {
+    isClosed = true;
+    controller.close();
+  }
+
+  _AsyncStarStreamController(this.asyncStarBody) {
+    controller = new StreamController(onListen: this.onListen,
+                                      onResume: this.onResume,
+                                      onCancel: this.onCancel);
+  }
+
+  onListen() {
+    assert(!onListenReceived);
+    onListenReceived = true;
+    scheduleMicrotask(asyncStarBody);
+  }
+
+  onResume() {
+    if (!isAdding) {
+      scheduleMicrotask(asyncStarBody);

[Lasse Reichstein Nielsen, 2015/02/26 14:54:52]

This is still error-prone if you get another resume (or just a resume) before
the already scheduled body is run.

I would move this into a helper function:
  scheduleBody();
which is:
  void scheduleBody() {
    if (_isScheduled || controller.isPaused || _isAdding) return;
    _isScheduled = true;
    scheduleMicrotask(runBody);
  }
  // could have two versions of runBody if you want to send a bool back.
  void runBody() {      
    _isScheduled = false;
    asyncStarBody();
  }
with a corresponding flag at the top.
That definitely ensures that you only schedule the body once until it has run
again.

[hausner, 2015/02/26 21:54:07]

Ok, done. I was thinking about this possibility yesterday but wanted to avoid
the extra function call. In the grand scheme of things, that is probably not
worth worrying about.

+    }
+  }
+
+  onCancel() {
+    if (!isClosed) {

[Lasse Reichstein Nielsen, 2015/02/26 14:54:52]

You can use controller.isClosed. It's set to true when you call
controller.close().

[hausner, 2015/02/26 21:54:07]

Done.

+      scheduleMicrotask(asyncStarBody);
+    }
+  }
+}
+
+
 // _SyncIterable and _syncIterator are used by the compiler to
 // implement sync* generator functions. A sync* generator allocates
 // and returns a new _SyncIterable object.
+
+typedef bool SyncGeneratorCallback(Iterator iterator);
+
 class _SyncIterable extends IterableBase {
   // moveNextFn is the closurized body of the generator function.
   final SyncGeneratorCallback moveNextFn;