Coroutine Context and Dispatchers — Kotlin 1.4

Содержание

Контекст и диспетчеры сопроцедур

Контекст и диспетчеры сопроцедур

Сопроцедуры всегда выполняются в некотором контексте, представленном значением типа CoroutineContext в стандартной библиотеке Kotlin.

Контекст сопроцедуры представляет собой набор различных элементов. Основные элементы — это Задача сопроцедуры, которую мы видели ранее, и её диспетчер, который рассматривается в этом разделе.

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

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

По умолчанию диспетчер, используемый при запуске сопроцедур в GlobalScope, представлен Dispatchers.Default и использует общий пул фоновых потоков, поэтому launch(Dispatchers.Default) { ... } использует тот же диспетчер, что и GlobalScope.launch { ... }.

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.

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

Debugging coroutines

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

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

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

Подробнее об отладке сопроцедур см. в учебнике.

Отладка с помощью логгирования

Другой подход к отладке приложений с потоками без отладчика сопроцедур — выводить имя потока в лог-файл в каждой строке лог-вывода. Эта функция поддерживается всеми фреймворками логгирования. При работе с сопроцедурами только имя потока не даёт контекста, поэтому kotlinx.coroutines включает средства отладки, чтобы упростить эту задачу.

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

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    
}

Существует три сопрограммы. Основная сопрограмма (#1) внутри runBlocking и две сопрограммы, вычисляющие отложенные значения a (#2) и b (#3). Все они выполняются в контексте runBlocking и ограничены главной нитью. Вывод этого кода:

[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

Функция log выводит имя потока в квадратных скобках, и вы можете видеть, что это поток main с идентификатором выполняемой в данный момент сопрограммы, добавленным к нему. Этот идентификатор последовательно назначается всем созданным сопрограммам, когда включён режим отладки.

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

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

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

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, one with GlobalScope
        GlobalScope.launch {
            println("job1: I run in GlobalScope 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 GlobalScope 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    
}

Вывод, который он генерирует с опцией 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().

Данные потока

Иногда удобно передавать данные текущего потока в корутины или между ними. Однако, так как корутины не привязаны к какому-либо конкретному потоку, это может привести к избыточному коду, если делать это вручную.

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

Это легко продемонстрировать на практике:

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    
}

В этом примере мы запускаем новую корутину в пуле фоновых потоков, используя Dispatchers.Default, поэтому она работает в другом потоке, но при этом сохраняет значение локальной переменной потока, которое мы задали с помощью threadLocal.asContextElement(value = "launch"), независимо от того, в каком потоке выполняется корутина. Таким образом, вывод (с отладкой) выглядит следующим образом:

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'

Легко забыть установить соответствующий элемент контекста. Тогда локальная переменная потока, к которой обращается корутина, может иметь неожиданное значение, если поток, выполняющий корутину, отличается. Для предотвращения таких ситуаций рекомендуется использовать метод ensurePresent и выполнять быструю проверку на некорректное использование.

ThreadLocal имеет поддержку первого класса и может использоваться с любыми примитивными типами kotlinx.coroutines предоставляет. Однако, у него есть одно ключевое ограничение: когда локальная переменная потока изменяется, новое значение не передаётся обратно вызывающей корутине (потому что элемент контекста не может отслеживать все ThreadLocal обращения к объекту), и обновлённое значение теряется при следующем приостановлении. Используйте withContext для обновления значения локальной переменной потока в корутине, см. asContextElement для получения более подробной информации.

В качестве альтернативы, значение можно хранить в изменяемой переменной, такой как class Counter(var i: Int), которая, в свою очередь, хранится в локальной переменной потока. Однако, в этом случае вы полностью несёте ответственность за синхронизацию потенциально одновременных изменений переменной в этой изменяемой переменной.

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

© 2010–2020 JetBrains s.r.o. and Kotlin Programming Language contributors
Licensed under the Apache License, Version 2.0.
https://kotlinlang.org/docs/reference/coroutines/coroutine-context-and-dispatchers.html