【Coroutines】キャンセルとタイムアウト

ソース記事はこちら
このセクションはコルーチンのキャンセルとタイムアウトをカバーしている。

コルーチンの実行のキャンセル

長時間実行するアプリケーションにおいて、そのバックグラウンドのコルーチンをきめ細かく制御する必要があるかるかもしれない。例えば、あるユーザーはコルーチンを起動したページを閉じ、その結果はもう必要なく、操作をキャンセルするかもしれない。launch関数は、実行しているコルーチンをキャンセルするために使うことができるJobを返却する。

import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch {
        repeat(1000) { i ->
            println("job: I'm sleeping $i ...")
            delay(500L)
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancel() // cancels the job
    job.join() // waits for job's completion 
    println("main: Now I can quit.") 
}

次のような出力となる。

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
main: Now I can quit.

mainからjob.cancelが呼び出されるとすぐに、ジョブはキャンセルされたため、ほかのコルーチンからの出力は見られなくなる。cancelとjoinを組み合わせるcancelAndJoin Job拡張関数もある。

キャンセルは協調的

コルーチンのキャンセルは協調的である。コルーチンのコードは、キャンセルに対して協調的でなければならない。kotlinx.coroutines内のすべてのsuspend関数は、キャンセル可能である。それらは、コルーチンのキャンセルをチェックして、キャンセルされると、CancellationExceptionをスローする。しかし、コルーチンが計算を行っており、キャンセルのチェックをしていない場合、キャンセルすることはできない。次の例に示すように。

import kotlinx.coroutines.*

fun main() = runBlocking {
    val startTime = System.currentTimeMillis()
    val job = launch(Dispatchers.Default) {
        var nextPrintTime = startTime
        var i = 0
        while (i < 5) { // computation loop, just wastes CPU
            // print a message twice a second
            if (System.currentTimeMillis() >= nextPrintTime) {
                println("job: I'm sleeping ${i++} ...")
                nextPrintTime += 500L
            }
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")  
    // ※次のような出力になる
    // job: I'm sleeping 0 ...
    // job: I'm sleeping 1 ...
    // job: I'm sleeping 2 ...
    // main: I'm tired of waiting!
    // job: I'm sleeping 3 ...
    // job: I'm sleeping 4 ...
    // main: Now I can quit.  
}

これを実行し、キャンセルの後でも"I'm sleeping"が出力され続け、5回の繰り返しの後、自分自身によってジョブが完了することを確認しよう。
同じ問題が、CancellationExceptionをキャッチし、再スローしないことによっても見られる。

import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch(Dispatchers.Default) {
        repeat(5) { i ->
            try {
                // print a message twice a second
                println("job: I'm sleeping $i ...")
                delay(500)
            } catch (e: Exception) {
                // log the exception
                println(e)
            }
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")    
    // ※次のような出力になる
    // job: I'm sleeping 0 ...
    // job: I'm sleeping 1 ...
    // job: I'm sleeping 2 ...
    // main: I'm tired of waiting!
    // kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled;     
    // job="coroutine#2":StandaloneCoroutine{Cancelling}@724c07e6
    // job: I'm sleeping 3 ...
    // kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled;
    // job="coroutine#2":StandaloneCoroutine{Cancelling}@724c07e6
    // job: I'm sleeping 4 ...
    // kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled;
    // job="coroutine#2":StandaloneCoroutine{Cancelling}@724c07e6
    // main: Now I can quit.
}

Exceptionをキャッチすることはアンチパターンであるが、この問題はより巧妙な方法で表面化するかもしれない。例えば、runcatching関数を使って、CancellationExceptionを再スローしないときのような場合である。

計算コードをキャンセル可能にする

計算コードをキャンセル可能にするには、二つのアプローチがある。一つは、キャンセルを定期的にチェックするsuspend関数を呼び出すことである。その目的のための良い選択肢として、yield関数がある。もう一つの方法は、明示的にキャンセルステータスをチェックすることである。後者のアプローチを試してみよう。
以前の例でwhile (i < 5)をwhile (isActive)に置き換え、それを返却するようにする。

import kotlinx.coroutines.*

fun main() = runBlocking {
    val startTime = System.currentTimeMillis()
    val job = launch(Dispatchers.Default) {
        var nextPrintTime = startTime
        var i = 0
        while (isActive) { // cancellable computation loop
            // print a message twice a second
            if (System.currentTimeMillis() >= nextPrintTime) {
                println("job: I'm sleeping ${i++} ...")
                nextPrintTime += 500L
            }
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")    
}
// ※以下の出力となる
// job: I'm sleeping 0 ...
// job: I'm sleeping 1 ...
// job: I'm sleeping 2 ...
// main: I'm tired of waiting!
// main: Now I can quit

ご覧のように、現在はこのループはキャンセル可能である。isActiveはCoruotineScopeオブジェクト経由のコルーチンの内部で利用可能な、拡張プロパティである。

finallyでリソースを閉じる

キャンセル可能なsuspend関数は、キャンセルにおいてCancellationExceptionをスローするが、それは通常のの方法で扱うことができる。例えば、try {...} finally {...}式とKotlinのuse関数はコルーチンがキャンセルされるときに、通常通りに完了のふるまいを実行する。

import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch {
        try {
            repeat(1000) { i ->
                println("job: I'm sleeping $i ...")
                delay(500L)
            }
        } finally {
            println("job: I'm running finally")
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")    
}

joinとcancelAndAndJoinは両方とも、すべての完了動作が完了するのを待つため、上記の例では、次のような出力が生成される。

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
job: I'm running finally
main: Now I can quit.

キャンセル不可能なブロックの実行

前回の例のfinallyブロックでsuspend関数を使おうとすると、CancellationExceptionを引き起こす。なぜならそのコードを実行しているコルーチンは、キャンセルされているからである。大抵は、これは問題ではない。というのも、すべての行儀の良い終了操作(ファイルのクローズ、ジョブのキャンセル、何らかの通信チャンネルのクローズ)は、大抵ノンブロッキングで、suspend関数を含まない。しかしレアケースでキャンセルされたコルーチンで、一時停止が必要なとき、次の例で見えるように、withContext関数と、NonCancellableコンテキストを使って、対応するコードをwithContext(NonCancellable) {...}でラップすることができる。

import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch {
        try {
            repeat(1000) { i ->
                println("job: I'm sleeping $i ...")
                delay(500L)
            }
        } finally {
            withContext(NonCancellable) {
                println("job: I'm running finally")
                delay(1000L)
                println("job: And I've just delayed for 1 sec because I'm non-cancellable")
            }
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")
}
// ※次のような出力になる
// job: I'm sleeping 0 ...
// job: I'm sleeping 1 ...
// job: I'm sleeping 2 ...
// main: I'm tired of waiting!
// job: I'm running finally
// job: And I've just delayed for 1 sec because I'm non-cancellable
// main: Now I can quit.

タイムアウト

コルーチンの実行をキャンセルするもっとも明らかで実用的な理由は、その実行のために、いくつかのタイムアウトを超過するということである。手動で対応するJobへの参照を追跡して、遅らせて追跡しているものをキャンセルする別のコルーチンを起動することもできるが、それを行うwithTimeoutをすぐに使うこともできる。次の例を見てみよう。

import kotlinx.coroutines.*

fun main() = runBlocking {
    withTimeout(1300L) {
        repeat(1000) { i ->
            println("I'm sleeping $i ...")
            delay(500L)
        }
    }
}

これは次の出力を生成する。

I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Exception in thread "main" kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1300 ms

withTimeoutによってスローされるTimeoutCancellationExceptionは、CancellationExceptionのサブクラスである。前のコンソールに表示されたスタックトレースを見ていない。それはキャンセルされたコルーチンの内部では、CancellationExceptionはコルーチンの完了にとって通常の理由とみなされているからである。しかし、この例ではwithTimeoutはmain関数のすぐ内側で使っていた。
キャンセルは単なる例外であるため、すべてのリソースは通常の方法でクローズされる。何らかのタイムアウトについて、特に追加のふるまいが必要な場合は、タイムアウトを含むコードをtry {...} catch (e: TimeoutCancellationException) {...}にラップするか、withTimeoutOrNull関数を使うことができ、それはwithTimeoutに似ているが、例外をスローする代わりに、タイムアウトでnullを返却する。

import kotlinx.coroutines.*

fun main() = runBlocking {
    val result = withTimeoutOrNull(1300L) {
        repeat(1000) { i ->
            println("I'm sleeping $i ...")
            delay(500L)
        }
        "Done" // will get cancelled before it produces this result
    }
    println("Result is $result")
}

このコードを実行するともう例外は存在しない。

I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Result is null

非同期のタイムアウトとリソース

withTimeout内のタイムアウトイベントは、そのブロックで実行しているコードに関して非同期であり、いつでも、タイムアウトブロックの内部から戻る直前でさえ、発生する可能性がある。もしブロックの内部で、あるリソースを開いたり獲得し、かつブロックの外側ではクローズまたは解放する必要がある場合は、このことを覚えておいてほしい。
例えば、ここにResourceクラスでクローズできるリソースを模倣している。それは、acquiredカウンタを増加することで作成され、close関数からカウンタを減少させる、その回数を単純に追跡する。短いタイムアウトを持つ多くのコルーチンを実行し、withTimeoutブロックの内側からリソースを獲得し、少し後で外側からそれを解放してみよう。

import kotlinx.coroutines.*

var acquired = 0

class Resource {
    init { acquired++ } // リソースを獲得する
    fun close() { acquired-- } // リソースを解放する
}

fun main() {
    runBlocking {
        repeat(100_000) { // 10万個のコルーチンを起動する。
            launch { 
                val resource = withTimeout(60) { // タイムアウト60 ms
                    delay(50) // ディレイ50 ms
                    Resource() // リソースを獲得し、withTimeoutブロックからそれを返却する     
                }
                resource.close() // リソースを解放する
            }
        }
    }
    // runBlockingの外側ですべてのコルーチンが完了する
    println(acquired) // リソースがまだ獲得されている数を出力する
}

上記のコードを実行すると、常にゼロを出力しないことがわかるだろう。それはPCのタイミングに依存するかもしれず、実際にゼロ以外の値を見るためにこの例のタイムアウトを微調整する必要があるかもしれないが。

留意すべきは、ここで10万個のコルーチンからacquiredカウンタを増加減少することは、完全に安全であるということである。というのは、それは常に同じメインスレッドから発生しているためである。そのことに関してはコルーチンコンテキストの章でより詳しく説明されるだろう。

問題を回避するため、リソースへの参照を、withTimeoutブロックから返却するのとは違って、変数に保存することができる。

import kotlinx.coroutines.*

var acquired = 0

class Resource {
    init { acquired++ } // リソースを獲得する
    fun close() { acquired-- } // リソースを解放する
}

fun main() {
    runBlocking {
        repeat(100_000) { // 10万個のコルーチンを起動する。
            launch { 
                var resource: Resource? = null // まだ獲得されていない
                try {
                    withTimeout(60) { // タイムアウト60 ms
                        delay(50) // ディレイ50 ms
                        resource = Resource() // 獲得した場合、変数にリソースを保存する      
                    }
                    // ここでリソースに関して他のことをすることができる
                } finally {  
                    resource?.close() // 獲得できた場合、リソースを解放する
                }
            }
        }
    }
    // runBlockingの外側ですべてのコルーチンが完了する
    println(acquired) // リソースがまだ獲得されている数を出力する
}

この例では常にゼロを出力する。リソースはリークしない。