ちょっと自分用のメモです。

こんな感じのコード。2つ並行で動かして片方がエラーの時にもう片方も止まる。

class Action() : CoroutineScope {
    val job = Job()
    override val coroutineContext: CoroutineContext get() = Dispatchers.Default + job

    fun zipLoadError() {
        launch(coroutineContext) {
            println("before")
            val result1Deferred = async {
                println("async1 before")
                val s = fetchString()
                println("async1 after")
                s
            }
            val result2Deferred = async {
                println("async2 before")
                fetchStringWithException()
                println("async2 after")
            }
            println("after1 ${result1Deferred.await()}")
            println("after2 ${result2Deferred.await()}")
        }
    }

    private suspend fun fetchString(): String = suspendCancellableCoroutine {
        thread {
            Thread.sleep(10000)
            it.resume("result1")
        }
    }

    private suspend fun fetchStringWithException(): String = suspendCancellableCoroutine {
        thread {
            it.resumeWithException(RuntimeException())
        }
    }

出力結果はこうなる。片方が死ぬのでもう片方も止まる。

***two async error
before
async1 before
async2 before
Exception in thread "DefaultDispatcher-worker-2 @coroutine#2" java.lang.RuntimeException
	at exception.Action$fetchStringWithException$2$1.invoke(Exception.kt:156)
	at exception.Action$fetchStringWithException$2$1.invoke(Exception.kt:23)
	at kotlin.concurrent.ThreadsKt$thread$thread$1.run(Thread.kt:30)

内部的になぜこうなるのかを知りたかったです。
コードを読んでいると、Coroutineの状態( JobSupport#_state)としてエラーなどを伝える先を保持しており、エラーが起きた時にその先に知らせることで、伝搬していることがわかりました。
内部のCoroutineの状態の変化をデバッグで出力してみます。
■で始まる行が変更されるCoroutineで、->の前が変更前の状態、->の後が変更後の状態になる。

Connected to the target VM, address: '127.0.0.1:60380', transport: 'socket'
■ JobImpl{Active}@5700d6b1
Empty{Active}
->
ChildHandle["coroutine#1":StandaloneCoroutine{Active}@1e397ed7]
before
■ "coroutine#1":StandaloneCoroutine{Active}@1e397ed7:
Empty{Active}
->
ChildHandle["coroutine#2":DeferredCoroutine{Active}@343253ac]
async1 before
■"coroutine#1":StandaloneCoroutine{Active}@1e397ed7:
ChildHandle["coroutine#2":DeferredCoroutine{Active}@343253ac]
->
List{Active}[ChildHandle["coroutine#2":DeferredCoroutine{Active}@343253ac]]
■"coroutine#2":DeferredCoroutine{Active}@343253ac:
Empty{Active}
->
ChildContinuation[CancellableContinuation(DispatchedContinuation[DefaultDispatcher, exception.Action$zipLoadError$1$result1Deferred$1@4f56af49]){Active}@6f8a0358]
async2 before
■"coroutine#3":DeferredCoroutine{Active}@3c294eef:
Empty{Active}
->
ChildContinuation[CancellableContinuation(DispatchedContinuation[DefaultDispatcher, exception.Action$zipLoadError$1$result2Deferred$1@4c75664f]){Active}@4364feea]
■ "coroutine#2":DeferredCoroutine{Active}@343253ac:
ChildContinuation[CancellableContinuation(DispatchedContinuation[DefaultDispatcher, exception.Action$zipLoadError$1$result1Deferred$1@4f56af49]){Active}@6f8a0358]
->
List{Active}[ChildContinuation[CancellableContinuation(DispatchedContinuation[DefaultDispatcher, exception.Action$zipLoadError$1$result1Deferred$1@4f56af49]){Active}@6f8a0358]]
■ "coroutine#3":DeferredCoroutine{Active}@3c294eef:
ChildContinuation[CancellableContinuation(DispatchedContinuation[DefaultDispatcher, exception.Action$zipLoadError$1$result2Deferred$1@4c75664f]){CompletedExceptionally}@4364feea]
->
Empty{Active}
■ "coroutine#3":DeferredCoroutine{Active}@3c294eef:
Empty{Active}
->
Finishing[cancelling=false, completing=true, rootCause=null, exceptions=null, list=List{Active}[]]
■ "coroutine#1":StandaloneCoroutine{Active}@1e397ed7:
List{Active}[ChildHandle["coroutine#2":DeferredCoroutine{Active}@343253ac], ChildHandle["coroutine#3":DeferredCoroutine{Cancelling}@3c294eef], ChildContinuation[AwaitContinuation(DispatchedContinuation[DefaultDispatcher, kotlinx.coroutines.experimental.DeferredCoroutine$await$1@4685392e]){Active}@243e0d69]]
->
Finishing[cancelling=true, completing=false, rootCause=java.lang.RuntimeException, exceptions=null, list=List{Active}[ChildHandle["coroutine#2":DeferredCoroutine{Active}@343253ac], ChildHandle["coroutine#3":DeferredCoroutine{Cancelling}@3c294eef], ChildContinuation[AwaitContinuation(DispatchedContinuation[DefaultDispatcher, kotlinx.coroutines.experimental.DeferredCoroutine$await$1@4685392e]){Active}@243e0d69]]]
■ "coroutine#2":DeferredCoroutine{Active}@343253ac:
List{Active}[ChildContinuation[CancellableContinuation(DispatchedContinuation[DefaultDispatcher, exception.Action$zipLoadError$1$result1Deferred$1@4f56af49]){Active}@6f8a0358], ResumeAwaitOnCompletion[AwaitContinuation(DispatchedContinuation[DefaultDispatcher, kotlinx.coroutines.experimental.DeferredCoroutine$await$1@4685392e]){Active}@243e0d69]]
->
Finishing[cancelling=true, completing=false, rootCause=kotlinx.coroutines.experimental.JobCancellationException: Parent job is Cancelling; job="coroutine#1":StandaloneCoroutine{Cancelling}@1e397ed7, exceptions=null, list=List{Active}[ChildContinuation[CancellableContinuation(DispatchedContinuation[DefaultDispatcher, exception.Action$zipLoadError$1$result1Deferred$1@4f56af49]){Active}@6f8a0358], ResumeAwaitOnCompletion[AwaitContinuation(DispatchedContinuation[DefaultDispatcher, kotlinx.coroutines.experimental.DeferredCoroutine$await$1@4685392e]){Active}@243e0d69]]]
■ JobImpl{Active}@5700d6b1:
ChildHandle["coroutine#1":StandaloneCoroutine{Cancelling}@1e397ed7]
->
List{Active}[ChildHandle["coroutine#1":StandaloneCoroutine{Cancelling}@1e397ed7]]
■ JobImpl{Active}@5700d6b1:
List{Active}[ChildHandle["coroutine#1":StandaloneCoroutine{Cancelling}@1e397ed7]]
->
Finishing[cancelling=false, completing=true, rootCause=null, exceptions=null, list=List{Active}[ChildHandle["coroutine#1":StandaloneCoroutine{Cancelling}@1e397ed7]]]
■ "coroutine#3":DeferredCoroutine{Cancelling}@3c294eef:
Finishing[cancelling=true, completing=true, rootCause=java.lang.RuntimeException, exceptions=SEALED, list=List{Active}[]]
->
CompletedExceptionally[java.lang.RuntimeException]
■ "coroutine#2":DeferredCoroutine{Cancelling}@343253ac:
Finishing[cancelling=true, completing=true, rootCause=kotlinx.coroutines.experimental.JobCancellationException: Parent job is Cancelling; job="coroutine#1":StandaloneCoroutine{Cancelling}@1e397ed7, exceptions=SEALED, list=List{Active}[ResumeAwaitOnCompletion[AwaitContinuation(DispatchedContinuation[DefaultDispatcher, kotlinx.coroutines.experimental.DeferredCoroutine$await$1@4685392e]){Cancelled}@243e0d69], ChildCompletion[ChildHandle["coroutine#2":DeferredCoroutine{Cancelling}@343253ac], CompletedExceptionally[kotlinx.coroutines.experimental.JobCancellationException: Parent job is Cancelling; job="coroutine#1":StandaloneCoroutine{Cancelling}@1e397ed7]]]]
->
CompletedExceptionally[kotlinx.coroutines.experimental.JobCancellationException: Parent job is Cancelling; job="coroutine#1":StandaloneCoroutine{Cancelling}@1e397ed7]
Exception in thread "DefaultDispatcher-worker-2 @coroutine#2" java.lang.RuntimeException
	at exception.Action$fetchStringWithException$2$1.invoke(Exception.kt:156)
	at exception.Action$fetchStringWithException$2$1.invoke(Exception.kt:23)
	at kotlin.concurrent.ThreadsKt$thread$thread$1.run(Thread.kt:30)
■ "coroutine#1":StandaloneCoroutine{Cancelling}@1e397ed7:
Finishing[cancelling=true, completing=true, rootCause=java.lang.RuntimeException, exceptions=SEALED, list=List{Active}[ChildCompletion[ChildHandle["coroutine#1":StandaloneCoroutine{Cancelling}@1e397ed7], CompletedExceptionally[java.lang.RuntimeException]]]]
->
CompletedExceptionally[java.lang.RuntimeException]
■ JobImpl{Cancelling}@5700d6b1:
Finishing[cancelling=true, completing=true, rootCause=java.lang.RuntimeException, exceptions=SEALED, list=List{Active}[]]
->
CompletedExceptionally[java.lang.RuntimeException]

長い
これを読み解いていきます。

最初の状態

まず最初は全てEmpty{Active}になっている。
これを図にするとこうなる。

このように木が作られる。
コードで言うと以下のようになる。

class Action() : CoroutineScope {
    val job = Job() // ← ****ここがJob****
    override val coroutineContext: CoroutineContext get() = Dispatchers.Default + job

    fun zipLoadError() {
        launch(coroutineContext) {  // ← ****ここがcoroutine#1****
            println("before")
            val result1Deferred = async {  // ← ****ここがcoroutine#2****
                println("async1 before")
                val s = fetchString()
                println("async1 after")
                s
            }
            val result2Deferred = async {  // ← ****ここがcoroutine#3****
                println("async2 before")
                fetchStringWithException()
                println("async2 after")
            }
            println("after1 ${result1Deferred.await()}")
            println("after2 ${result2Deferred.await()}")
        }
    }

JobSupport.ktにある図でいうとここになっている。

Exceptionが起きる直前

これがasync2 beforeの直前のExceptionが起きる直前はこうなる。
この状態のところに親や子に伝えるリスナーみたいなのが入っている。ChildContinuationが親にExceptionなどを知らせ、ChildHandleが子にExceptionなどを知らせるために使われる。

JobSupport.ktにある図でいうとここになっている。(たぶん)

Exceptionが起きた直後

終了する前にExceptionが起きた直後はまずFinishingになる。

まずcoroutine#3、つまりExceptionが起きるcoroutinesがまずFinishingになり、
その次にcoroutine#1、つまりlaunchがFinishingになる。その次にcoroutine#2がFinishingになる。
その3つが渡るとこうなります。

Exceptionが起きた後

Exceptionが起きた後は以下のようにCompletedExceptionally状態になります。

コードリーディング

完全には理解できていないが、DispatchedTaskのここが中断の再開などのタイミングで呼ばれる

DispatchedTask.kt

public override fun run() {
        try {
            val delegate = delegate as DispatchedContinuation<T>
            val continuation = delegate.continuation
            val context = continuation.context
            val job = if (resumeMode.isCancellableMode) context[Job] else null
            val state = takeState() // NOTE: Must take state in any case, even if cancelled
            withCoroutineContext(context) {
                if (job != null && !job.isActive)
                    continuation.resumeWithException(job.getCancellationException())
                else {
                    val exception = getExceptionalResult(state)
                    if (exception != null)
                        continuation.resumeWithException(exception)
                    else
                        continuation.resume(getSuccessfulResult(state))
                }

ここから入っていく

                    if (exception != null)
                        continuation.resumeWithException(exception)

thisは渡しているasync()で渡しているFunction。exceptionにはRuntimeExceptionが入っている。
AbstractCoroutine.kt

CoroutineImpl.kt

    override fun resumeWithException(exception: Throwable) {
        processBareContinuationResume(completion!!) {
            doResume(null, exception)
        }
    }

processBareContinuationResumeはこんなふうになっている。以下で出てくるblockはdoResume(null, exception)を呼んでいるブロック。

@kotlin.internal.InlineOnly
internal inline fun processBareContinuationResume(completion: Continuation<*>, block: () -> Any?) {
    try {
        val result = block()
        if (result !== COROUTINE_SUSPENDED) {
            @Suppress("UNCHECKED_CAST")
            (completion as Continuation<Any?>).resume(result)
        }
    } catch (t: Throwable) {
        completion.resumeWithException(t)
    }
}

仕組みはわからないが、おそらくこのdoResume(null, exception)を呼び出すと中断している場所でExceptionが起こる。そしてそれをまた自分でcatchして、async()で渡しているFunctionがもっているメンバ変数のcompletion(実態は"DeferredCoroutine")のresumeWithException()を呼び出す

(this = DeferredCoroutine)

AbstractCoroutine.kt

    public final override fun resumeWithException(exception: Throwable) {
        makeCompletingOnce(CompletedExceptionally(exception), defaultResumeMode)
    }

(this = DeferredCoroutine)

JobSupport.kt

    internal fun makeCompletingOnce(proposedUpdate: Any?, mode: Int): Boolean = loopOnState { state ->
        when (tryMakeCompleting(state, proposedUpdate, mode)) {
            COMPLETING_ALREADY_COMPLETING -> throw IllegalStateException("Job $this is already complete or completing, " +
                "but is being completed with $proposedUpdate", proposedUpdate.exceptionOrNull)
            COMPLETING_COMPLETED -> return true
            COMPLETING_WAITING_CHILDREN -> return false
            COMPLETING_RETRY -> return@loopOnState
            else -> error("unexpected result")
        }
    }

completeしているかどうかのチェックなどがあり、最終的にtryFinalizeFinishingState()が呼ばれる。
(this = DeferredCoroutine)

JobSupport.kt

    private fun tryMakeCompleting(state: Any?, proposedUpdate: Any?, mode: Int): Int {
....
        val child = firstChild(state)
        if (child != null && tryWaitForChild(finishing, child, proposedUpdate))
            return COMPLETING_WAITING_CHILDREN
        // otherwise -- we have not children left (all were already cancelled?)
        if (tryFinalizeFinishingState(finishing, proposedUpdate, mode)) ←ここにはいっていく
            return COMPLETING_COMPLETED
        // otherwise retry
        return COMPLETING_RETRY

JobSupportのドキュメントに以下のように書いてある。メインの処理は3つあるみたい。

       ## COMPLETE: state !is Incomplete (CompletedExceptionally | Cancelled)
        ------ completion listeners are not admitted anymore, invokeOnCompletion returns NonDisposableHandle
         + parentHandle.dispose
         + notifyCompletion (invoke all completion listeners)
         + onCompletionInternal / onCompleted / onCompletedExceptionally

state.list?notifyCompletionを呼び出す。その処理内容はstateが持っているlistに全てinvokeを呼び出す。

(this = DeferredCoroutine)

JobSupport.kt

    private fun completeStateFinalization(state: Incomplete, update: Any?, mode: Int, suppressed: Boolean) {
...
            state.list?.notifyCompletion(cause)
...

ここで面白いクラスが登場します。親によって使われる、子供のキャンセルを待つクラスです。このinvokeが呼び出されます。そしてこのparentはStandaloneCoroutineです。

JobSupport.kt

    // Used by parent that is waiting for child completion
    private class ChildCompletion(
        private val parent: JobSupport,
        private val state: Finishing,
        private val child: ChildHandleNode,
        private val proposedUpdate: Any?
    ) : JobNode<Job>(child.childJob) {
        override fun invoke(cause: Throwable?) {
            parent.continueCompleting(state, child, proposedUpdate)
        }
        override fun toString(): String =
            "ChildCompletion[$child, $proposedUpdate]"
    }

StandaloneCoroutineのcontinueCompleting
this=StandaloneCoroutine

JobSupport.kt

    private fun continueCompleting(state: Finishing, lastChild: ChildHandleNode, proposedUpdate: Any?) {
        require(this.state === state) // consistency check -- it cannot change while we are waiting for children
        // figure out if we need to wait for next child
        val waitChild = lastChild.nextChild()
        // try wait for next child
        if (waitChild != null && tryWaitForChild(state, waitChild, proposedUpdate)) return // waiting for next child
        // no more children to wait -- try update state
        if (tryFinalizeFinishingState(state, proposedUpdate, MODE_ATOMIC_DEFAULT)) return
    }

ここでStandaloneCoroutineに対してtryFinalizeFinishingState()がまた呼び出され、最終的に終了状態になるみたいです。こんな感じで伝わっていきます。