Coroutines execution order when launching them sequentially

Language Design
1

Can we make any assumptions on coroutines execution order when launching them sequentially with launch()/async()? I immediately stress that I’m not looking for “sequential asynchronous” execution, I don’t need strong guarantees about the execution order and I understand that launch()/async() is by design not sequential. I just wonder if we can assume that in some simple cases like launching coroutines with the same dispatcher, from a single coroutine in a simple loop, the execution order will be at least approximately FIFO.

Let’s get this example:

flow {
    (0 until 100)
        .map { async { process(it) } }
        .forEach { emit(it.await()) }
}.flowOn(Dispatchers.Default)

We have 100 items to process, processing is CPU intensive, so we do it concurrently. We would like to collect results as they become available, but we need to maintain items ordering.

Above code already guarantees proper ordering by creating a list of Deferred and awaiting on them in order. We don’t care that processing order will be e.g.: 22, 23, 26, 25, 24, 27 or 99, 0, 1, 2, 3 - some results will wait just a little and this is it. However, it would be bad if coroutines would launch in totally random order or even worse, in LIFO order. In the latter case, we would need to wait for processing of all 100 items and then all results would be emitted at once. Is there such a risk, e.g. in some future version of Kotlin or on some specific platform?

Also, what are alternatives for solving above problem? I know I can create a Channel, use it to feed coroutines with items and then perform some synchronization voodoo when collecting results to emit them in order. I just feel like my case is really pretty trivial and I’m reinventing the wheel here.

1 Like
2

Hi @broot,
the single thread dispatcher is a FIFO executor, others are almost FIFO.
None of the predefined Dispatcher is random or LIFO and further implementations in this direction can be unpractical.

Take care that dispatching occurs on each suspension, therefore a single coroutine can be enqueued multiple times.

3

Thank you, I think your explanation matches my definition of “approximately FIFO”, so above solution should be sufficient in most cases.

I’m also aware that coroutines need to be rescheduled after suspending, so this is much more complicated than it looks. If dispatchers would always guarantee strict FIFO ordering and we would launch 100 coroutines like this:

Thread.sleep(1000) // CPU-intensive
delay(100)
Thread.sleep(1000) // CPU-intensive

Then as a result we would begin with processing the first half of each coroutine and then we would move to second half - which is probably not optimal. Coroutines/dispatchers are smarter than that and they seem to somehow prioritize resuming suspended tasks.

4

Kotlin’s coroutine library does not prioritize suspended tasks over new ones. Consider 100s of this block, instead:

while(isActive()) {
    Thread.sleep(1000) // CPU-intensive
    yield()
}

Do you really want the already started jobs to always have priority so then the jobs that were started later never even start?

1 Like
5

Hmm… interesting… So what is the reason I observe that e.g. task0 is resumed before starting of task8 even if task8 is launched much earlier that task0 is scheduled to be resumed? I use the following code:

runBlocking(Dispatchers.Default) {
    repeat(100) { i ->
        launch {
            println("$i: starting")
            Thread.sleep(500)
            println("$i: suspending")
            delay(100)
            println("$i: resuming")
            Thread.sleep(500)
            println("$i: ended")
        }
    }
    println("end")
}

What I observe is:

  • First group of coroutines is started (0, 1, 2, 99 on my machine)
  • Finished launching of all coroutines.
  • First group suspends, second group is started
  • Second group suspends, first group is resumed

All coroutines are launch()'ed before the first group even get to the delay() point. Still, the first group is resumed before starting other coroutines. Pattern continues, second group is resumed before the forth one is started, etc.

Could it be caused by the fact I use delay() which, as I imagine, may be scheduled in a little different way?

I think I need to read the source code of both coroutines library and ForkJoinPool one day. Sounds like a very interesting stuff.

6

There’s a lot more to the details of how the shared scheduler behind Dispatchers.Default and Dispatchers.IO works. Each worker has its own queue and there’s global queues too. The IO tasks are treated a bit differently in places. Workers will randomly grab from global queues instead of the local queue and they’ll steal from each other too looking for work to do. In some narrow scenarios it uses LIFO (hence the 99). Frankly, I really don’t understand the details, just that it’s not a naive FIFO queue.

But there is nothing special about launch that I’ve noticed. It builds a Continuation and then calls resume just like any mid-progress coroutine.

You’ll likely see pretty different behavior using Dispatchers.IO or if you use the built-in runBlocking dispatcher.

7

Thank you very much.

I guess the answer to my question is more or less:

  1. There are no guarantees about the execution order.
  2. In some simple cases the ordering might be predictable, so sometimes it might make sense to just assume it is “almost FIFO”, but see 1. It makes sense if different ordering would e.g. slightly degrade the performance, but the application should not misbehave because of that.
  3. Generally, it is a better idea to just control the execution order one way or another (if we need it).