Unexpected Behavior with Nested Async Generators

[felangel]

• This tracker is for issues related to: Dart core libraries ("dart:async")
• Dart SDK Version (dart --version): Dart SDK version: 2.12.0-133.2.beta (beta) (Tue Dec 15 09:55:09 2020 +0100) on "macos_x64"
• Whether you are using Windows, MacOSX, or Linux (if applicable): MacOSX

Reproduction Sample

import 'dart:async';

void main() async {
  final bloc = Counter()..listen(print);
  bloc.add(CounterEvent.increment);
}

enum CounterEvent { increment }

class Counter extends Bloc<CounterEvent, int> {
  Counter() : super(0);

  @override
  Stream<int> mapEventToState(CounterEvent event) async* {
    // Does not work
    yield* _increment();

    // Works
    // yield state + 1;
    // yield state + 1;
  }

  Stream<int> _increment() async* {
    yield state + 1;
    yield state + 1;
  }
}

abstract class Bloc<E, S> {
  Bloc(this._state) {
    _eventController.stream.asyncExpand(mapEventToState).listen((s) {
      _state = s;
      _stateController.add(_state);
    });
  }

  final _eventController = StreamController<E>();
  final _stateController = StreamController<S>.broadcast();

  S _state;
  S get state => _state;

  StreamSubscription<S> listen(void Function(S) onData) {
    return _stateController.stream.listen(onData);
  }

  Stream<S> mapEventToState(E event);

  void add(E event) => _eventController.add(event);
}

Reproduction Steps

1. Run above sample observe output is:

1
1

2. Comment line 16 and uncomment lines 19-20

@override
  Stream<int> mapEventToState(CounterEvent event) async* {
    // Does not work
    // yield* _increment();

    // Works
    yield state + 1;
    yield state + 1;
  }

3. Re-run and observe output is:

1
2

Expectations

I would expect that yielding inline would exhibit the same behavior as using yield* with a nested async generator.

[lrhn]

I actually think it should work. I'm guessing that the yield* introduces an extra microtask delay, instead of synchronously (re-)delivering the events of the _increment() stream immediately when they are received.

[felangel]

I actually think it should work. I'm guessing that the yield* introduces an extra microtask delay, instead of synchronously (re-)delivering the events of the _increment() stream immediately when they are received.

Yeah, adding a delay of Duration.zero between the yield statements of the nested generator makes the behavior consistent. Is this a bug then? If so, I’m happy to try to open a pull request to fix it.

[lrhn]

I think it's a bug, but I'm not absolutely sure we have specified yield* in precise enough detail to state with conviction that the behavior is not within the specified bounds.

The behavior of yield was specified precisely for Dart 2.0, and yet not implemented correctly yet on all platforms.

If you have an easy fix, do feel free to upload a PR. If it's complicated, we'll probably have to coordinate it so all the platforms behave the same way. I'll look into the spec.

[felangel]

@lrhn have you had a chance to look into this? I would love to help get this fixed as it’s a pretty high priority for me but would need some guidance on how best to proceed. It looks like the issue might be here

sdk/sdk/lib/_internal/vm/lib/async_patch.dart

Line 136 in d10d17c

scheduleMicrotask(runBody);

but I’m not familiar with modifying the sdk. Any help would be greatly appreciated, thanks!

[lrhn]

I've been reading the specification, and it's ... remarkably vague on everything. I'm working on a rewording.

The scheduleMicrotask you're point to could be a culprit, but it's called from many places. I think a yield* is implemented using addStream instead, and that it should be continuing immediately when the stream ends, not slightly later.

The individual events should not be delayed, though.

[felangel]

@lrhn thanks for the update! Let me know if there's anything I can do to help 👍

[lrhn]

I'm attempting a fix: https://dart-review.googlesource.com/c/sdk/+/187000

Have to make sure the changed timing doesn't break people (too much), otherwise we might be stuck with the current behavior for a while.

[felangel]

@lrhn is there any update on this? Looks like you managed to get it working but are waiting on a test-run? Thanks!

Update: @lhrn do you have a sense of if the fix will be made or if we'll be stuck with the current behavior for a while? Thanks again and sorry for being annoying 😛

[lrhn]

I've been running some internal tests, and the change does break a few tests. Not many, hopefully only a few root causes and if very lucky, it's a problem in the tests, not the real code. It's still code I'm not at all familiar with, so it might take some time to figure out.
I still want to land this change if possible, but it's always a risk to change timing of asynchronous constructs, and a pain to figure out why something starts failing.

[felangel]

@lrhn thanks for the update. Do you have any sense of how difficult addressing the test failures will be? I just want to know if I should start considering alternative approaches as opposed to waiting for a fix. Thanks!

[lrhn]

I'd probably consider alternative approaches. Release cycles are long, and I'm not sure when this will make it (if it will, which I still hope).

[jeroen-meijer]

@felangel Do you have a workaround approach you commonly deploy in situations where this bug arises, or any other tips?

I've been doing something along the lines of

 Stream<int> _increment() async* {
    var newState = state;
    
    newState = state + 1;
    yield newState;
    // ...
    newState = state + 1;
    yield newState;
  }

But I'd love to know if there's a cleaner way of doing this.

Or the bug gets fixed. That'd be cool, too 👀

[felangel]

@felangel Do you have a workaround approach you commonly deploy in situations where this bug arises, or any other tips?

I've been doing something along the lines of

 Stream<int> _increment() async* {
    var newState = state;
    
    newState = state + 1;
    yield newState;
    // ...
    newState = state + 1;
    yield newState;
  }

But I'd love to know if there's a cleaner way of doing this.

Or the bug gets fixed. That'd be cool, too 👀

I generally do something like:

Stream<int> mapEventToState(CounterEvent event) async* {
  switch (event) {
    case CounterEvent.increment:
      yield* _mapIncrementToState(state);
      break;
  }
}

Stream<int> _mapIncrementToState(int state) async* {
  yield state + 1;
  yield state + 2;  
}

But yeah I was hoping this bug would be fixed. I'm currently playing around with alternative APIs which don't rely on nested generators to avoid this bug so I'll keep you posted 👍

[PlugFox]

Problematic without external dependencies

https://dartpad.dev/8fe1bb47b6b5a6a496ced110f426c3a6

//@dart=2.13

typedef State = int;

/// Choose between generators
final stateGenerator = bad; // good;

// Listen generator, update and print current state
void main() => stateGenerator().listen((value) => print(state = value));

/// Current state +
State _state = 0;
set state(State value) {
  print('* set $value');
  _state = value;
}
State get state {
  print('* get $_state');
  return _state;
}
/// Current state -

/// Working properly generator
Stream<State> good() async* {
  yield state + 1;
  yield state + 1;
}

/// Not working properly inline generator
Stream<State> bad() async* {
  Stream<State> _inline() async* {
    yield state + 1;
    yield state + 1;
  }

  yield* _inline();
}

[PlugFox]

Workaround (if someone need quickfix in current project):

Stream<State> mapEventToState(Event _) async* {
  Stream<State> _inline() async* {
    yield state + 1;
    yield state + 1;
  }

  // Problem:
  //yield* _inline();
  
  // Workaround:
  yield* _inline().asyncMap<State>(
    (i) => Future<State>.microtask(() => i),
  );
}

[PlugFox]

Trouble:

Stream<int> mapEventToState(CounterEvent event) async* {
  switch (event) {
    case CounterEvent.increment:
      yield* _mapIncrementToState(state);
      break;
  }
}

Stream<int> _mapIncrementToState(int state) async* {
  yield state + 1;
  yield state + 2;  
}

Solve:

Stream<int> mapEventToState(CounterEvent event) {
  switch (event) {
    case CounterEvent.increment:
      return _mapIncrementToState(state);
  }
}

Stream<int> _mapIncrementToState(int state) async* {
  yield state + 1;
  yield state + 2;  
}

[zs-dima]

@vsmenon @lrhn it looks like Dart language critical issue and not just certain 'library' issue.
Could I kindly ask you if this issue going to be prioritized?

[zs-dima]

@vsmenon @lrhn any updates are welcome

[w461]

I guess, I have noticed the same issue with a slightly different code: I am using a single state and I am updating it with copyWith(). In general, my functions called with yield* work. However in the case of
Stream<CoreStatementState> _mapWordSelectedToState(CsWordSelected event) async* {
statement[event.IndexPosition].selected = !statement[event.IndexPosition].selected;
yield state.copyWith(statement: statement);
}

I get it only do update the UI by chaining this to

Stream<CoreStatementState> _mapWordSelectedToState(CsWordSelected event) async* {
statement[event.IndexPosition].selected = !statement[event.IndexPosition].selected;
yield state.copyWith(coreStatements: problem.coreStatements.map((s) => s.statement).toList(),);
yield state.copyWith(statement: statement, coreStatements: null);
}

[zs-dima]

Any news are welcome

[lrhn]

Having looked at StreamController recently, it seems addStream does introduce an extra delay for asynchronous controllers.
It probably shouldn't, but changing any timing is a precarious endeavor given the horde of timing-specific tests that already exist.

[hawkkiller]

😶

[zs-dima]

Any news about? Or we going to celebrate issue second birthday in a 8 weeks?

[Fasust]

Has this issue been silently fixed in dart version 2.18.0 @lrhn ?

If I execute the original example on a fresh install of dart version 2.18.0, I am getting this print out for the "Does not work" case:

1
2

My dart --version print out:

Dart SDK version: 2.18.0 (stable) (Fri Aug 26 10:22:54 2022 +0000) on "macos_x64"

If I downgrade to dart 2.17.6 the print out is back to

1
1

I stumbled over this while migrate an app to flutter 3.3.10 & some of our async*-methods suddenly had a different behaviour.
Strangely enough, I am unable to reproduce the fixed behaviour on DartPad, despite setting it to the exact same version as my local one.