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

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

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

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

Существует два подхода к тому, чтобы сделать вычислительный код отменяемым. Первый — периодически вызывать приостанавливающую функцию, которая проверяет отмену. Для этой цели подходит функция 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 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. Давайте запустим большое количество корутин с небольшим таймаутом, попробуем получить доступ к этому ресурсу изнутри блока 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
}

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