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

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

Контекст сопрограммы включает в себя диспетчер сопрограммы (см. 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, либо храниться в переменной верхнего уровня и повторно использоваться на протяжении всего приложения.

Неограниченный против ограниченного диспетчера

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

С другой стороны, диспетчер по умолчанию наследуется от внешнего 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` является частью её контекста и может быть извлечена с помощью выражения coroutineContext[Job].

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` — это просто удобный сокращённый вариант coroutineContext[Job]?.isActive == true.

Дочерние сопрограммы

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

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

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

2. Если в качестве контекста для новой сопрограммы передаётся другой объект `CoroutineContext` (как показано в примере ниже), то он переопределяет `CoroutineContext` родительского 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'