Implement async* functions in VM

runtime/lib/core_patch.dart

Index: runtime/lib/core_patch.dart
===================================================================
--- runtime/lib/core_patch.dart	(revision 43971)
+++ runtime/lib/core_patch.dart	(working copy)
@@ -22,11 +22,88 @@
   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 isCancelled = false;

[Lasse Reichstein Nielsen, 2015/02/24 13:12:34]

"isCanceled"
I've recently been informed that one "l" is US English spelling :(

[hausner, 2015/02/25 22:07:41]

I'm using controller.hasListener now. Apparently the spelling with one l is
recent American English. Both spellings are still in use.

+  bool isAdding = false;
+
+  Stream get stream => controller.stream;
+
+  // Adds element to steam, returns true iff the caller should suspend
+  // execution of the generator.
+  bool add(event) {
+    // TODO(hausner): what is the expected behavior if the stream
+    // has been cancelled?
+    if (isCancelled) return true;

[Lasse Reichstein Nielsen, 2015/02/24 13:12:34]

Since you are using a single-subscription controller, and you don't start
running until onListen, you should be able to use !controller.hasListener as
equivalent to isCancelled, and save the cancel callback.

[hausner, 2015/02/25 22:07:41]

Done.

+    controller.add(event);
+    if (true) {

[Lasse Reichstein Nielsen, 2015/02/24 13:12:34]

if (true) ?

[hausner, 2015/02/25 22:07:41]

That was a way to have the dart2js and the spec-compliant behavior in the code
at the same time. I'm implementing the dart2js version now, since apparently the
spec is going to change to conform to that behavior (suspending the generator
after each yield, whether the listener is paused or not.)

+      // Always suspend. Matches dart2js implementation but not Spec.
+      scheduleMicrotask(asyncStarBody);

[Lasse Reichstein Nielsen, 2015/02/24 13:12:34]

If the stream subscription is paused, you should not reschedule the body until
the subscription is resumed.
I think this would work:
  if (!controller.isPaused) scheduleMicrotask(asyncStarBody);
  return true;

[hausner, 2015/02/25 22:07:41]

Done.

+      return true;
+    }
+    // Tell caller to suspend the generator if the controller is paused.
+    return controller.isPaused;
+  }
+
+  // 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) {
+    // TODO(hausner): what is the expected behavior if the stream
+    // has been cancelled?

[Lasse Reichstein Nielsen, 2015/02/24 13:12:34]

If controller.stream has been canceled or if the argument stream has been
canceled?

In the latter case, the addStream will return a future which will be completed
soon (or is already completed, I don't remember - there is no practical
difference).

If the controller.stream has been canceled, adding to it throws, so checking
isCancelled first is fine, but then the asyncStarBody function should start
exiting, as if by a return, so you do need to schedule it.
That is, the behavior of asyncStartBody, when called as a microtask, depends on
whether the subscription has been canceled. 

Maybe let asyncStartBody take a boolean (optional?) argument, and when you pass
it false, it starts exiting due to the cancel.

[hausner, 2015/02/25 22:07:41]

Done.

+    if (isCancelled) return true;
+    isAdding = true;
+    var whenDoneAdding =
+        controller.addStream(stream as Stream, cancelOnError: false);
+    whenDoneAdding.then((_) {
+      isAdding = false;
+      if (!controller.isPaused) {
+        scheduleMicrotask(asyncStarBody);
+      }
+    });
+    return true;
+  }
+
+  addError(error, stackTrace) {
+    controller.addError(error, stackTrace);
+  }
+
+  close() {
+    controller.close();

[Lasse Reichstein Nielsen, 2015/02/24 13:12:34]

If the subscription was paused, you will need to schedule the remainder of the
asyncStartBody, so it can do clean up (execute finallies). 
There won't be an "onResume" call if you cancel in a paused state, it goes
directly to "onCancel", and controller.isPaused will return false in the
canceled state, so you will need to keep the isPaused (or probably isScheduled)
state manually.

[hausner, 2015/02/25 22:07:41]

Done. One thing that is still sup-optimal: the controller does not know whether
the body is already sheduled to run again, so in the worst case it gets
scheduled twice after the stream is cancelled.

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

[Lasse Reichstein Nielsen, 2015/02/24 13:12:34]

You can get two onResumes in a row if someone does:
  subscription.resume();
  subscription.pause();
  subscription.resume();
when there are no buffered elements.
This code would schedule the body twice in that case.
You probably need to keep track of whether it has already been scheduled, and
not reschedule (an _isScheduled flag).


I think rescheduling immediately is better than my implementation in an earlier
mail where I kept checking whether the subscription was still paused. The async*
function should continue executing the first time it's resumed, and not check
again until the next yield.

[hausner, 2015/02/25 22:07:41]

Acknowledged.

+    }
+  }
+
+  onCancel() {
+    isCancelled = true;

[Lasse Reichstein Nielsen, 2015/02/24 13:12:34]

If the subscription was paused, you will need to schedule the remainder of the
asyncStartBody, so it can do clean up (execute finallies). 
There won't be an "onResume" call if you cancel in a paused state, it goes
directly to "onCancel", and controller.isPaused will return false in the
canceled state, so you will need to keep the isPaused (or probably _isScheduled)
state manually.

[hausner, 2015/02/25 22:07:41]

See comment above.

+  }
+}
+
+
 // _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;