This may be interesting to those who started worki...

# coroutines

[Dmytro Danylyk]

This may be interesting to those who started working with coroutines: https://twitter.com/dmytrodanylyk/status/1034357859293622272

We have been discussing with @nemi today, maybe custom coroutine dispatcher would be a way to handle such use-case?

[gildor]

What kind use-case?
delay is non blocking operation and it obviously do not block thread

[gildor]

I agree, this is misunderstanding how dispatcher and coroutine works

[gildor]

Mutex is also probably bad choice

[gildor]

Actually it really depends on use case, because this synthetic example doesn’t have enough context

[Dmytro Danylyk]

Use-case:
I want to launch several coroutines and I want to make sure they will be executed in order.

[gildor]

Why?

[gildor]

if operation is non-blocking just run 1 coroutine

[Vsevolod Tolstopyatov [JB]]

Use-case:

I want to launch several coroutines and I want to make sure they will be executed in order.

This is what actor does 🙂
https://github.com/Kotlin/kotlinx.coroutines/pull/485

[gildor]

Yes, Actor mentioned in Dmytro’s tweet, but also Mutex

[gildor]

and I think use Mutex is usually bad advice and there are better solutions

[gildor]

But I think it’s good to mention that dispather and coroutins is not like thread pool for blocking operations.

[Dmytro Danylyk]

Let’s say (just example) I have Register request which may be executed in several places in the app, I want to make sure Register will happen only once.

[gildor]

I just think that would be good to have some real example of code, it allows to propose proper solution

[gildor]

Because not clear what delay does in this example. If you replace delay with sleep to emulate long blocking job, everything will work as ThreadPool with blocking operations

[gildor]

Yeah, Actor is probably the best approach in such use case

[Dmytro Danylyk]

Another example: I start multiple coroutines which have suspend functions with their own dispatcher. In this case I can’t guarantee that my coroutines are executed sequentially.

[gildor]

Probably make sense to add some explanation how dispatchers work in case of blocking and non-blocking operations to avoid misunderstanding.

[gildor]

Why do you need multiple coroutines in this case?

[gildor]

Multiple coroutines means that you want to run something in parallel

[Dmytro Danylyk]

Well it’s the same coroutine but it can be launched from different places in parallel.

[gildor]

I mean if you want to run something sequentially, just wrap all those operations to coroutine and use suspend function or join/await, this is exactly the main use case of coroutines: write non-blocking async code that looks like sync

[gildor]

Oh, I got what you mean

[gildor]

Use case still not completely clear, but I can imagine such code, so yeah, actor is also solution

[dekans]

+1 for actor, but you also have to use job.join() for some dispatched work. Otherway, you'll still have works launched in //

[nemi]

I used actors for solving a case when I wanted the behaviour outlined mentioned by @Dmytro Danylyk. I was sending suspend functions to the actor which invoked them in the order the were received. Solving the particular problem with a custom dispatcher might pose some problems. Let’s say the dispatcher executes the coroutines sequentially without interleaving them. What happens when you dispatch a new piece of work from within the dispatcher ? Should it execute execute it immediately or in a deferred fashion. If it is executed deferred calling wait/join will result in a deadlock as due to the serial execution it will never be actually executed since we are running things in a serial manner and the current one will never finish. If you run it immediately we might not have a dead lock but then we are not running things in a serial fashion and we are effectively running things interleaved. So the best solution depends on the use case.

For my particular use case I wanted certain “top level” coroutines to be interleaved, but did not want to prevent the “top level” coroutines themselves to execute things internally in a parallel fashion.

[bdawg.io]

The part that I’m confused with is that you’re trying to dispatch 2 jobs that are not connected in any way other than you need the first one to be finished before the second one… I think that launching two independent jobs might just be a bad design unless the second job knows to wait for the first job to finish…

val processOne = launch(dispatcher) {
    println("first start")
    delay(2000)
    println("first end")
}

launch(dispatcher) {
     //this process depends on the first being complete
    processOne.join()
    println("second start")
    delay(100)
    println('second end")
}

OR without the 2nd job directly knowing about the first

val processOne = launch(dispatcher) {
    println("first start")
    delay(2000)
    println("first end")
}

// process two cannot start until process one is finished
val processTwo = launch(dispatcher, start = CoroutineStart.LAZY) {
    println("second start")
    delay(100)
    println('second end")
}

processOne.join()
processTwo.join()

OR using an actor to manage the joins

val jobManager = actor<Job>(capacity = Channel.UNLIMITED) {
    channel.consumeEach { it.join() }
}

launch(dispatcher, start = CoroutineStart.LAZY) {
    println("first start")
    delay(2000)
    println("first end")
}.also { jobManager.send(it) }

launch(dispatcher, start = CoroutineStart.LAZY) {
    println("second start")
    delay(100)
    println('second end")
}.also { jobManager.send(it) }

[nemi]

The thing that I was explaining to @Dmytro Danylyk was that if you wanted to execute to tasks sequentially with kotlin coroutines, a non experienced person might consider simply using the single threaded context.
Which on it’s own will not guarantee serial execution as you they might expect it.
If you were to write the following

val runnable1 = Runnable {
    println("launching first")
    Thread.sleep(1000)
    println("finishing first")
}

val runnable2 =  Runnable {
    println("second")
    Thread.sleep(250)
    println("second finished")
}

with(Executors.newSingleThreadExecutor()) {
    submit(runnable1)
    submit(runnable2)
    shutdown()
    awaitTermination(2000, TimeUnit.MILLISECONDS)
}

The two runnables would be executed sequentially, and the two executions would not be interleaved with one another. The output would be:

“launching first”
“finishing first”
“launching second”
“finishing second”

However if you were to rewrite this naively with coroutines

val dispathcher = newSingleThreadContext("foo")

val job1 = launch(context = dispathcher) {
    println("launching first")
    delay(1000)
    println("finishing first")
}

val job2 = launch(context = dispathcher) {
    println("launching second")
    delay(250)
    println("finishing second")
}

job1.join()
job2.join()

You’d be in for a surprise, as the output will be:

“launching first”
“launching second”
“finishing second”
“finishing first”

And that’s because kotlin coroutines don’t quite behave as threads. The two coroutines are dispatched on a single thread, but interleaved due to the suspension points.
One of the power of coroutines is that it is utilizing threads much more efficiently (done properly). You could executed thousands of jobs “concurrently” without exhausting resources.

What you are mentioning about bad design is a valid point. You might need to have proper external coordination or one job needs to know that that it has to wait for another to finish in order to execute.
How you implement that depends on your use case and requirements.

However the question arose if this can be done with a dispatcher in a way that the “top level” coroutines dispatched are executed sequentially not interleaved, but the coroutine internally is allowed to do things concurrently.

Using this approach is prone to a lot of errors and may most probably be bad design as you are not managing state at all.

[bdawg.io]

@nemi I agree, if you don’t know about the start parameter and the available start = CoroutineStart.LAZY that I used above, you would indeed be scratching your head how to perform such a task. Perhaps an option to my actor example could be to provide an extension function to manage that for you

fun SendChannel<Job>.launchNext(context: CoroutineContext, block: suspend CoroutineScope.() -> Unit) = launch(Unconfined) {
    send(launch(context, start = CoroutineStart.LAZY, block = block, parent = coroutineContext[Job]))
}

val jobManager = actor<Job>(capacity = Channel.UNLIMITED) {
    channel.consumeEach { it.join() }
}

val dispatcher = newSingleThreadContext("foo")
jobManager.launchNext(dispatcher) {
    println("first start")
    delay(2000)
    println("first end")
}
jobManager.launchNext(dispatcher) {
    println("second start")
    delay(100)
    println("second end")
}