【Coroutinesガイド】suspend関数の構成

※ソース記事はこちら
このセクションではsuspend関数の構成に対する様々なアプローチについてカバーしている。

デフォルトはシーケンシャル

リモートのサービスや計算のような、有益なことをする二つのsuspend関数がどこかに定義されていると仮定する。それらは単に有益なふりをしており、実際は次の例のため、それぞれ1秒遅延するだけとする。

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // ここで有益なことをするふりをする
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // ここでも有益なことをするふりをする
    return 29
}

それらをシーケンシャル-最初はdoSomethingUsefulOne、その後でdoSomethingUsefulTwo-に呼び出しそれらの結果を合計する必要がある場合、どうするか？実際は、二つ目を呼び出す必要があるかどうか決めるため、あるいは呼び出し方を決めるために、最初の関数の結果を使う場合にそれを行う。
我々は通常のシーケンシャルな呼び出しを使う。なぜなら、コルーチンのコードは通常のコードとまったく似ており、デフォルトがシーケンシャルだからである。次の例は、両方のsuspend関数を実行するのにかかった合計の時間を計測するために、実際にやってみる。

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
    val time = measureTimeMillis {
        val one = doSomethingUsefulOne()
        val two = doSomethingUsefulTwo()
        println("The answer is ${one + two}")
    }
    println("Completed in $time ms")    
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) //  ここで有益なことをするふりをする
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // ここでも有益なことをするふりをする
    return 29
}

これは次のようなものを生成する。

The answer is 42
Completed in 2017 ms

asyncの使用による並列

doSomethingUsefulOneとdoSomethingUsefulTwo呼び出しの間に依存関係がなく、両方を同時に実行することで、より早く答えが欲しい場合はどうするか？
そこはasyncが役立つ所である。
概念的にasyncはlaunchにちょうど良く似ている。それは別のコルーチンを開始し、他のすべてのコルーチンと並列に動作する軽量なスレッドである。違いは、launchはJobを返却し、それは結果の値を持たない。一方asyncはDeferredを返却し、それは軽量なノンブロッキングなfutureであり、結果を後で提供するpromiseを表す。deferredの値について.await()を使うことで、最終的な結果を得ることができるが、DeferredはJobでもあるため、必要ならキャンセルすることができる。

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
    val time = measureTimeMillis {
        val one = async { doSomethingUsefulOne() }
        val two = async { doSomethingUsefulTwo() }
        println("The answer is ${one.await() + two.await()}")
    }
    println("Completed in $time ms")
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // ここで有益なことをするふりをする
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // ここでも有益なことをするふりをする
    return 29
}

これはこのような結果となる。

The answer is 42
Completed in 1017 ms

これは二倍速い。なぜなら二つのコルーチンが並列に実行されているためである。留意すべきは、コルーチンの並列性は常に明示的だということである。

lazyに開始するasync

オプションで、asyncはstartパラメータにCoroutineStart.LAZYをセットして作ることができる。このモードでは、結果がawaitによって必要になるか、Jobのstart関数が呼び出されるときのみ、コルーチンが開始される。次の例を実行してみる。

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
    val time = measureTimeMillis {
        val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
        val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
        // 何らかの計算
        one.start() // 最初のものを開始
        two.start() // 二番目のものを開始
        println("The answer is ${one.await() + two.await()}")
    }
    println("Completed in $time ms")    
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // ここで有益なことをするふりをする
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // ここでも有益なことをするふりをする
    return 29
}

これはこのようなものを出力する。

The answer is 42
Completed in 1017 ms

この通り、ここでは二つのコルーチンが定義されているが、以前の例のように実行はされない。しかし、startを呼び出すことにより、まさにいつ実行を開始するかをプログラマに制御が与えられている。最初にoneを開始し、その後twoを開始し、その後で個々のコルーチンが完了するのを待つ。
留意すべきは、もし個々のコルーチンについて、最初にstartを呼ばずにprintlnの中でawaitを呼ぶだけの場合、これはシーケンシャルなふるまいにつながるだろう。というのは、awaitはコルーチンの実行を開始し、完了を待つため、lazyの意図したユースケースではない。async(start = CoroutineStart.LAZY)のユースケースは、suspend関数を含む値の計算をする場合における標準のlazy関数の置き換えである。

asyncスタイルの関数

構造化された並列性から離れたGlobalScopeの参照を使って、asyncコルーチンビルダーを使って非同期にdoSomethingUsefulOneとdoSomethingUsefulTwoを呼び出すasyncスタイルの関数を定義することができる。そのような関数を"...Async"サフィックスをつけて名づけ、それらは非同期に計算が開始され、誰かが結果を得るために生じるdeferredな値を使うことを必要とする、という事実を強調する。

GlobalScopeは、以下で説明するように、簡単でない方法に裏目に出ることがある、繊細なAPIである。そのためGlobalScopeを使うには、@OptIn(DelicateCoroutinesApi::class)とともに明示的なopt-inが必要である。

// somethingUsefulOneAsyncの結果の型はDeferred<Int>である
@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulOneAsync() = GlobalScope.async {
    doSomethingUsefulOne()
}

// somethingUsefulTwoAsyncの結果の型はDeferred<Int>である
@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulTwoAsync() = GlobalScope.async {
    doSomethingUsefulTwo()
}

留意すべきは、xxxAsync関数は、suspend関数ではない。それらはどこからでも使える。しかしそれらを呼び出すコードで実行するふるまいは、常に非同期(ここでは並列を意味する)であることを示す。
次の例で、それらをコルーチンの外で使うことをお見せする。

import kotlinx.coroutines.*
import kotlin.system.*

// 留意すべきは、この例では`main`の右に`runBlocking`が無いということである
fun main() {
    val time = measureTimeMillis {
        // コルーチンの外でasync実行を開始することができる
        val one = somethingUsefulOneAsync()
        val two = somethingUsefulTwoAsync()
        // しかし、結果を待つにはsuspendingかblockingに含めなければならない。
        // ここでは、結果を待つ間、メインスレッドをブロックするため`runBlocking { ... }`を使用する
        runBlocking {
            println("The answer is ${one.await() + two.await()}")
        }
    }
    println("Completed in $time ms")
}

@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulOneAsync() = GlobalScope.async {
    doSomethingUsefulOne()
}

@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulTwoAsync() = GlobalScope.async {
    doSomethingUsefulTwo()
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // ここで有益なことをするふりをする
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // ここでも有益なことをするふりをする
    return 29
}

このasync関数のプログラミング形式は、例示のために提供されている。なぜならそれは他のプログラム言語では一般的な形式だからである。Kotlinコルーチンに関してこの形式は、以下で説明する理由により、推奨しない。

val one = somethingUsefulOneAsync()の行と、one.await()式の間で、コード内に何かロジックエラーがあり、プログラムが例外をスローし、プログラムにより実行されていた操作が異常終了するとどうなるか、考えてみよう。通常は、グローバルなエラーハンドラーがその例外をキャッチし、開発者にエラーのログと報告をするが、プログラムはそれ以外は他の操作を継続することがありうる。しかしここで、たとえ開始した操作が異常終了したとしても、somethingUsefulOneAsyncはまだバックグラウンドで実行している。この問題は、以下のセクションでお見せする構造化された並行性では発生しない。

asyncを使う構造化された並行性

"asyncの使用による並列"の例を取り上げ、同時にdoSomethingUsefulOneとdoSomethingUsefulTwoを実行して、その結果の合計を返却する関数を取り出そう。asyncコルーチンビルダーは、CoroutineScopeの拡張として定義されているため、そのスコープ内で、それを持つ必要があり、それはCoroutineScope関数が提供するものである。

suspend fun concurrentSum(): Int = coroutineScope {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    one.await() + two.await()
}

この方法では、concurrentSum関数のコード内で何か変なことが起きる場合、例外をスローし、そのスコープ内で起動したすべてのコルーチンはキャンセルされるだろう。

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
    val time = measureTimeMillis {
        println("The answer is ${concurrentSum()}")
    }
    println("Completed in $time ms")    
}

suspend fun concurrentSum(): Int = coroutineScope {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    one.await() + two.await()
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // ここで有益なことをするふりをする
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // ここでも有益なことをするふりをする
    return 29
}

上のmain関数の出力から明らかなように、まだ並行に実行をしている。

The answer is 42
Completed in 1017 ms

キャンセルは、常にコルーチン階層を通じて伝播する。

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    try {
        failedConcurrentSum()
    } catch(e: ArithmeticException) {
        println("Computation failed with ArithmeticException")
    }
}

suspend fun failedConcurrentSum(): Int = coroutineScope {
    val one = async<Int> { 
        try {
            delay(Long.MAX_VALUE) // 非常に長い計算をエミュレートする
            42
        } finally {
            println("First child was cancelled")
        }
    }
    val two = async<Int> { 
        println("Second child throws an exception")
        throw ArithmeticException()
    }
    one.await() + two.await()
}

最初のasyncと待っている親は、(twoという名前の)子の一つの失敗で、両方ともキャンセルされることに注意してもらいたい。

Second child throws an exception
First child was cancelled
Computation failed with ArithmeticException