I have a question about

Flow

with and without a buffer, in this example:

val flow = flow {
    for (i in 1..100) {
        emit(1)
        delay(200)
    }
}.flowOn(Dispatchers.Main) // try replacing it with Dispatchers.IO... 

MainScope().launch {
    flow.collect { 
        println(it)
        delay(1000)
    }
}

This prints out an item each 1200 milliseconds or so. This makes sense. When using

buffer()

, I can speed up the collecting a bit and print out an item each 1000 milliseconds or so. This also make sense to me. However, just replacing

.flowOn(Dispatchers.Main)

with

.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

, the print-outs now happen each 1000 milliseconds as well…. I expected the frequency to still be 1200 milliseconds. Is this to be expected, that even without a buffer (with a size > 0), the

flow

now acts as if it has a buffer with using different dispatchers like this?

I think

flowOn

is effectively a no-op when you pass the same dispatcher that you’re collecting on. If that’s true, this flow would be entirely synchronous which is why you’re seeing 1200ms. If you use a different dispatcher, then the

flow { … }

body is being executed in a different coroutine from the collection, so the delays will happen concurrently. The same thing happens when you add a buffer – the original

flow

body is executed in a different coroutine.

I see. The core of your answer is “the delays happen concurrently” Still, it is a different in behavior, solely based on the dispatchers used. When using the same dispatcher, why can’t the delays happen simultaneous too? That would keep the behavior identical. It seems a short-cut is take when the dispatchers are the same the the

emit

winds up as a ‘simple’ call to the collectors lambda.

because when the dispatchers are the same,

emit

is a synchronous call to your

collect

body, so the

delay(200)

can’t happen until it returns (after

delay(1000)

)

I know all about using channels or a buffer to ‘fix’ this. And I understand how it may happen. But why is it implemented this way? Why are the behaviors not the same, regardless of the dispatchers used? It is possible to make the behaviors the same. If there’s no buffer, have

emit

suspend until the collector’s lambda finishes.

Oh, you want the 1200 delay even if buffers/different dispatchers are used?

I expected the frequency of the print-outs to be 1200 milliseconds all the time, regardless of the Dispatchers used if there is no buffer (or Channel with a buffer) The Flow’s

emit

suspends and resumes only when the collector’s lambda finishes. If buffers are used, the delay should be 1000. But the difference in dispatchers shouldn’t matter, and the delays should be 1200…

The whole point of the

flowOn

operator is to allow you to run the upstream flow concurrently with the downstream flow though. Like if you replace your

delay

with

Thread.sleep

(or some long-running IO operation), you don’t want that blocking the collecting coroutine.

Not a real use case 🙂 Just curiosity From Roman’s post about this, (when using Flow and Flow with a buffer), i understood that without a buffer, the

emit

suspends until the collector’s lambda/action returns/finishes. https://medium.com/@elizarov/kotlin-flows-and-coroutines-256260fb3bdb With a buffer, the

emit

suspends only if the buffer is full. I didn’t expect the difference in dispatchers to make a difference in the suspending behavior of the

emit

. Much like a regular

suspend fun

, the function returns when the final result is returned, no matter with what context that function was called and no matter with what context the function is implemented. Calling the a

suspend fun

is like doing an

emit

, the implementation of that

suspend fun

is like doing a collect-action (but for a Flow, you can do it multiple times instead of just once 🙂).

I think it should. Synchronous execution is the default for Flow, but almost more as an optimization than anything, and only for the simplest case when there is no concurrency. Coroutine code likes being asynchronous, so as soon as anything in the flow is asynchronous, it uses multiple coroutines to do stuff. In general, more concurrency is better - I think that's one of the main premises behind the coroutines feature. If you have logic that depends on downstream flows being executed synchronously, you probably need explicit synchronization, or maybe a different pattern entirely. Hard to discuss without an actual use case. But in general, if you accept the premise that "more concurrency is better", this behavior makes sense.

My confusion is about the

FlowCollector

, the behavior defined for its

emit

function. I understood it to be defined that the

emit

suspends until the collector’s action returns. This is based on this quote from Roman from his article on Medium:

… because both emitter and collector are parts of a sequential execution here and it alternates between them.

What dispatchers are used should not matter, in my opinion. When introducing a buffer, the intermediate buffer’s collector(’s action) now just puts an item onto its queue and it returns almost immediately. But this still would mean that the original

emit

suspends until the buffer’s collector returns (quickly now, without any additional delay; the collector just put something on a queue without delay). The total delay between print-outs goes down to 1000 (unless the buffer gets full).

I understood it to be defined that the

emit

suspends until the collector’s action returns.

This is true, but “the collector” refers to the immediate/direct collector of the operator, not the final collector of the entire chain.

That could be the case, but I'm not sure (yet). I may put an issue/question in the bug tracker about this and see what the JetBrain folks have to say 😃

Thank you, @elizarov! If I understand it all correctly: Calling

flowOn

basically adds a channel with a buffer of default size. The frequency goes down to 1000 msec. The fact that using

Dispatchers.Main

when calling

flowOn

, in my example, still has a 1200 msec frequency is due to an optimization (short-path when dispatcher is the same).

Yes