Kotlin CoroutinesのStructured Concurrencyを確認する

これは何？

Structured Concurrencyというものがあり、kotlinx.coroutinesはこれをサポートしています。
そもそもStructured Concurrencyはどのようなものなのか、Kotlinではどのような形でサポートしているのかを確認していきます。

Structured Concurrency と Coroutine

wiki

https://en.wikipedia.org/wiki/Structured_concurrency

Structured concurrency is a programming paradigm aimed at improving the clarity, quality, and development time of a computer program by using a structured approach to concurrent programming.

The core concept is the encapsulation of concurrent threads of execution (here encompassing kernel and userland threads and processes) by way of control flow constructs that have clear entry and exit points and that ensure all spawned threads have completed before exit. Such encapsulation allows errors in concurrent threads to be propagated to the control structure's parent scope and managed by the native error handling mechanisms of each particular computer language. It allows control flow to remain readily evident by the structure of the source code despite the presence of concurrency. To be effective, this model must be applied consistently throughout all levels of the program – otherwise concurrent threads may leak out, become orphaned, or fail to have runtime errors correctly propagated.

Structured concurrency is analogous to structured programming, which uses control flow constructs that encapsulate sequential statements and subroutines.

(Google日本語翻訳)
構造化並行処理とは、並行プログラミングに構造化されたアプローチを用いることで、コンピュータプログラムの明瞭性、品質、開発時間を改善することを目的としたプログラミングパラダイムである。

中核となる概念は、明確な開始点と終了点を持ち、生成されたすべてのスレッドが終了前に完了することを保証する制御フロー構造によって、並行実行スレッド（ここではカーネルスレッドとユーザーランドスレッドおよびプロセスを含む）をカプセル化することです。このようなカプセル化により、並行スレッドのエラーは制御構造の親スコープに伝播され、各プログラミング言語固有のエラー処理メカニズムによって処理されます。これにより、並行処理が存在する場合でも、ソースコードの構造から制御フローを容易に把握できます。このモデルを効果的に機能させるには、プログラムのすべてのレベルで一貫して適用する必要があります。そうしないと、並行スレッドが漏洩したり、孤立したり、実行時エラーが正しく伝播されなかったりする可能性があります。

構造化並行処理は、順次実行されるステートメントとサブルーチンをカプセル化する制御フロー構造を用いる構造化プログラミングに類似している。

ここでのカプセル化というのはKolinでは以下になります。

coroutineScope {
    launch { println("A") }
    launch { println("B") }
}
println("Finish")

この処理は必ず"Finish"が最後に出力されることが保証されています。「scopeを抜ける = println("Finished")に辿り着く = scope内の処理の完了」が保証されているためです。通常の非構造な並行処理だと「関数を抜けても、裏でまだ何か動いている」という状態になりますが、構造化された並行処理（Structured Concurrency）の場合、Scopeを正常に抜けることが、そのスコープに属する子タスクは残っていないことを保証します。親Coroutineは、自身の処理が終了しても子Coroutineが残っている間は完了せず、すべての子Coroutineの完了を待ってから完了します。

ここで勘違いしやすいポイントは以下のKotlinのコードです。

viewModelScope.launch {
    launch { println("A") }
    launch { println("B") }
}
println("Finish")

上記コードは"Finish"が最後に来る保証はありません。launchは起動したCoroutineの完了を待たずに、Jobを返して先に進む関数のためです。

もし順序を保証する場合は

val job = viewModelScope.launch {
    launch { println("A") }
    launch { println("B") }
}
job.join()
println("Finish")

joinでCoroutineの完了を待つことにより、println("Finished")が最後に実行されることを保証することができます。

ここで１つ重要なポイントとしてCoroutineの概念とStructured Concurrencyは別の概念であるということです。名前が違うので別の概念なのは当たり前ですが改めて整理すると、

Coroutine ... 停止と再開ができる処理の実行単位（前回の記事で調査済み）
Structured Concurrency ... 複数の並行タスクに親子関係と明確なライフタイムを持たせて、並行処理を安全で理解可能にするための設計原則

ということです。

Kotlin Coroutinesは、Coroutineという中断・再開可能な実行単位を使った非同期・並行処理を提供し、そのCoroutineのライフタイム管理にStructured Concurrencyの原則を取り入れているということです。

これは明確にドキュメントにも記載されています。

Coroutines basics

https://kotlinlang.org/docs/coroutines-basics.html?utm_source=chatgpt.com#coroutine-scope-and-structured-concurrency

Job

Kotlin CoroutineがStructured Concurencyをサポートするために用意したクラス（Job）を見ていきます。

public interface Job : CoroutineContext.Element {
    public companion object Key : CoroutineContext.Key<Job>
    public val isActive: Boolean
    public val isCompleted: Boolean
    public val isCancelled: Boolean
    public fun start(): Boolean
    public fun cancel(cause: CancellationException? = null)
    public suspend fun join()
    public val children: Sequence<Job>
    public fun attachChild(
        child: ChildJob
    ): ChildHandle
    public fun invokeOnCompletion(
        handler: CompletionHandler
    ): DisposableHandle
    public fun invokeOnCompletion(
        onCancelling: Boolean,
        invokeImmediately: Boolean,
        handler: CompletionHandler
    ): DisposableHandle
    public fun getCancellationException(): CancellationException
    override val key: CoroutineContext.Key<*>
        get() = Job
}

https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job/

まずはJobを見ていきます。箇条書きでポイントそうな場所と自分の理解を記載します。

JobとはCoroutineのライフサイクル（isActive/isCompleted/isCancelled）を表現し、また制御（start/cancel/join）するオブジェクト
Jobは子Jobを保持している
Jobは親Jobを知らない（JobSupport経由で確認可能）
親がキャンセルされると、その子も全て再起的にキャンセルされる
CancellationExceptionを子がthrowした場合
- 親はキャンセルされない
- 他の子もキャンセルされない
CancellationException以外を子がthrowした場合
- 親は即座にキャンセルされる
- 結果として他の子もキャンセルされる

ここでよくあるCancellationExceptionを握り潰してはいけない理由が見えてきます。

例えばCancellationException握り潰した場合、ジョブはCoroutineを停止することができず、処理を続けてしまいます。

val job = launch {
    try {
        delay(1.seconds)
    } catch (e: Exception) {
        // 握りつぶす
        e.printStackTrace()
    }
    println("まだ処理する")
}
delay(0.5.seconds)
job.cancel()

上記のコードではlaunchでCoroutineが作成されています。Coroutine内のdelayはCancellationExceptionを投げることができます。

ここでもしjobをdelay中にcancelした場合、そのCoroutineは不要になり、delay自身がCancellationExceptionを投げますが、上記コードでは握り潰しているのでCoroutineが停止せず、println(”まだ処理する”)まで処理が動いてしまいます。

val job = launch {
    try {
        delay(1.seconds)
    } catch (e: CancellationException) {
        throw e
    } catch (e: Exception) {
        e.printStackTrace()
    }
    println("まだ処理する")
}
delay(0.5.seconds)
job.cancel()

そのため上記のようにCancellationExceptionは握り潰さないようにしましょう。

JobSupportと内部状態とJobNode

JobSupportはJobを実装した抽象クラスです。startやcancelといったJobを制御する処理を持ち、またJobの状態をstateというフィールドで管理しています。

stateについて見ていきます。以下のリンクから具体的に状態遷移図を確認することができます。
https://github.com/Kotlin/kotlinx.coroutines/blob/master/kotlinx-coroutines-core/common/src/JobSupport.kt?utm_source=chatgpt.com#L24

10個の内部状態があることが確認できました。

EMPTY_N
EMPTY_A
SINGLE
SINGLE+
LIST_N
LIST_A
COMPLETING
CANCELLING
FINAL_C
FINAL_R

listenerを持たない（EMPTY）、1つ持つ（SINGLE）、複数持つ（LIST）で大別されています。
具体的なクラスとしてはEmpty(EMPTY_N, EMPTY_A)・JobNode(SINGLE, SINGLE+)・NodeList(LIST_N, LIST_A)・Finishing(COMPLETING, CANCELLING)というクラスが存在します。これらのクラスはImcompleteを実装しており未完了を表す状態です。完了を表すクラスはなくstate !is Incimpleteで表されます。stateはAny型で管理されており、正常完了時は具体的な結果の値になります。launchで作成されるJobは結果を必要としないため完了時はstateがUnitになります。

private class Empty(override val isActive: Boolean) : Incomplete
 
internal abstract class JobNode : LockFreeLinkedListNode(), DisposableHandle, Incomplete
 
internal class NodeList : LockFreeLinkedListHead(), Incomplete
 
private class Finishing(
    override val list: NodeList,
    isCompleting: Boolean,
    rootCause: Throwable?
) : SynchronizedObject(), Incomplete

JobNodeはLockFreeLinkedListNodeを、NodeListはLockFreeLinkedListHeadを実装しています。そのためNodeListでは複数のJobNodeを連結リストで管理していることがわかります。

次に、JobNodeとはなんでしょうか？

JobNodeはJobの状態変化に反応する１個の登録ノードの規定クラスであることがわかります。具体的にJobNodeの具象クラス（InvokeCancelling）を見ていきます

private class InvokeOnCancelling(
    private val handler: CompletionHandler
) : JobNode()  {
    // delegate handler shall be invoked at most once, so here is an additional flag
    private val _invoked = atomic(false)
    override val onCancelling get() = true
    override fun invoke(cause: Throwable?) {
        if (_invoked.compareAndSet(expect = false, update = true)) handler.invoke(cause)
    }
}

InvokeOnCancellingのJobNodeはjob#invokeOnCompletionで登録できます。

job.invokeOnCompletion {
    println("job is completed")
}

このようにするとJobの完了時に任意のメソッドを実行することができます。invokeOnCompletionはInvokeOnCancellingというJobNodeを登録する処理になります。引数次第で別のJobNodeが登録されることはあります。

次に、ResumeOnCompletionというJobNodeを見てみます。

private class ResumeOnCompletion(
    private val continuation: Continuation<Unit>
) : JobNode() {
    override val onCancelling get() = false
    override fun invoke(cause: Throwable?) = continuation.resume(Unit)
}

このJobNodeはJob#joinで利用されます。具体的にjoinのコードを確認すると以下のようになっています。

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(cont)))
}

つまり、suspendCancellableCoroutineで引数に渡したコードブロックを実行し、呼び出し元のCoroutineを停止します。コードブロックでinVokeOnCompletionを呼び、Coroutineが終了したらResumeOnCompletion#invokeが実行されるようにJobNodeを登録しています。Coroutineの終了とは子Coroutineの終了を待つ事を意味します。RsumeOnCompletion#invokeではcontinuation.resume(Unit)を実行し、Coroutineの再開を行なっています。つまり子Jobの完了を待つjoinとしての動きが実現できています。cont.disposeOnCancellationでは呼び出し元のCoroutineのjob.cancelが呼ばれた場合に不要になったInvokeOnCancellingのJobNodeをクリアされるようにしています。

かなり長くなりましたが、JobSupportには状態が複数あり、JobNodeという特定のタイミングで動くハンドラーを登録できる機構があることがわかりました。

cancel

次にJobのcancelメソッドを見ていきます。Jobを実装したJobSupportに具体的な処理が記載されています。

    // external cancel with cause, never invoked implicitly from internal machinery
    public override fun cancel(cause: CancellationException?) {
        cancelInternal(cause ?: defaultCancellationException())
    }

cancelメソッドの引数に何も渡さなかった場合、defaultCancellationException()というメソッドが呼ばれます。このメソッドではJobCancellationExceptionのインスタンスを作成します。

internal fun cancelImpl(cause: Any?): Boolean {
    var finalState: Any? = COMPLETING_ALREADY
    if (onCancelComplete) {
        // make sure it is completing, if cancelMakeCompleting returns state it means it had make it
        // completing and had recorded exception
        finalState = cancelMakeCompleting(cause)
        if (finalState === COMPLETING_WAITING_CHILDREN) return true
    }
    if (finalState === COMPLETING_ALREADY) {
        finalState = makeCancelling(cause)
    }
    return when {
        finalState === COMPLETING_ALREADY -> true
        finalState === COMPLETING_WAITING_CHILDREN -> true
        finalState === TOO_LATE_TO_CANCEL -> false
        else -> {
            afterCompletion(finalState)
            true
        }
    }
}

まず、onCancelCompleteを見ていきます。これがtrueの場合は、そのJobを完了（cancelled=true && completed=true）にします。

では、どのような場合にonCancelCompleteはtrueになるのでしょうか？それは、そのJobがCoroutineの実行ブロックを持つかどうかです。もし実行ブロックを持たないのであれば即座にそのJobをキャンセルして問題ありません。

// 実行ブロックなし
val job2 = Job() // cancelMakeCompletingでcancel
// 実行ブロックあり
val job1 = launch {} // makeCancellingでcancel

例えば、Job()で作成したJobは実行ブロックを持たないため、即座に完了状態まで処理を進めることができます。Jobコンストラクターで作成されるJobはCompletableJobというインタフェースを実装したJobImplという具象クラスであり、JobImplはonCancelCompleteがtrueになっています。一方でCoroutienScope.launchで作成したJobはStandaloneCoroutineであり実行ブロックをもつCoroutineのため、即座にそのJobを完了状態に進めることができません。

cancelMakeCompleting関数では、そのJobの状態をActive -> Finishing -> キャンセル通知 -> 子Jobキャンセル -> Completing -> Finalまで遷移します。一方でmakeCancelling関数では、そのJobの状態をActive -> Finishing -> キャンセル通知 -> 子Jobキャンセルまでで完了までは遷移しません。

親子Jobの関連

親Jobに子Jobを追加する処理を通して親Jobと子Jobがどのように結び付いているのか見ていきます。

launch { // Aジョブ
    launch { //Bジョブ
 
    }
}

このような処理をした場合、Aジョブが親、Bジョブは子となります。

この親子関係をAジョブは知る必要がありますがどのように知るのでしょうか？

BジョブのlaunchでStandaloneCoroutineが作成されることは他の記事でも触れました。StandaloneCoroutineはAbstractCoroutineを実装しており、AbstractCoroutineのinitで親子関係の登録がなされています。

@OptIn(InternalForInheritanceCoroutinesApi::class)
@InternalCoroutinesApi
public abstract class AbstractCoroutine<in T>(
    parentContext: CoroutineContext,
    initParentJob: Boolean,
    active: Boolean
) : JobSupport(active), Job, Continuation<T>, CoroutineScope {
 
    init {
        /*
         * Setup parent-child relationship between the parent in the context and the current coroutine.
         * It may cause this coroutine to become _cancelling_ if the parent is already cancelled.
         * It is dangerous to install parent-child relationship here if the coroutine class
         * operates its state from within onCancelled or onCancelling
         * (with exceptions for rx integrations that can't have any parent)
         */
        if (initParentJob) initParentJob(parentContext[Job])
    }

initのinitParentJobで登録します。initParentJobはJobSupport側に実装があります。

protected fun initParentJob(parent: Job?) {
    assert { parentHandle == null }
    if (parent == null) {
        parentHandle = NonDisposableHandle
        return
    }
    parent.start() // make sure the parent is started
    val handle = parent.attachChild(this)
    parentHandle = handle
    // now check our state _after_ registering (see tryFinalizeSimpleState order of actions)
    if (isCompleted) {
        handle.dispose()
        parentHandle = NonDisposableHandle // release it just in case, to aid GC
    }
}

parent.attachChild(this)で親子関係の登録をしています。ここでparentはAジョブ、thisはBジョブです。次にattachChildを見ていきます。

    @Suppress("OverridingDeprecatedMember")
    public final override fun attachChild(child: ChildJob): ChildHandle {
        /*
         * Note: This function attaches a special ChildHandleNode node object. This node object
         * is handled in a special way on completion on the coroutine (we wait for all of them) and also
         * can't be added simply with `invokeOnCompletionInternal` -- we add this node to the list even
         * if the job is already cancelling.
         * It's required to properly await all children before completion and provide a linearizable hierarchy view:
         * If the child is attached when the job is already being cancelled, such a child will receive
         * an immediate notification on cancellation,
         * but the parent *will* wait for that child before completion and will handle its exception.
         */
        val node = ChildHandleNode(child).also { it.job = this }
        val added = tryPutNodeIntoList(node) { _, list ->

attachChildはAジョブ側のJobSupportで呼ばれいます。ChildHandleNodeというJobNodeが作成されlistに登録されており、前述のJobNodeを利用して親子関係を紐づけている事が理解できます。ではChildHandleNodeを見てみます。

private class ChildHandleNode(
    @JvmField val childJob: ChildJob
) : JobNode(), ChildHandle {
    override val parent: Job get() = job
    override val onCancelling: Boolean get() = true
    override fun invoke(cause: Throwable?) = childJob.parentCancelled(job)
    override fun childCancelled(cause: Throwable): Boolean = job.childCancelled(cause)
}

整理すると以下です。

invoke
- 親ジョブAが子ジョブBをキャンセルするための処理です。childJob.parentCancelledは単に子ジョブのcancelImplを呼んでいます。
childCancelled
- 子ジョブBが親ジョブAに失敗を通知するための処理

では次に、job.childCancelledの実装を見て見ましょう。

public open fun childCancelled(cause: Throwable): Boolean {
    if (cause is CancellationException) return true
    return cancelImpl(cause) && handlesException
}

この実装から子がCancellationExceptionをthrowしても親はキャンセルされないことがわかります。また、IOExceptionのようなその他の例外の場合は親をキャンセルしていることがわかります。

ここで前述のCancellationExceptionを握り潰してはいけない理由を説明した際に、CancellationExceptionはrethrowしても問題ないという部分とつながりました。

launch {
    launch {
        try {
            delay(1.seconds)
        } catch (e: CancellationException) {
            throw e
        } catch (e: Exception) {
            // 握りつぶす
            e.printStackTrace()
        }
    }
}

CancellationExceptionは親Jobをキャンセルさせるための例外ではなく、Coroutineのキャンセルを検知したsuspend関数などが、Coroutineの実行を中断させるための例外です。そのため、CancellationExceptionをcatchした場合は原則としてrethrowし、キャンセルされたCoroutineの処理を継続しないようにしましょう。

まとめ

Kotlin CoroutinesがStructured Concurrencyをどのように実現しているのか、Jobの実装を中心に確認しました。

CoroutineのライフサイクルはJobによって表現される
JobSupportがJobの状態遷移やキャンセル、親子関係などの具体的な処理を実装している
親JobはChildHandleNodeを自身のNodeListに登録することで子Jobを管理している
親JobがキャンセルされるとChildHandleNodeを経由して子Jobへキャンセルが伝播する
子JobがCancellationException以外で失敗すると、ChildHandleNodeを経由して親Jobへ失敗が伝播する
親Jobは自身のCoroutine bodyが終了しても、子Jobが残っていれば完了せず、子Jobの完了を待つ
join()はResumeOnCompletionをJobに登録してCoroutineをsuspendすることで、Jobの最終完了を待っている

つまりKotlin CoroutinesのStructured Concurrencyは、単にcoroutineScopeというAPIによって実現されているわけではありません。

内部ではJob同士が親子関係を持ち、

親 → 子へのキャンセル伝播
子 → 親への失敗伝播
親による子の完了待ち

といった仕組みをJobSupportが提供することで、Coroutineのライフタイムが構造化されています。

最初は

「親Coroutineは子Coroutineが終わるまで完了しない」

という単純なルールに見えましたが、実装まで追ってみると、

Coroutine ↓ Job ↓ JobSupport ↓ JobNode / NodeList ↓ ChildHandleNode

という仕組みによって、このルールが実現されていることが分かりました。