Cancelling Async Generator Hangs Indefinitely

[tustvold]

Problem

Given the following stream

Stream<String> foo() {
  final controller = StreamController<String>();
  controller.onListen = () => controller.add("FOO");
  controller.onCancel = () => print("Shutting down");
  controller.onPause = () => print("Pause");
  controller.onResume = () => print("Resume");
  return controller.stream;
}

If we then wrap this in

Stream<String> watch() async* {
  final response = foo();
  final mapped = response.map((response) {
    return response.toUpperCase();
  });
  yield* mapped;
}

Everything works correctly

Future main() async {
  final stream = watch().listen((x) => print(x));
  
  await Future.delayed(Duration(seconds: 1));
  
  await stream.cancel();
  print("finished");
}

Prints as expected

FOO
Shutting down
finished

If, however, we change watch to be

Stream<String> watch() async* {
  final response = foo();
  await for (var response in response) {
    yield response.toUpperCase();
  }
}

Cancellation hangs indefinitely.

The specification has the following to say on the topic.

The stream associated with an asynchronous generator could be canceled
by any code with a reference to that stream at any point
where the generator was passivated.
Such a cancellation constitutes an irretrievable error for the generator.
At this point, the only plausible action for the generator is
to clean up after itself via its \FINALLY{} clauses.%

I'm not entirely sure what passivated means, but I think hanging indefinitely is at least not an obvious behaviour if it is intentional.

Platform

I have tested this on Linux (flutter), Android (flutter) and DartPad and the behaviour seems to be consistent

[dart-github-bot]

Summary: Async generator using await for with a cancellable stream hangs indefinitely when the stream is cancelled, contrary to expected behavior and specification. The oncancel callback isn't triggered.

[tustvold]

Interestingly if you modify the reproducer to be

Stream<String> foo() {
  final controller = StreamController<String>();
  controller.onListen = () {
    controller.add("FOO");
    Future.delayed(Duration(seconds: 2), () => controller.add("BAR"));
  };
  controller.onCancel = () => print("Shutting down");
  controller.onPause = () => print("Pause");
  controller.onResume = () => print("Resume");
  return controller.stream;
}

The generator shuts down once BAR is produced, in which case this may be related to, or a duplicate of, #45645.

In particular

Our async* functions are implemented incorrectly on some platforms (maybe all, not sure any more). They do not check after the yield has synchronously delivered an event whether they have been cancelled or paused, they just continue execution.
That means that the subscription.cancel call must wait until the next yield to exit the async* function.

[lrhn]

Cancellation hangs indefinitely.

That seems to be correct and expected behavior.

You have an await for that is currently waiting for the next event, which never comes, when you cancel the surrounding async* function's subscription.
That changes nothing. Such a cancellation has no effect until control reaches a yield or yield* statement, which it never does.
It's no different from the sync*function being stuck at an await future where the future never completes, or an async function being stuck at the same places, it's a stuck-ness unrelated to being async*, so cancel has no effect on it.

The spec text is descriptive. Being "passivated" is presumably it being waiting at a yield or yield*.
It's a little misleading to use the word "error", because there is no error involved.

The effect of cancelling an async* function execution is simple: an active yield* cancels its subscription and waits for the returned future. Then every yield and yield* statement behaves just like return;. No more and no less.
If it's stuck at any point that is not a yield or yield*, it stays stuck.

[tustvold]

Ok, it's a little odd to me that cancellation wouldn't be preemptive, but if that is the way it is that's the way it is.

My challenge is the stream corresponds to a network request, and with this behaviour there is no way to cancel it until the server sends more data. Is there some alternative cancellation mechanism, or is it simply a case of generators can't be used here?

Unless I'm missing something this makes generators a potential gold mine for resource leaks, which should probably be called out as a big warning, unless this behaviour can be changed?

[lrhn]

It's non-trivial to integrate cancellation into the await for loop, probably requires it to not be an async generator to begin with.

You can only cancel at yields because cancelling at any normal await is a bigger risk of resource leak.

If you can cancel at an await future or await for(...), then the future result or next stream value will be dropped on the floor. If that value requires disposing, you have a leak.

It's unexpected to have a future that never completes or a steam that never emits another event (not even a done event). You may have to wait for a while, but you shouldn't wait forever. That means that leaks should be at most temporary.

[chipweinberger]

I just hit this somewhat surprising behavior as well.

Being preemptive would be way too crazy, but I assumed if a generator was fully stopped at an "await", it could be canceled without waiting for the next yield.

Now that I know this behavior it's fine (and very likely the right decision), but I did have to do some hacky looking stuff to fix perf in my app. otherwise I had too many streams waiting to be canceled.

e.g.

Future<void> untilCancel = _streamSubscription.cancel(); // mark canceled
_pokeStreams(); // trigger yields in generators
await untilCancel;