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

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

В приложении с длительным выполнением вам может потребоваться точный контроль над фоновыми сопрограммами. Например, пользователь может закрыть страницу, которая запустила сопрограмму, и теперь ее результат больше не нужен, и ее работу можно отменить. Функция 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    
}

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

Такую же проблему можно наблюдать, перехватывая 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    
}

Таймаут

Самая очевидная практическая причина отмены выполнения сопрограммы — это превышение времени её выполнения над заданным таймаутом. Хотя вы можете вручную отслеживать ссылку на соответствующую задачу Job и запустить отдельную сопрограмму для отмены отслеживаемой сопрограммы после задержки, существует готовая функция 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. Теперь давайте создадим много сопрограмм, каждая из которых создаёт Resource в конце блока withTimeout и освобождает ресурс за пределами блока. Мы добавим небольшую задержку, чтобы таймаут с большей вероятностью произошёл именно тогда, когда блок 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(10_000) { // Launch 10K 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(10_000) { // Launch 10K 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
}

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