Sorry folks, I have another question around suspending function and structured concurrency. As far a...

# coroutines

[Andrea Giuliano]

Sorry folks, I have another question around suspending function and structured concurrency. As far as I understood, suspect modifier will make the function suspendable, which means that the underneath thread can continue doing some work. So far so good. Now say that I have a function like this

suspend fun doSomething() = coroutineScope {
   val channel: Channel<Int> = Channel(capacity)
   val results = aggregator.aggregate(channel)
   channel.close()
   results
}

where aggregator is a class that exposes a suspending function aggregate that uses the channel to listen to messages with channel.receive()

Now, would channel.close() be ever invoked before the aggregator finishes to receive all the messages? Or since I’m not creating a new scope (by using launch or async for example) guarantees that each line of my snippet is executed one after the other finishes? And if that’s the case what’s the purpose of marking a function like aggregate() with a suspend modifier (although it suspends since it will wait for mesages in the channel)?

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

would channel.close() be ever invoked before the aggregator finishes to receive all the messages?

Not if aggregate is doing structured concurrency right and not leaking any coroutines. To which point, calling close (or cancel) on such a channel is unnecessary.

Or since I’m not creating a new scope (by using launch or async for example) guarantees that each line of my snippet is executed one after the other finishes?

Yes. Although, again, if aggregate is being naughty and launching new coroutines that don’t follow scoping rules (e.g. using GlobalScope), then those coroutines could be running after aggregate returns, but if aggregate is just looping over channel.receive(), for instance, then it’s well-behaved and your assumption would be correct.

what’s the purpose of marking a function like aggregate() with a suspend modifier?

You answered your own question: “it suspends since it will wait for mesages in the channel”. Another question however might be, why do suspend modifiers need to be propagated up the call chain if methods are only doing sequential things anyway? And the answer to that is because if you call a method that calls a method that suspends, everything after that innermost suspension point needs to be wrapped up into the continuation (the “callback”) that will be resumed when the suspend resumes. That requires every method up the chain to have the state machine infrastructure that the Kotlin compiler uses to actually implement continuations. And the only way the compiler knows to generate that state machine infra is by the suspend modifier on each of these methods. Another, less technical, answer is that it serves as a documentation hint to people reading this method signature that it (probably) won’t do any blocking work on the current thread, so it’s safe to call, e.g., from somewhere like the Android main thread.

[Andrea Giuliano]

Thanks for the answers 🙂

Not if aggregate is doing structured concurrency right and not leaking any coroutines. To which point, calling close (or cancel) on such a channel is unnecessary.

Does that mean that channels gets closed automatically when there is no reference to it?
Also, when you say “doing structure concurrency right”, imagine that aggregator is part of a library and I don’t really know what it does, so it can do some launch() inside it. I believe in that case channel will be closed earlier in my snippet, is that right? So would you suggest to just not close the channel and leave it to be garbage collected?

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

Does that mean that channels gets closed automatically when there is no reference to it?

No, just that there’s no need to close a channel if you know there aren’t any coroutines waiting for it to be closed. A channel isn’t a low-level resource like a file that’s managed by the OS and must always be closed. It’s just a queue, and if nobody’s reading from the queue anymore, it’s fine to just let it be garbage-collected without closing it.

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

imagine that aggregator is part of a library and I don’t really know what it does

If it’s doing something bad, I would consider that a bug and file a bug report to the library. If the library really wants to, it can run as many coroutines as it wants that you have no control over, whether or not you close the channel.

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

The only time I would expect to see a channe.close() there would probably be with a comment explaining that it’s working around some specific bug in the library, with a link to the bug report 😂 (or some other comment explaining why it’s actually necessary)

[Andrea Giuliano]

ok sorry, I’m a bit confused 🙂 you say that it’s not correct that channels gets automatically closed, but then to not close the channel explicitly, does it mean that they get closed when garbage collected?

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

Channels don’t get automatically closed (unless you’re using something like the produce coroutine builder).

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

I’m saying there’s no need to close channels at all unless you actually expect a coroutine to be waiting for the close message on the channel.

[Andrea Giuliano]

perfect, but if no one is doing send() or receive() or similar actions, then it gets garbage collected, right?

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

“Closing” a channel is effectively just sending a special value to the channel, that gets enqueued like all other sends, which, when it gets dequeued, will tell the consuming coroutine that the channel has been “closed”.

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

Yes

[Andrea Giuliano]

ok I get it, so it’s pretty much just do the work you need to do and leave the GC do its job 🙂

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

exactly

[Andrea Giuliano]

[on the suspending modifier, I think JetBrains guys should consider to copy paste your answer 😄 it’s the best explanation I’ve found so far )

[Andrea Giuliano]

but on that, you got me more curious, because in aggregate() method although I’m doing receive so I thought it would have been nice to have the suspend modifier, the compiler didn’t force me to do so. in that case does it mean that the aggregate function would have been called “blocking” and only the receive would actually suspend. is that correct?

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

doSomething needs the suspend modifier because you call coroutineScope actually, not because of aggregate.

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

Although unless aggregate actually has a CoroutineScope receiver, you don't need to call coroutineScope at all.

[Andrea Giuliano]

aggregate function is something like this

suspend fun aggregate(channel: ReceiveChannel<T>):

it has suspend modifier because I call channel.receive()

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

So you don't need coroutineScope at all.

[Andrea Giuliano]

nice, so I really didn’t understand anything 😄 can I ask why’s that? and in which circumstance I should have use coroutineScope?

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

You really only need coroutineScope if you need to call one of the coroutine builder functions (launch, async, etc) in the current scope.

[julian]

@Andrea Giuliano Checkout @Zach Klippenstein (he/him) [MOD]’s illuminating explanation of coroutineScope in this thread https://kotlinlang.slack.com/archives/C1CFAFJSK/p1598198784089700