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

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

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

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

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

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

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

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
    delay(1000) // delay a second to see what happens
    println("main: Who has survived request cancellation?")
//sampleEnd
}

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

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

Ответственность родителя

Родительская сопрограмма всегда ждет завершения всех своих подзадач. Родитель не должен явно отслеживать все запущенные им дочерние сопрограммы, и ему не нужно использовать 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    
}

Выходной код с опцией -Dkotlinx.coroutines.debug JVM похож на:

[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    
}

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

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'