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

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

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

Таймаут

Самая очевидная практическая причина отмены выполнения сопрограммы — превышение времени выполнения заданного таймаута. Хотя вы можете вручную отслеживать ссылку на соответствующую задачу 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(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
}

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