Hi, I do not understand the following behavior: ``...

# coroutines

[Gilles Barbier]

Hi, I do not understand the following behavior:

fun main() {
    val flow = MutableSharedFlow<Int>(replay = 0)

    runBlocking {
        launch {
            println( flow.first { it >= 90 } )
        }

        launch(<http://Dispatchers.IO|Dispatchers.IO>) {
            repeat(1000) {
                flow.emit(Random.nextInt(0, 100))
            }
        }
    }
}

This code does not print anything. It prints when I remove <http://Dispatchers.IO|Dispatchers.IO> at the second launch or (sometimes) if I add <http://Dispatchers.IO|Dispatchers.IO> at the first launch... any explanation?

[Zach Klippenstein (he/him) [MOD]]

I think this is what’s happening:

The first launch doesn’t start immediately – it schedules the coroutine to start on the event loop created by runBlocking. The second launch schedules its coroutine to start on the IO dispatcher, which is backed by a thread pool. It’s possible for one of the threads in that pool to pick up the new coroutine, and finish executing it (all 1000 iterations of your loop) before the first launched coroutine has a chance to start collecting from your flow.

[Zach Klippenstein (he/him) [MOD]]

Likely the println("sending") takes enough time (it’s actual IO) that it gives your original thread a chance to process the event loop and start the collecting coroutine.

[Zach Klippenstein (he/him) [MOD]]

Another way to make this deterministic is to tell your first launch to run immediately until its first suspension point: launch(start = CoroutineStart.UNDISPATCHED) {

[Gilles Barbier]

very interesting - you are right, it's consistent with the observation that it's sensitive to replay value. I'm going to check start = CoroutineStart.UNDISPATCHED that I did not know. Thanks a lot