No CancellationException thrown when join on a crashed Job

[bennyhuo]

The job2 will throw an ArithmaticException thus making the parent job cancelled. Normally we will get a CancellationException when calling the job2.join() but there is otherwise.

suspend fun main() = runBlocking {
    log(1)
    val job = launch(Dispatchers.Unconfined) {
        log(2)
        val job2 = launch(Dispatchers.Default) {
            log(3)
            val x = 1 / 0
            log(4)
        }
        job2.join()
        log(5)
    }
    log(6)
    job.join()
    log(7)
}

val dateFormat = SimpleDateFormat("HH:mm:ss:SSS")

val now = {
    dateFormat.format(Date(System.currentTimeMillis()))
}

fun log(msg: Any?) = println("${now()} [${Thread.currentThread().name}] $msg")

The result may be:

09:35:15:065 [main @coroutine#1] 1
09:35:15:074 [main @coroutine#2] 2
09:35:15:082 [DefaultDispatcher-worker-1 @coroutine#3] 3
09:35:15:104 [main @coroutine#2] 5
Exception in thread "main" java.lang.ArithmeticException: / by zero
	at com.bennyhuo.coroutines.sample2.exceptions.E2Kt$main$2$job$1$job2$1.invokeSuspend(e2.kt:12)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(Dispatched.kt:238)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:594)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.access$runSafely(CoroutineScheduler.kt:60)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:742)
09:35:15:105 [main @coroutine#1] 6

When falling into the slow-path, in other words, the joinSuspend path:

public final override suspend fun join() {
    if (!joinInternal()) { // fast-path no wait
        coroutineContext.checkCompletion()
        return // do not suspend
    }
    return joinSuspend() // slow-path wait
}

it is gonna check the job result again here:

public suspend inline fun <T> suspendCancellableCoroutine(
    crossinline block: (CancellableContinuation<T>) -> Unit
): T =
    suspendCoroutineUninterceptedOrReturn { uCont ->
        val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_CANCELLABLE)
        // NOTE: Before version 1.1.0 the following invocation was inlined here, so invocation of this
        // method indicates that the code was compiled by kotlinx.coroutines < 1.1.0
        // cancellable.initCancellability()
        block(cancellable)
        cancellable.getResult() // HERE!!!!!
    }

If the job just completed, the result will be returned.

In our case, job2 completed with a Exception, but the state of the job will turn into Unit for a ResumeOnCompletion is installed when calling joinSuspend and the ResumeOnCompletion will be invoked immediately if the result is ready.

private suspend fun joinSuspend() = suspendCancellableCoroutine<Unit> { cont ->
    // We have to invoke join() handler only on cancellation, on completion we will be resumed regularly without handlers
    cont.disposeOnCancellation(invokeOnCompletion(handler = ResumeOnCompletion(this, cont).asHandler))
}

[bennyhuo]

I think the problem may be the regardless of state when ResumeOnCompletion invoked:

private class ResumeOnCompletion(
    job: Job,
    private val continuation: Continuation<Unit>
) : JobNode<Job>(job)  {
    override fun invoke(cause: Throwable?) = continuation.resume(Unit) // why ignoring the job state?
    override fun toString() = "ResumeOnCompletion[$continuation]"
}

[elizarov]

Good catch. There is indeed a race in the slow path that may result in job.join() returning normally instead of throwing a CancellationException.

[bennyhuo]

#1124 I have been trying to fix this by working on the ResumeOnCompletion, and it seems to be working properly.

private class ResumeOnCompletion(
    job: JobSupport,
    private val continuation: Continuation<Unit>
) : JobNode<JobSupport>(job)  {
    override fun invoke(cause: Throwable?) {
        val state = job.state
        check(state !is Incomplete)
        if (state is CompletedExceptionally) {
            val parentJob = continuation.context[Job]
            // Only if the parent job is cancelled.
            if (parentJob != null && !parentJob.isActive) {
                // Resume with the CancellationException as designed.
                continuation.resumeWithExceptionMode(job.getCancellationException(), MODE_ATOMIC_DEFAULT)
                return
            }
        }
        continuation.resume(Unit)
    }
    override fun toString() = "ResumeOnCompletion[$continuation]"
}

[elizarov]

Unfortunately, the fix in #1124 is not correct. It throws cancellation exception from the job that is being joined, but, instead, the cancellation exception of the job that is joining should be thrown. See #1127 for a fix.

[bennyhuo]

Thanks for your reply!