If I have a class that implements CoroutineContext, how would I go about to perform a graceful shutd...

# coroutines

[gotoOla]

If I have a class that implements CoroutineContext, how would I go about to perform a graceful shutdown?
A graceful shutdown in this case meaning wait for any ongoing child to finish (so joinAll() on any child-coroutine spawned)

It seems CouroutineConext.cancel() is quite aggressive and does cancel instead of join (which the name kinda indicates as well :P)

[zak.taccardi]

I’ve never extended CoroutineContext - why are you doing it?

[diesieben07]

I think you may be confusing CoroutineScope with CoroutineContext? A CoroutineScope has "children" and has "wait for children" built-in. CoroutineContext can contain arbitrary elements, such as Job.

[diesieben07]

In any case, it sounds like you want a Job or maybe a SupervisorJob for your "container" which you can then wait for

[Adam Powell]

or structure the code a bit differently as a

coroutineScope {
  // launch/do work here
}
// all children joined by here

[gotoOla]

🤦‍♂️ you are correct in that I meant CoroutineScope...

[diesieben07]

As for CoroutineScope it (usually) has an associated Job in it's associated CoroutineContext (see docs on CoroutineScope.coroutineContext). So:

checkNotNull(myScope.coroutineContext[Job]) { "CoroutineScope is missing Job" }.join()

[gotoOla]

so if you use the coroutineScope { } everything that is launched in that block is finished to assure that you don't leak any coroutines. I kinda want that behavior on my class that implements CouroutineScope.

In my class I have a SendChannel setup where one channel is fetching messages from a queue (so when I receive messages from the queue I perform a channel.send()). The SendChannel is producing while(isRunning) { ... } . To this I connect X number of "consumers" (ReceiveChannels) that processes whatever the fetching one produces (so I have X ongoing coroutines)

I was thinking that I would like to expose a stop function in this class, e.g.

fun stop() {
    isRunning = false
    /* perform the "stop action on coroutinescope
       so that the consumers that are active finishes /*
}

[gotoOla]

ah ok @diesieben07 so that job would be my "supervisor-job" and if I join on that one all children should be joined as well?

[diesieben07]

Yes, exactly.

[diesieben07]

If you implement CoroutineScope manually you might have to create your Job manually as well, depending on your usage.

[diesieben07]

Note that SupervisorJob is actually a factory function which produces a special kind of parent-job (see the docs for details).

[gotoOla]

today all I've done is actually

class MyClass: CoroutineScope by CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>)

[gotoOla]

if I have to do it manually I don't really see the benefit or the "scope" of it here. What I do today is

processors.addAll(
    (1..maxNumberOfMessagesInParallel).map {
        <http://logger.info|logger.info>("Starting processing channel.")
        launchMessageProcessor()
    }
)

with processor as a class field and then in my stop function I do

fun stop() {
    isRunning = false
    processors.joinAll()
}

[diesieben07]

That's fine, the CoroutineScope factory function automatically introduces a Job into the CoroutineContext if one is not already present

[gotoOla]

hm if I debug and check what I have in the this.coroutineContext it doesn't seem to hold a reference to any job

[diesieben07]

Well, this is what that function does:

ContextScope(if (context[Job] != null) context else context + Job())

[diesieben07]

So it should hold a Job.

[gotoOla]

ah right, looking into the source now as well. I think I used the wrong import. I saw that there was an object Job with a companion object Key

[dekans]

Maybe something like that:

coroutineContext[Job]?.children?.forEach { it.cancelAndJoin() }

Right before cancelling the scope

[dekans]

with some parallelization to not lose time