Контекст сопрограммы и диспетчеры

Диспетчеры и потоки

Контекст сопрограммы включает в себя диспетчер сопрограммы (см. CoroutineDispatcher), который определяет, какой поток или потоки использует соответствующая сопрограмма для своего выполнения. Диспетчер сопрограммы может ограничивать выполнение сопрограммы определённым потоком, отправлять его в пул потоков или позволять ему выполняться без привязки к потоку.

Все билдеры сопрограмм, такие как launch и async, принимают необязательный параметр CoroutineContext, который можно использовать для явного указания диспетчера для новой сопрограммы и других элементов контекста.

Попробуйте следующий пример:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    launch { // context of the parent, main runBlocking coroutine
        println("main runBlocking      : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(Dispatchers.Unconfined) { // not confined -- will work with main thread
        println("Unconfined            : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(Dispatchers.Default) { // will get dispatched to DefaultDispatcher 
        println("Default               : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(newSingleThreadContext("MyOwnThread")) { // will get its own new thread
        println("newSingleThreadContext: I'm working in thread ${Thread.currentThread().name}")
    }
//sampleEnd    
}

Он выводит следующий результат (возможно, в другом порядке):

Unconfined            : I'm working in thread main
Default               : I'm working in thread DefaultDispatcher-worker-1
newSingleThreadContext: I'm working in thread MyOwnThread
main runBlocking      : I'm working in thread main

Когда launch { ... } используется без параметров, она наследует контекст (и, следовательно, диспетчер) от CoroutineScope, из которой она запускается. В этом случае она наследует контекст основной runBlocking сопрограммы, которая выполняется в main потоке.

Dispatchers.Unconfined — это специальный диспетчер, который также, по-видимому, выполняется в main потоке, но на самом деле это другой механизм, который объясняется позже.

По умолчанию используется диспетчер, если в области видимости не указан другой диспетчер. Он представлен Dispatchers.Default и использует общий фоновый пул потоков.

newSingleThreadContext создаёт поток для выполнения сопрограммы. Специализированный поток — очень дорогостоящий ресурс. В реальном приложении он должен быть либо освобождён, когда больше не нужен, с помощью функции close, либо храниться в переменной верхнего уровня и повторно использоваться на протяжении всего приложения.

Диспетчеры Unconfined и Confined

Диспетчер сопрограмм Dispatchers.Unconfined запускает сопрограмму в потоке вызывающего объекта, но только до первой точки приостановки. После приостановки он возобновляет сопрограмму в потоке, который полностью определяется выполняемой функцией приостановки, которая была вызвана. Неограниченный диспетчер подходит для сопрограмм, которые не потребляют время ЦП и не обновляют общие данные (например, UI), ограниченные определённым потоком.

С другой стороны, диспетчер по умолчанию наследуется от внешнего CoroutineScope. По умолчанию диспетчер сопрограммы runBlocking ограничен потоком вызывающего объекта, поэтому наследование этого эффекта приводит к ограничению выполнения этим потоком с предсказуемым планированием FIFO.

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    launch(Dispatchers.Unconfined) { // not confined -- will work with main thread
        println("Unconfined      : I'm working in thread ${Thread.currentThread().name}")
        delay(500)
        println("Unconfined      : After delay in thread ${Thread.currentThread().name}")
    }
    launch { // context of the parent, main runBlocking coroutine
        println("main runBlocking: I'm working in thread ${Thread.currentThread().name}")
        delay(1000)
        println("main runBlocking: After delay in thread ${Thread.currentThread().name}")
    }
//sampleEnd    
}

Выводит результат:

Unconfined      : I'm working in thread main
main runBlocking: I'm working in thread main
Unconfined      : After delay in thread kotlinx.coroutines.DefaultExecutor
main runBlocking: After delay in thread main

Таким образом, сопрограмма с контекстом, унаследованным от runBlocking {...}, продолжает выполняться в main потоке, в то время как сопрограмма с неограниченным контекстом возобновляется в фоновом потоке исполнителя по умолчанию, который используется функцией delay.

Отладка сопрограмм и потоков

Сопрограммы могут приостанавливаться в одном потоке и возобновляться в другом. Даже с диспетчером одного потока может быть сложно понять, что делала сопрограмма, где и когда, если у вас нет специальных инструментов.

Отладка с IDEA

Отладчик сопрограмм плагина Kotlin упрощает отладку сопрограмм в IntelliJ IDEA.

Окно инструментов Отладка содержит вкладку Сопрограммы. В этой вкладке можно найти информацию о текущих и приостановленных сопрограммах. Сопрограммы сгруппированы по диспетчеру, на котором они выполняются.

С помощью отладчика сопрограмм вы можете:

• Проверить состояние каждой сопрограммы.

• Просмотреть значения локальных и захваченных переменных для выполняемых и приостановленных сопрограмм.

• Просмотреть полный стек создания сопрограммы, а также стек вызовов внутри сопрограммы. Стек включает все фреймы с значениями переменных, даже те, которые были бы потеряны при стандартной отладке.

• Получить полный отчёт, содержащий состояние каждой сопрограммы и её стек. Для получения отчёта щелкните правой кнопкой мыши внутри вкладки Сопрограммы и выберите Получить дамп сопрограмм.

Для начала отладки сопрограмм достаточно установить точки останова и запустить приложение в режиме отладки.

Узнайте больше об отладке сопрограмм в учебнике.

import kotlinx.coroutines.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main() = runBlocking<Unit> {
//sampleStart
    val a = async {
        log("I'm computing a piece of the answer")
        6
    }
    val b = async {
        log("I'm computing another piece of the answer")
        7
    }
    log("The answer is ${a.await() * b.await()}")
//sampleEnd    
}

[main @coroutine#2] I'm computing a piece of the answer
[main @coroutine#3] I'm computing another piece of the answer
[main @coroutine#1] The answer is 42

Переключение между потоками

Запустите следующий код с опцией JVM -Dkotlinx.coroutines.debug (см. отладку):

import kotlinx.coroutines.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main() {
//sampleStart
    newSingleThreadContext("Ctx1").use { ctx1 ->
        newSingleThreadContext("Ctx2").use { ctx2 ->
            runBlocking(ctx1) {
                log("Started in ctx1")
                withContext(ctx2) {
                    log("Working in ctx2")
                }
                log("Back to ctx1")
            }
        }
    }
//sampleEnd    
}

Он демонстрирует несколько новых техник. Одна из них — использование runBlocking со явно заданным контекстом, а другая — использование функции withContext для изменения контекста корутины, оставаясь в той же корутине, как вы можете видеть в выводе ниже:

[Ctx1 @coroutine#1] Started in ctx1
[Ctx2 @coroutine#1] Working in ctx2
[Ctx1 @coroutine#1] Back to ctx1

Обратите внимание, что в этом примере также используется функция use из стандартной библиотеки Kotlin для освобождения потоков, созданных с помощью newSingleThreadContext, когда они больше не нужны.

Корутина Job в контексте

Корутина Job является частью её контекста и может быть извлечена из него с помощью выражения %%%CODE_BLOCK_26%%:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    println("My job is ${coroutineContext[Job]}")
//sampleEnd    
}

В режиме отладки отладка он выводит что-то вроде этого:

My job is "coroutine#1":BlockingCoroutine{Active}@6d311334

Обратите внимание, что isActive в CoroutineScope — это просто удобный ярлык для %%%CODE_BLOCK_29%%.

Дочерние корутины

Когда корутина запускается в CoroutineScope другой корутины, она наследует её контекст через CoroutineScope.coroutineContext, и Job новой корутины становится дочерним заданием для задания родительской корутины. Когда родительская корутина отменяется, все её дочерние корутины также рекурсивно отменяются.

Однако это родительско-дочернее отношение может быть явно переопределено одним из двух способов:

1. Когда при запуске корутины явно указывается другой scope (например, GlobalScope.launch), тогда она не наследует Job от родительского scope.

2. Когда в качестве контекста для новой корутины передаётся другой объект Job, тогда он переопределяет Job родительского scope.

В обоих случаях запущенная корутина не привязана к scope, из которого она была запущена, и работает независимо.

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    // launch a coroutine to process some kind of incoming request
    val request = launch {
        // it spawns two other jobs
        launch(Job()) { 
            println("job1: I run in my own Job and execute independently!")
            delay(1000)
            println("job1: I am not affected by cancellation of the request")
        }
        // and the other inherits the parent context
        launch {
            delay(100)
            println("job2: I am a child of the request coroutine")
            delay(1000)
            println("job2: I will not execute this line if my parent request is cancelled")
        }
    }
    delay(500)
    request.cancel() // cancel processing of the request
    println("main: Who has survived request cancellation?")
    delay(1000) // delay the main thread for a second to see what happens
//sampleEnd
}

Вывод этого кода:

job1: I run in my own Job and execute independently!
job2: I am a child of the request coroutine
main: Who has survived request cancellation?
job1: I am not affected by cancellation of the request

Родительские обязанности

Родительская корутина всегда ожидает завершения всех своих дочерних корутин. Родительской корутине не нужно явно отслеживать все запущенные дочерние корутины, и ей не нужно использовать Job.join для ожидания их завершения:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    // launch a coroutine to process some kind of incoming request
    val request = launch {
        repeat(3) { i -> // launch a few children jobs
            launch  {
                delay((i + 1) * 200L) // variable delay 200ms, 400ms, 600ms
                println("Coroutine $i is done")
            }
        }
        println("request: I'm done and I don't explicitly join my children that are still active")
    }
    request.join() // wait for completion of the request, including all its children
    println("Now processing of the request is complete")
//sampleEnd
}

Результат будет:

request: I'm done and I don't explicitly join my children that are still active
Coroutine 0 is done
Coroutine 1 is done
Coroutine 2 is done
Now processing of the request is complete

Именование корутин для отладки

Автоматически присваиваемые идентификаторы хороши, когда корутины часто регистрируют и вам нужно только сопоставить записи журналов, исходящие из одной корутины. Однако, когда корутина связана с обработкой конкретного запроса или выполнением какой-либо определенной фоновой задачи, лучше явно назвать её для целей отладки. Элемент контекста CoroutineName служит той же цели, что и имя потока. Оно включается в имя потока, выполняющего эту корутину, когда включен режим отладки.

Следующий пример демонстрирует эту концепцию:

import kotlinx.coroutines.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main() = runBlocking(CoroutineName("main")) {
//sampleStart
    log("Started main coroutine")
    // run two background value computations
    val v1 = async(CoroutineName("v1coroutine")) {
        delay(500)
        log("Computing v1")
        252
    }
    val v2 = async(CoroutineName("v2coroutine")) {
        delay(1000)
        log("Computing v2")
        6
    }
    log("The answer for v1 / v2 = ${v1.await() / v2.await()}")
//sampleEnd    
}

Вывод, который он производит с опцией JVM -Dkotlinx.coroutines.debug, аналогичен:

[main @main#1] Started main coroutine
[main @v1coroutine#2] Computing v1
[main @v2coroutine#3] Computing v2
[main @main#1] The answer for v1 / v2 = 42

Комбинирование элементов контекста

Иногда нам нужно определить несколько элементов для контекста корутины. Для этого можно использовать оператор +. Например, мы можем запустить корутину с явно указанным диспетчером и явно указанным именем одновременно:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    launch(Dispatchers.Default + CoroutineName("test")) {
        println("I'm working in thread ${Thread.currentThread().name}")
    }
//sampleEnd    
}

Вывод этого кода с опцией JVM -Dkotlinx.coroutines.debug:

I'm working in thread DefaultDispatcher-worker-1 @test#2

Область видимости сопрограммы

Давайте объединим наши знания о контекстах, дочерних задачах и задачах. Предположим, что наше приложение имеет объект с жизненным циклом, но этот объект не является сопрограммой. Например, мы пишем приложение для Android и запускаем различные сопрограммы в контексте активности Android для выполнения асинхронных операций, таких как получение и обновление данных, анимация и т.д. Все эти сопрограммы должны быть отменены при уничтожении активности, чтобы избежать утечек памяти. Конечно, мы можем вручную управлять контекстами и задачами, чтобы связать жизненные циклы активности и её сопрограмм, но kotlinx.coroutines предоставляет абстракцию, оборачивающую это: CoroutineScope. Вы, вероятно, уже знакомы с областью видимости сопрограммы, так как все билдеры сопрограмм объявлены как расширения для неё.

Мы управляем жизненным циклом наших сопрограмм, создавая экземпляр CoroutineScope, связанный с жизненным циклом нашей активности. Экземпляр CoroutineScope можно создать с помощью фабричных функций CoroutineScope() или MainScope(). Первая создаёт область видимости общего назначения, а вторая — область видимости для приложений с графическим интерфейсом пользователя и использует Dispatchers.Main в качестве диспетчера по умолчанию:

class Activity {
    private val mainScope = MainScope()
    
    fun destroy() {
        mainScope.cancel()
    }
    // to be continued ...

Теперь мы можем запускать сопрограммы в области видимости этого Activity с помощью определённых scope. В демонстрационном примере мы запускаем десять сопрограмм, которые ожидают разное время:

    // class Activity continues
    fun doSomething() {
        // launch ten coroutines for a demo, each working for a different time
        repeat(10) { i ->
            mainScope.launch {
                delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
                println("Coroutine $i is done")
            }
        }
    }
} // class Activity ends

В нашей главной функции мы создаём активность, вызываем нашу тестовую doSomething функцию и уничтожаем активность через 500 мс. Это отменяет все сопрограммы, запущенные из doSomething. Мы видим это, потому что после уничтожения активности больше сообщений не выводится, даже если мы подождём немного дольше.

import kotlinx.coroutines.*

class Activity {
    private val mainScope = CoroutineScope(Dispatchers.Default) // use Default for test purposes
    
    fun destroy() {
        mainScope.cancel()
    }

    fun doSomething() {
        // launch ten coroutines for a demo, each working for a different time
        repeat(10) { i ->
            mainScope.launch {
                delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
                println("Coroutine $i is done")
            }
        }
    }
} // class Activity ends

fun main() = runBlocking<Unit> {
//sampleStart
    val activity = Activity()
    activity.doSomething() // run test function
    println("Launched coroutines")
    delay(500L) // delay for half a second
    println("Destroying activity!")
    activity.destroy() // cancels all coroutines
    delay(1000) // visually confirm that they don't work
//sampleEnd    
}

Вывод этого примера:

Launched coroutines
Coroutine 0 is done
Coroutine 1 is done
Destroying activity!

Как видите, только первые две сопрограммы выводят сообщение, а остальные отменяются одним вызовом job.cancel() в Activity.destroy().

import kotlinx.coroutines.*

val threadLocal = ThreadLocal<String?>() // declare thread-local variable

fun main() = runBlocking<Unit> {
//sampleStart
    threadLocal.set("main")
    println("Pre-main, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
    val job = launch(Dispatchers.Default + threadLocal.asContextElement(value = "launch")) {
        println("Launch start, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
        yield()
        println("After yield, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
    }
    job.join()
    println("Post-main, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
//sampleEnd    
}

Pre-main, current thread: Thread[main @coroutine#1,5,main], thread local value: 'main'
Launch start, current thread: Thread[DefaultDispatcher-worker-1 @coroutine#2,5,main], thread local value: 'launch'
After yield, current thread: Thread[DefaultDispatcher-worker-2 @coroutine#2,5,main], thread local value: 'launch'
Post-main, current thread: Thread[main @coroutine#1,5,main], thread local value: 'main'