Отмена и таймауты

Отмена выполнения сопрограммы

В долго выполняющихся приложениях может потребоваться точный контроль над фоновыми сопрограммами. Например, пользователь может закрыть страницу, которая запустила сопрограмму, и теперь её результат больше не нужен, и её операцию можно отменить. Функция launch возвращает Job, которую можно использовать для отмены выполняемой сопрограммы:

import kotlinx.coroutines.*

fun main() = runBlocking {
//sampleStart
    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.")
//sampleEnd    
}

Она выведет следующий результат:

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, мы не увидим никакого вывода от другой сопрограммы, поскольку она была отменена. Также есть функция расширения Job — cancelAndJoin, которая объединяет вызовы cancel и join.

Отмена — это кооперативный процесс

Отмена сопрограмм — кооперативный процесс. Код сопрограммы должен сотрудничать, чтобы быть отменяемым. Все приостанавливающие функции в kotlinx.coroutines являются отменяемыми. Они проверяют отмену сопрограммы и выбрасывают CancellationException при отмене. Однако, если сопрограмма работает с вычислениями и не проверяет отмену, то её нельзя отменить, как показано в следующем примере:

import kotlinx.coroutines.*

fun main() = runBlocking {
//sampleStart
    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.")
//sampleEnd    
}

Запустите его, чтобы увидеть, что он продолжает печатать «I'm sleeping», даже после отмены, пока задача не завершит работу самостоятельно после пяти итераций.

Такую же проблему можно наблюдать, перехватывая CancellationException и не перебрасывая его:

import kotlinx.coroutines.*

fun main() = runBlocking {
//sampleStart
    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.")
//sampleEnd    
}

Хотя перехват Exception — это антипаттерн, эта проблема может проявляться более скрытно, например, при использовании функции runCatching, которая не перебрасывает CancellationException.

Делаем вычислительный код отменяемым

Существует два подхода к тому, чтобы сделать вычислительный код отменяемым. Первый — периодически вызывать приостанавливающую функцию, которая проверяет отмену. Для этой цели хорошо подходит функция yield. Другой — явно проверять состояние отмены. Давайте попробуем последний подход.

Замените while (i < 5) в предыдущем примере на while (isActive) и запустите его повторно.

import kotlinx.coroutines.*

fun main() = runBlocking {
//sampleStart
    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.")
//sampleEnd    
}

Как видите, теперь этот цикл отменён. Свойство isActive доступно внутри сопрограммы через объект CoroutineScope.

Закрытие ресурсов с помощью finally

При отмене отменяемые приостанавливающие функции выбрасывают CancellationException, которые можно обработать обычным способом. Например, выражение try {...} finally {...} и функция Kotlin's use обычно выполняют свои действия финализации при отмене сопрограммы:

import kotlinx.coroutines.*

fun main() = runBlocking {
//sampleStart
    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.")
//sampleEnd    
}

И join, и cancelAndJoin ожидают завершения всех действий финализации, поэтому в примере выше выводится следующий результат:

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 предыдущего примера вызывает CancellationException, поскольку сопрограмма, выполняющая этот код, отменена. Обычно это не проблема, так как все корректно работающие закрывающие операции (закрытие файла, отмена задачи или закрытие любого канала связи) обычно неблокирующие и не включают в себя приостанавливающие функции. Однако в редких случаях, когда требуется приостановление в отменённой сопрограмме, можно обернуть соответствующий код в withContext(NonCancellable) {...} с помощью функции withContext и контекстом NonCancellable, как показано в следующем примере:

import kotlinx.coroutines.*

fun main() = runBlocking {
//sampleStart
    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.")
//sampleEnd    
}

Таймаут

Наиболее очевидной практической причиной отмены выполнения корутины является превышение её времени выполнения заданного таймаута. Хотя вы можете вручную отслеживать ссылку на соответствующую задачу и запускать отдельную корутину для отмены отслеживаемой корутины после задержки, существует готовая к использованию функция withTimeout, которая выполняет эту задачу. Рассмотрим следующий пример:

import kotlinx.coroutines.*

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

Он генерирует следующий вывод:

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

Исключение TimeoutCancellationException, которое генерируется функцией withTimeout, является подклассом CancellationException. Мы ранее не видели его трассировку стека, выведенную в консоль. Это связано с тем, что внутри отменённой корутины CancellationException считается нормальной причиной завершения корутины. Однако в этом примере мы использовали withTimeout непосредственно внутри функции main.

Поскольку отмена — это просто исключение, все ресурсы закрываются обычным способом. Вы можете обернуть код с таймаутом в блок try {...} catch (e: TimeoutCancellationException) {...}, если вам нужно выполнить какие-либо дополнительные действия, связанные с таймаутом, или использовать функцию withTimeoutOrNull, которая похожа на withTimeout, но возвращает null при таймауте вместо генерации исключения:

import kotlinx.coroutines.*

fun main() = runBlocking {
//sampleStart
    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")
//sampleEnd
}

При выполнении этого кода исключений больше нет:

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

Асинхронный таймаут и ресурсы

Событие таймаута в withTimeout является асинхронным по отношению к коду, выполняемому в её блоке, и может произойти в любое время, даже непосредственно перед возвратом из блока таймаута. Имейте это в виду, если вы открываете или приобретаете какой-либо ресурс внутри блока, который необходимо закрыть или освободить вне блока.

Например, здесь мы имитируем закрываемый ресурс с классом Resource, который просто отслеживает, сколько раз он был создан, увеличивая счётчик acquired и уменьшая этот счётчик из его функции close. Давайте запустим множество корутин с небольшим таймаутом, попробуем получить доступ к этому ресурсу внутри блока withTimeout после небольшой задержки и освободить его снаружи.

import kotlinx.coroutines.*

//sampleStart
var acquired = 0

class Resource {
    init { acquired++ } // Acquire the resource
    fun close() { acquired-- } // Release the resource
}

fun main() {
    runBlocking {
        repeat(100_000) { // Launch 100K coroutines
            launch { 
                val resource = withTimeout(60) { // Timeout of 60 ms
                    delay(50) // Delay for 50 ms
                    Resource() // Acquire a resource and return it from withTimeout block     
                }
                resource.close() // Release the resource
            }
        }
    }
    // Outside of runBlocking all coroutines have completed
    println(acquired) // Print the number of resources still acquired
}
//sampleEnd

Если вы запустите указанный выше код, вы увидите, что он не всегда выводит ноль, хотя это может зависеть от времени вашей машины; вам может потребоваться настроить таймауты в этом примере, чтобы увидеть значения, отличные от нуля.

Чтобы обойти эту проблему, вы можете сохранить ссылку на ресурс в переменной вместо возврата её из блока withTimeout.

import kotlinx.coroutines.*

var acquired = 0

class Resource {
    init { acquired++ } // Acquire the resource
    fun close() { acquired-- } // Release the resource
}

fun main() {
//sampleStart
    runBlocking {
        repeat(100_000) { // Launch 100K coroutines
            launch { 
                var resource: Resource? = null // Not acquired yet
                try {
                    withTimeout(60) { // Timeout of 60 ms
                        delay(50) // Delay for 50 ms
                        resource = Resource() // Store a resource to the variable if acquired      
                    }
                    // We can do something else with the resource here
                } finally {  
                    resource?.close() // Release the resource if it was acquired
                }
            }
        }
    }
    // Outside of runBlocking all coroutines have completed
    println(acquired) // Print the number of resources still acquired
//sampleEnd
}

В этом примере всегда выводится ноль. Ресурсы не утекают.