Kotlin 1.3: コルーチンコンテキストとディスパッチャ (1/3)

Table of Contents

Kotlin コルーチンについてのドキュメント、コンテキストとディスパッチャのセクションを読みました。分量が多いため、3つの記事に分けて記述しました。

コルーチンは常に、 Kotlin 標準ライブラリで定義されている CoroutineContext 型の値によって表される、あるコンテキストで実行されます。

コルーチンのコンテキストは、様々な要素の集合です。主な要素はこれまでに見たコルーチンの Job とそのディスパッチャです。ディスパッチャをこのセクションで説明します。

検証環境

Kotlin 1.3.0-rc-146
kotlinx-coroutines-core 0.30.2

ディスパッチャとスレッド

コルーチンのコンテキストはコルーチンディスパッチャを含んでいます。コルーチンディスパッチャは対応するコルーチンが使用するスレッドを決定します。コルーチンディスパッチャはコルーチンの実行を特定のスレッドに限定したり、スレッドプールにディスパッチしたり、スレッドプールに非同期で実行させることができます。

launch や async をはじめとする全てのコルーチンビルダではオプショナルの CoroutineContext パラメータが使用できます。それによって新しいコルーチンとその他のコンテキスト要素のディスパッチャを明示的に指定することができます。

例として次のコードを実行してみます。

fun main(args: Array<String>) = runBlocking<Unit> {
// context of the parent, main runBlocking coroutine
launch {
println("main runBlocking      : in thread ${Thread.currentThread().name}")
}
// not confined -- will work with main thread
launch(Dispatchers.Unconfined) {
println("Unconfined            : in thread ${Thread.currentThread().name}")
}
// will get dispatched to DefaultDispatcher
launch(Dispatchers.Default) {
println("Default               : in thread ${Thread.currentThread().name}")
}
// will get its own new thread
launch(newSingleThreadContext("MyOwnThread")) {
println("newSingleThreadContext: in thread ${Thread.currentThread().name}")
}
}

fun main(args: Array<String>) = runBlocking<Unit> {

// context of the parent, main runBlocking coroutine

launch {

println("main runBlocking : in thread ${Thread.currentThread().name}")

}

// not confined -- will work with main thread

launch(Dispatchers.Unconfined) {

println("Unconfined : in thread ${Thread.currentThread().name}")

}

// will get dispatched to DefaultDispatcher

launch(Dispatchers.Default) {

println("Default : in thread ${Thread.currentThread().name}")

}

// will get its own new thread

launch(newSingleThreadContext("MyOwnThread")) {

println("newSingleThreadContext: in thread ${Thread.currentThread().name}")

}

出力は次のようになります。

Unconfined            : in thread main
Default               : in thread DefaultDispatcher-worker-1
newSingleThreadContext: in thread MyOwnThread
main runBlocking      : in thread main

Unconfined : in thread main

Default : in thread DefaultDispatcher-worker-1

newSingleThreadContext: in thread MyOwnThread

main runBlocking : in thread main

launch { ... } を引数なしで実行した場合、 launch が実行された CoroutineScope を引き継ぎます。サンプルコードでは、メインスレッドで動くメイン関数内の runBlockng コルーチンのコンテキストを継承します。

Dispatchers.Unconfined は特別なディスパッチャで、引数に渡した場合はメインスレッドで実行されます。しかしそれは後述するように異なるメカニズムです。

デフォルトディスパッチャ (Dispatchers.Default) はコルーチンが GlobalScope で起動された時に使われ、共通のバックグラウンドスレッドプールを利用します。すなわち launch(Dispatchers.Default) { ... } は GlobalScope.launch { ... } を実行した時と同じディスパッチャを使います。

newSingleThreadContext 関数はコルーチンのために新しいスレッドを作成します。専用のスレッドは非常に高価なリソースです。実際のアプリケーションでは、不要になったときに解放するか、close 関数を使用するか、トップレベルの変数に格納してアプリケーション全体で再利用する必要があります。

実験として次のコードを実行してみます。

fun main(args: Array<String>) = runBlocking<Unit> {
val outputList = mutableListOf<String>()
// context of the parent, main runBlocking coroutine
launch {
outputList.add("main runBlocking 1      : in thread ${Thread.currentThread().name}")
}
// not confined -- will work with main thread
launch(Dispatchers.Unconfined) {
outputList.add("Unconfined 1            : in thread ${Thread.currentThread().name}")
}
// will get dispatched to DefaultDispatcher
launch(Dispatchers.Default) {
outputList.add("Default 1               : in thread ${Thread.currentThread().name}")
}
// will get its own new thread
launch(newSingleThreadContext("MyOwnThread 1")) {
outputList.add("newSingleThreadContext 1: in thread ${Thread.currentThread().name}")
}
// context of the parent, main runBlocking coroutine
launch {
outputList.add("main runBlocking 2      : in thread ${Thread.currentThread().name}")
}
// not confined -- will work with main thread
launch(Dispatchers.Unconfined) {
outputList.add("Unconfined 2            : in thread ${Thread.currentThread().name}")
}
// will get dispatched to DefaultDispatcher
launch(Dispatchers.Default) {
outputList.add("Default 2               : in thread ${Thread.currentThread().name}")
}
// will get its own new thread
launch(newSingleThreadContext("MyOwnThread 2")) {
outputList.add("newSingleThreadContext 2: in thread ${Thread.currentThread().name}")
}
delay(1000L)
outputList.sorted().forEach { println(it) }
}

fun main(args: Array<String>) = runBlocking<Unit> {

val outputList = mutableListOf<String>()

// context of the parent, main runBlocking coroutine

launch {

outputList.add("main runBlocking 1 : in thread ${Thread.currentThread().name}")

}

// not confined -- will work with main thread

launch(Dispatchers.Unconfined) {

outputList.add("Unconfined 1 : in thread ${Thread.currentThread().name}")

}

// will get dispatched to DefaultDispatcher

launch(Dispatchers.Default) {

outputList.add("Default 1 : in thread ${Thread.currentThread().name}")

}

// will get its own new thread

launch(newSingleThreadContext("MyOwnThread 1")) {

outputList.add("newSingleThreadContext 1: in thread ${Thread.currentThread().name}")

}

// context of the parent, main runBlocking coroutine

launch {

outputList.add("main runBlocking 2 : in thread ${Thread.currentThread().name}")

}

// not confined -- will work with main thread

launch(Dispatchers.Unconfined) {

outputList.add("Unconfined 2 : in thread ${Thread.currentThread().name}")

}

// will get dispatched to DefaultDispatcher

launch(Dispatchers.Default) {

outputList.add("Default 2 : in thread ${Thread.currentThread().name}")

}

// will get its own new thread

launch(newSingleThreadContext("MyOwnThread 2")) {

outputList.add("newSingleThreadContext 2: in thread ${Thread.currentThread().name}")

}

delay(1000L)

outputList.sorted().forEach { println(it) }

}

出力は次のようになります。

Default 1               : in thread DefaultDispatcher-worker-1
Default 2               : in thread DefaultDispatcher-worker-4
Unconfined 1            : in thread main
Unconfined 2            : in thread main
main runBlocking 1      : in thread main
main runBlocking 2      : in thread main
newSingleThreadContext 1: in thread MyOwnThread 1
newSingleThreadContext 2: in thread MyOwnThread 2

Default 1 : in thread DefaultDispatcher-worker-1

Default 2 : in thread DefaultDispatcher-worker-4

Unconfined 1 : in thread main

Unconfined 2 : in thread main

main runBlocking 1 : in thread main

main runBlocking 2 : in thread main

newSingleThreadContext 1: in thread MyOwnThread 1

newSingleThreadContext 2: in thread MyOwnThread 2

Dispatchers.Default を利用した時はプログラムが自動的にスレッドを選択しているのがわかります。

Unconfined デイスパッチャと Confined ディスパッチャ

コルーチンディスパッチャ Dispatchers.Unconfined はコルーチンを呼び出し元のスレッドで実行します。しかしそれは最初のサスペンドポイント(一時停止地点)までです。サスペンド後は呼び出されたサスペンディングファンクションによって決定されたスレッドで処理を再開します。 Unconfined ディスパッチャはコルーチンが CPU時間を消費せず、特定のスレッドに固定化されている UI のような共通のデータを更新しない場合に適しています。

一方で、デフォルトでは、外部の CoroutineScope のディスパッチャは継承されます。特に、 runBlocking コルーチンのデフォルトのディスパッチャは呼び出し元のスレッドに固定化されていますので、継承することで予測可能な先入先出(FIFO)スケジューリングで当該スレッドに実行権限を限定することができます。

次のコードを実行して、挙動を確認します。

fun main(args: Array<String>) = runBlocking<Unit> {
launch(Dispatchers.Unconfined) {
// not confined -- will work with main thread
println("Unconfined      : I'm working in thread ${Thread.currentThread().name}")
delay(500)
println("Unconfined      : After delay in thread ${Thread.currentThread().name}")
}
launch {
// context of the parent, main runBlocking coroutine
println("main runBlocking: I'm working in thread ${Thread.currentThread().name}")
delay(1000)
println("main runBlocking: After delay in thread ${Thread.currentThread().name}")
}
}

fun main(args: Array<String>) = runBlocking<Unit> {

launch(Dispatchers.Unconfined) {

// not confined -- will work with main thread

println("Unconfined : I'm working in thread ${Thread.currentThread().name}")

delay(500)

println("Unconfined : After delay in thread ${Thread.currentThread().name}")

}

launch {

// context of the parent, main runBlocking coroutine

println("main runBlocking: I'm working in thread ${Thread.currentThread().name}")

delay(1000)

println("main runBlocking: After delay in thread ${Thread.currentThread().name}")

}

出力は次のようになります。

Unconfined      : I'm working in thread main
main runBlocking: I'm working in thread main
Unconfined      : After delay in thread kotlinx.coroutines.DefaultExecutor
main runBlocking: After delay in thread main

Unconfined : I'm working in thread main

main runBlocking: I'm working in thread main

Unconfined : After delay in thread kotlinx.coroutines.DefaultExecutor

main runBlocking: After delay in thread main

Unconfined ディスパッチャを使った処理は、サスペンド関数 (delay) が実行されるまでは呼び出し元と同じスレッドで、その後は異なるスレッド(DefaultExecutorスレッド)で実行されています。

Unconfined ディスパッチャは高度なメカニズムです。コルーチンの操作はすぐに実行する必要があるため、後で実行するためのコルーチンのディスパッチが不要であるまたは望まない副作用を生じるという特定のコーナーケースで役立ちます。一般的なコードでは、 Unconfined ディスパッチャを使用しない方がいいです。

コルーチンとスレッドのデバッグ

コルーチンは、ひとつのスレッドの上で停止し他のスレッドで処理を再開することができます。シングルスレッドのディスパッチャでも、コルーチンがそのとき「どこ」で「なに」をしているのか判別するのは難しいでしょう。一般的な、スレッドを使ったアプリケーションのデバッグ手法は、スレッド名をログファイルに書き出すことです。この昨日は広くログ書き込みのフレームワークでサポートされています。コルーチンを使う場合は、スレッド名だけは状況を十分に表示できませんから、 kotlinx.coroutines が簡単にデバッグするための機構を含んでいます。

次のコードを、 JVM オプション -Dkotlinx.coroutines.debug をつけて実行してみます。

fun log(msg: String) =
println("[${Thread.currentThread().name}] $msg")
fun main(args: Array<String>) = runBlocking<Unit> {
val a = async {
log("I'm computing a piece of the answer")
6
}
val b = async {
log("I'm computing another piece of the answer")
7
}
log("The answer is ${a.await() * b.await()}")
}

fun log(msg: String) =

println("[${Thread.currentThread().name}] $msg")

fun main(args: Array<String>) = runBlocking<Unit> {

val a = async {

log("I'm computing a piece of the answer")

}

val b = async {

log("I'm computing another piece of the answer")

}

log("The answer is ${a.await() * b.await()}")

}

出力は次のようになります。

[main @coroutine#2] I'm computing a piece of the answer
[main @coroutine#3] I'm computing another piece of the answer
[main @coroutine#1] The answer is 42

[main @coroutine#2] I'm computing a piece of the answer

[main @coroutine#3] I'm computing another piece of the answer

[main @coroutine#1] The answer is 42

3つのコルーチンが出力されています。ひとつめ “#1” はメインのコルーチンで、 runBlocking で生成されたものになります。他のコルーチンは値 a を計算するもの (“#2”) と値 b を計算するもの (“#3”) で、 runBlocking のコンテキストで実行されており、メインスレッドに固定されています。

もしオプション -Dkotlinx.coroutines.debug をつけないと、出力は次のようになります。コルーチンの番号が出力されません。

[main] I'm computing a piece of the answer
[main] I'm computing another piece of the answer
[main] The answer is 42

[main] I'm computing a piece of the answer

[main] I'm computing another piece of the answer

[main] The answer is 42

Kotlin 1.3: コルーチンコンテキストとディスパッチャ (2/3)に続きます。

The Life

Kotlin 1.3: コルーチンコンテキストとディスパッチャ (1/3)

検証環境

ディスパッチャとスレッド

Unconfined デイスパッチャと Confined ディスパッチャ

コルーチンとスレッドのデバッグ

関連記事

この記事を見た人は次の記事も見ています

A Life Summary of an Gypsy

検証環境

ディスパッチャとスレッド

Unconfined デイスパッチャ と Confined ディスパッチャ

コルーチンとスレッドのデバッグ

関連記事

この記事を見た人は次の記事も見ています

A Life Summary of an Gypsy

Unconfined デイスパッチャと Confined ディスパッチャ