hey folks, I have what is probably a naive question, while I'm learning about Kotlin's coroutines. I was confused by the behaviour of switching between two different ways of launching a sub-coroutine, and wanted to make sure I understood the consequences properly. My scenario is that I have a long-running coroutine listening to a channel, and acting on it. eg.

for (msg in channel) { ... }

This initial coroutine was created via

CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>).launch { .. }

Before entering that loop, I launch another coroutine, which is regularly send()ing messages into the channel. eg.

while (true) { delay(..); channel.send(..) }

Initially I was creating the second coroutine with

coroutineScope { launch { ... } }

, but later switched to

CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>).launch { .. }

. I switched because in the first version, it felt like this coroutine was able to block the other one! If the _channel capacity was exceeded, this second coroutine starts to block on the channel.send() call, which is expected. However, it also seemed to be blocking my first coroutine too, which I did not expect. Can you help me understand what was happening here? And, also, I'm concerned about whether my second coroutine will still be properly considered a child of the first co-routine, with regard to it being properly cancelled if the parent is cancelled.

coroutineScope { launch { ... } }

is the same as

...

I'm finding this hard to follow without seeing the actual code. Note that with CoroutineScope().launch() you usually need to capture the return value of CoroutineScope() so that you can clean it up if needed. Unlike coroutineScope{launch{}} where structured concurrency will handle most things for you

Jeffrey: Oh! Right.. that's interesting. So it doesn't create a new

Job

that can run in parallel if you

launch

inside your own coro scope?

when

coroutineScope { ... }

(or any other normal

suspend fun

) returns, everything it has launched has also completed. this is the structure in structured concurrency

I had a feeling that was going to be the case (breaking the parent-child relationship). But I'm not quite sure of the right way to get an asynchronous job running, then, without blocking the parent? (Which is perhaps happening due to a side effect of something else..)

coroutineScope {
    launch { one() }
    coroutineScope {
        launch { two() }
        launch { three() }
    }
    launch { four() }
}

one+two+three may run concurrently. one+four may run concurrently. but two+three will never run concurrently with four; they must be complete before control flows to where four is launched. (regardless of whether you split out some parts of this into other functions or not)

That does sound like what I'm seeing.

suspend fun inner(channel: Channel<String>) {
    for (i in 1..10) {
        delay(1000)
        channel.send("inner coro")
    }
    channel.close()
}

runBlocking {
    val channel = Channel<String>(2)
    launch {
        inner(channel)
    }

    for (s in channel) {
        println("Hello $s")
    }
}

doesn’t work?

the immediate way to get rid of

CoroutineScope()

is

suspend fun outer(channel: Channel<String>): Unit = coroutineScope {
    launch(<http://Dispatchers.IO|Dispatchers.IO>) {
        inner(channel)
    }

    for (message in channel) {
        println("Hello ${message}")
    }
}

runBlocking {
    val chan = Channel<String>(2)
    withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
        withTimeout(11000) {
            outer(chan)
        }
    }
}

and everything should run concurrently as expected

oh, for the sake of the example, I'm specifically trying to launch the inner routine from the outer routine. Imagine that from a caller's perspective, it only wants to know about the

outer()

function call.

nothing about

outer

leaks past its lifetime to its caller - such as whether it `launch`es child coroutines

Thanks.. just digesting your changes, ephemient

no

ah, got it! However, that adds a bit of confusion -- because in my example, I didn't think there was anything that was actually hard-blocking like that. as in,

delay()

and

channel.send()

are both suspend functions, yeah?

they are

It's slightly cut down, but it does demonstrate my issue -- that the two loops don't seem to run concurrently, even though both are suspend functions and not hard blocking

maybe it only looked stuck because your program was never terminating (due to the

CoroutineScope(...).launch {}

)?

Maybe, but I thought i'd see i/o output while it was running, even if it never terminated?

might be buffered by whatever's running your program (such as Gradle)

it's unbuffered, if I use the version with Dispatchers.IO

sure. it does seem that everybody comes into this with a different intuition at first, which doesn't always match how kotlinx.coroutines was designed to behave

I know all the ins and outs and oddities of Akka Actors. 😂

from what I know, it's using the same model as Java, so unstructured concurrency - anything goes, anywhere

Scala's Akka's Actors were a bit like coroutines and channels.. All the Actors ran in the same single thread, unless you specifically told them to run in another context. Actors were meant to be non-blocking. Actors would process a message off their incoming queue (like a kotlin channel) and send results out in the same way.

I mean in that an actor passing work to other actors doesn't retain any chain of ownership (afaik)

An Actor can create child actors, and those work like structured concurrency, where if the parent actor exits/dies, the children are automatically cleaned up.

I feel like it's hard to explain this - you can implement actors on coroutines, but the other way around doesn't make much sense

It's OK, I'm more mentioning this in case you were curious about where my biases come from 🙂

Dispatchers.Default

and

<http://Dispatchers.IO|Dispatchers.IO>

actually share threads, so that "switching" from one dispatcher to the other can be optimized

The issue does seem to be that the

launch {}

wants to wait for the routine inside it to complete, yet in my system, that's an infinite loop (until cancelled)

The default dispatcher has threads too. It just usually has much less. A common setup is 4 and 64

Ah.. right.

Launch is the right way as in the previous example

What is the significance of having coroutineScope twice, in the snippet there?

it introduces a child scope (just like

withContext(EmptyCoroutineContext)

), and is the way you are supposed to use to make use of

CoroutineScope.*

extensions inside a

suspend fun

So I think I've worked out what is different about my simple example vs my example code I pasted.. and it's that in the example,

launch { inner() }

is called in outer(). Whereas in my real code, it was the equivalent of:

fun outer {
  launchInnerThing()
  doStuff...
}
fun launchInnerThing() {
  coroutineScope { launch {
    while (true) { // do stuff }
  } }
}

yes, as we said previously,

coroutineScope { launch { ... } }

does not return until the body of the launch is complete

That's why I was checking about the significance of

coroutineScope

occurring twice in your own example.. I hadn't quite made the connection

but you can redefine launchInnerThing as

fun CoroutineScope.launchInnerThing()

and lose the inner coroutineScope builder (and DON’T mark it as suspend!)

you can but I don't think that conveys the intent very well and makes it harder for the caller to actually know what the bounds are

I’d reverse the phrasing of that: in general, just like suspend functions, a non-suspend function should… but otherwise 👍

I agree -- it's not good behaviour for a function to just decide to keep running in the background. My bad for writing it! I am just experimenting around, and hadn't realised the significant of suspend vs non-suspend functions vs coroutine scope.

You might actually be better off with a Flow rather than a Channel.

CoroutineScope(…).launch { }

can lead to silent malfunction if any underlying callback happens to be registered in a WeakReference (something you might not know): https://github.com/Kotlin/kotlinx.coroutines/issues/1061

Thanks for the tip

Sorry Toby I was sleeping before answering your question but I guess you got a good conversation there already. One thing I haven't seen clearly mentioned, and which might be the source of your confusion, was that the following does express concurrency like you want to:

coroutineScope {
    launch {
        doStuff1()
    }
    doStuff2()
}

In this case

doStuff1

and

doStuff2

run concurrently, because

coroutineScope

only waits for the launch at the end of the block. That's how you construct hierarchies of concurrent coroutines. Ephemient had mentioned it but I'm not sure it was clear for you.

I think the root cause of my confusion is the way that the placement of coroutineScope and launch matters, depending on whether it's inside a separate suspend function. (At least, that was one of the things I've taken away from the conversation. And need to experiment with further to truly understand.)

Mmmh it doesn't matter whether you extract

coroutineScope

into a suspend function (I mean it doesn't change the behaviour). It is itself a suspend function. What matters is the placement of code inside or outside the

launch

or the

coroutineScope

block

I think I am, slowly, starting to get it. So, a

coroutineScope { ... }

is essentially a blocker, that won't return until everything inside it has completed -- including if what it contains are

launch {...}

statements. If you want two coroutineScopes to be running in parallel, they need to be inside

launch

statements at a higher scope. ie

runBlocking {
    launch {
      coroutineScope {
        launch { verySlowThing() }
        doAnotherRelatedThing()
      }
    }
    launch {
      coroutineScope {
        launch { oneThing }
        launch { otherRelatedThing() }
      }
   }
}

And in fact, I don't need the

coroutineScope

inside

launch

if I'm only doing one thing in there.

Yes, but make sure there's still a strong reference to this

CoroutineScope(…)

, or the result of the only

launch

result (its

Job

), or you can risk the GC coming for it. And mind how Exception handling and cancellation propagation isn't going to work the same.

Thanks. I hope that's an area I have much more familiarity with -- ie. I'm familiar with managing worker threads, which sound like they have the same issues of error and cancellation management

but also leads me to wonder:

Is there a way to get access to launch without forcing me to create a new coroutineScope?

Yes, as I mentioned earlier, you can make your function an extension function on CoroutineScope and not suspend. Also see https://elizarov.medium.com/coroutine-context-and-scope-c8b255d59055

If you need to launch a coroutine that keeps running after your function returns, then make your function an extension of CoroutineScope or pass scope: CoroutineScope as parameter to make your intent clear in your function signature. Do not make these functions suspending: