Coroutines/Flow vs add/remove listener (synchronous execution)

[valeriyo]

1. Consider the traditional "add/remove listener" pattern, with synchronous event handlers:

interface MyEventListener {
  fun handle(myEvent: MyEvent)
}

fun add(listener: MyEventListener)
fun remove(listener: MyEventListener)

2. With RxJava, the same can be accomplished with:

val myEvent: Observable<MyEvent>

because RxJava is synchronous by default. (To opt-out from synchronous execution, a subscriber can apply observeOn with some Scheduler - but this is not what we're trying to do here.. we really want synchronisity.)

2. With coroutines, this would be the analogue to the RxJava solution:

val myEvent: Flow<MyEvent>

However, coroutines are asynchronous by default, and there seems to be no way to make their execution synchronous (reliably). Neither the CoroutineStart parameter of launch, nor Main.immediate guarantee that the execution will be synchronous in all cases.

The Main.immediate is strange in particular, because it is "immediate" only under perfect conditions:

1. "dispatcher that executes coroutines immediately when it is already in the right context"
2. "dispatcher is safe from stack overflows and in case of nested invocations forms event-loop"

Am I missing something?

[elizarov]

You can collect a flow in Dispatchers.Unconfined for best-effort synchronicity. Your experience will be similar to RxJava in this case. That is, everything will be synchronous unless you are using some asynchronous operator.

[qwwdfsad]

It would be nice to understand what exactly you are trying to achieve and what exactly you mean by asynchronous and synchronous.

Flows are asynchronous as in "flow supports backpressure by default and backpressure can be triggered by any suspending operator". In that sense, it's impossible to prevent upstream/operators in a chain from suspending if you are willing to.

In the sense of execution context, it is the same as with Flowable:

• By default, both flow(...) and Flowable.from(...) are executed in the context (for flow) or in the thread of (for flowable) the collector/subscriber, preserving execution context
• Once you have flowOn or subscribeOn prior to laying your hands on the resulting flow/flowable, all bets are off -- you cannot really change the execution context of the upstream, you can only affect downstream operators

[valeriyo]

You can collect a flow in Dispatchers.Unconfined for best-effort synchronicity. Your experience will be similar to RxJava in this case. That is, everything will be synchronous unless you are using some asynchronous operator.

You just said it -- "best-effort". It's stuff like this:

Nested coroutines launched in this dispatcher form an event-loop to avoid stack overflows.

withContext(Dispatchers.Unconfined) {
   println(1)
   withContext(Dispatchers.Unconfined) { // Nested unconfined
       println(2)
   }
   println(3)
}
println("Done")

Can print both "1 2 3" and "1 3 2". This is an implementation detail that can be changed.

How is this randomness is acceptable as "normal" behavior? Why has stack-overflow prevention become such an important matter that it superseded execution ordering and predictability? As a result, there is no reliable way to keep execution synchronous.

[qwwdfsad]

Why has stack-overflow prevention become such an important matter that it superseded execution ordering and predictability? As a result, there is no reliable way to keep execution synchronous.

Two points about that:

• Dispatchers.Unconfined main purpose is not to keep execution order deterministic, it has a completely different purpose -- to provide a way to execute coroutines somewhere in the context of the caller, yet to prevent stack-overflows which is crucial -- we didn't have it in our early days, and it was a steady state of both weird-hacky bugs and behaviours, as well as rarely-reproducible crashes in people's production apps.
• "Can print both ... " is intentional to keep people away from relying on the particular order of execution and keep our hands free in case if we ever want to change event loop's implementation details. Yet the order is deterministic and, for the fixed version of kotlinx.coroutines and fixed coroutines hierarchy, it never changes. We just don't provide a guarantee that the particular order won't change.

I would appreciate it if you could be more constructive and provide actual use cases and the kind of problems you are trying to solve instead of implying that the behaviour of well-documented and battle-tested API should not be "acceptable as normal"

[valeriyo]

It would be nice to understand what exactly you are trying to achieve and what exactly you mean by asynchronous and synchronous.

As I explained in the original post, I'm simply trying to replace the "add/remove listener" use-cases with an observer pattern. Why is synchronisity important? Because there are definitely situations when an event has to be propagated to all registered listeners before the next operation can be performed.

[valeriyo]

Dispatchers.Unconfined

I only mentioned it because @elizarov brought it up as a recommendation. The Main.immediate dispatcher suffers from the same problem - it's "immediate" only sometimes, not always.

[dkhalanskyjb]

So, what you actually want is to be able to block the generation of the next element in a Flow until all the consumers have consumed it?

[qwwdfsad]

As I explained in the original post, I'm simply trying to replace the "add/remove listener" use-cases with an observer pattern.

As I've mentioned in #3506 (comment), by default everything should be synchronous unless context-switching operators are used.

If you find it different, please post a snippet of code with a small reproducer, for example:

val flowable = ... flowable ...
val flow = ...create flow ...

suspend fun main() {
    flowable.onEach { println(it) }.getBlocking() // Prints "1 2 3" as expected
    flow.onEach { println(it) }.collect() // Prints "1 3 2" though you clearly expected "1 2 3"
}

[valeriyo]

So, what you actually want is to be able to block the generation of the next element in a Flow until all the consumers have consumed it?

There doesn't need to be "next element". I am trying to achieve same behavior as I would do with normal callbacks:

1. Some event occurs (e.g. user is authenticated)
2. All registered listeners are notified.
3. Now I can navigate to the user's app screen.

With coroutines/Flow this seems to be impossible to do (reliably) since you can't "know" that all listeners have been notified (i.e. "collectors" have been invoked) - because, well, even the "immediate" dispatcher isn't always immediate.

[valeriyo]

by default everything should be synchronous unless context-switching operators are used.

Well, if an event is emitted in a nested context, then instead of executing immediately, an event queue will being used, and the "immediate" collector will run asynchronously -- that's the "gotcha" with the immediate dispatcher.

This is done by design in the name of stack overflow prevention, and from my point of view, it is counter-productive. It essentially makes the immediate dispatcher a misnomer - it's not immediate, and if you use it (as the name suggests), you will hit a situation when it becomes a source of bugs.

[valeriyo]

Consider this example:

    val isAuthed = MutableStateFlow(false)
    var hasUser = false

    GlobalScope.launch(Main.immediate) {
      isAuthed.collect { hasUser = it }
    }

    GlobalScope.launch {
      isAuthed.value = true
      check(hasUser)
    }

When it's run without interruptions, it works (hasUser == true):

However, when I set a breakpoint at the "{ hasUser = it }" lambda, it fails:

And adding immediate dispatcher to the second launch also makes the check fail:

    val isAuthed = MutableStateFlow(false)
    var hasUser = false

    GlobalScope.launch(Main.immediate) {
      isAuthed.collect { hasUser = it }
    }

    GlobalScope.launch(Main.immediate) {
      isAuthed.value = true
      check(hasUser)
    }

I'm using:

org.jetbrains.kotlinx:kotlinx-coroutines-android:1.6.4
Kotlin 1.7.10
Android Studio Dolphin | 2021.3.1 Patch 1

[valeriyo]

I opened #3511 for the breakpoint thing... since it's not specific to the immediate dispatcher.

[dkhalanskyjb]

From your code:

    val isAuthed = MutableStateFlow(false)
    var hasUser = false

    GlobalScope.launch(Main.immediate) {
      isAuthed.collect { hasUser = it }
    }

    GlobalScope.launch {
      isAuthed.value = true
      check(hasUser)
    }

Here's what happens here: GlobalScope runs its coroutines in Dispatchers.Default, a coroutine scheduler that uses a thread pool. isAuthed.value = true asks the collecting coroutine to hasUser = it, but Main.immediate is a dispatcher that is only immediate if the resumption request is coming from the UI thread already. The worker in Dispatchers.Default is not the UI thread. Therefore, hasUser = it is not executed immediately, it's put into the UI thread queue to execute. This behavior is important, as some operations are only allowed to be executed on the UI thread. If your code was

    val isAuthed = MutableStateFlow(false)
    var hasUser = false

    GlobalScope.launch(Main.immediate) {
      isAuthed.collect { updateUi() }
    }

    GlobalScope.launch {
      isAuthed.value = true
      check(hasUser)
    }

it would be incorrect for isAuthed.value = true to immediately also call updateUi() from the Dispatchers.Default thread pool.

What you can do is

    val isAuthed = MutableStateFlow(false)
    var hasUser = false

    GlobalScope.launch(Dispatchers.Unconfined) {
      isAuthed.collect { hasUser = true }
    }

    GlobalScope.launch {
      isAuthed.value = true
      check(hasUser)
    }

This way, isAuthed.value = true will immediately also call hasUser = true.

Does this clear things up?