Kotlin: про Structured Concurrency, CoroutineContext и CoroutineScope

В прошлой статье разбирались с проблематикой, которую решает Kotlin. Теперь разберем какие идеи заложены при реализации асинхронности в Kotlin.

Вспомним ещё раз чем отличается настоящий параллелизм (parallel) от конкурентного(concurrent):

При настоящем параллелизме задачи выполняются на разных потоках, управляемых планировщиком операционной системой.

При конкурентном параллелизме общая задача разбивается на небольшие блоки-задания, управление выполнением которых берет на себя планировщик среды исполнения (не OS!!), который сам является частью программы. Блоки-задания, относящиеся к одной задаче могут выполнится на разных потоках операционной системы.

Structured Concurrency

В основе механизма управления корутин в Kotlin лежит подход, с подачи Мартина Сустрика получивший название — парадигма программирования Structured Concurrency (*).

Согласно парадигме Structured Concurrency:

• рабочие задания упорядочиваются в древовидную структуру по отношению родитель — потомок (для Kotlin — рабочие задания это корутины)

• жизненный цикл дочернего задания строго связан с жизненным циклом родительского задания таким образом, что:

  • родительское задание ожидает завершения (успешного или с ошибкой) всех дочерних заданий

  • отмена родительского задания приводит к завершению всех дочерних заданий

  • ошибка при выполнении дочернего задания пробрасывается в родительское задание

* идею витали в воздухе и к моменту, когда Мартин Сустрик сформулировал свой принцип, разработчики Kotlin уже пришли к аналогичному решению и даже его реализовали в корутинах. Но само название «Structured Concurrency» оказалось удачным и закрепилось в документации.

CoroutineContext

В основе реализации механизма корутин лежит интерфейс CoroutineContext и набор базовых реализаций этого интерфейса (AbstractCoroutineContextElement, EmptyCoroutineContext, CombinedContext).

Все вместе (интерфейс CoroutineContext и базовые реализации), с одной стороны, реализуют паттерн компоновщик, а с другой, благодаря возможностям Kotlin, позволяют работать с контекстом и объектами, которые в нем лежат через операторы ‘+’, ‘[ ]‘.

Если ваши классы реализуют интерфейс CoroutineContext.Element, то можно использовать CoroutineContext для компоновки собственных элементов в свой отдельный контекст, не привязываясь к контексту корутин.

class AnyContextElement(val name: String) : CoroutineContext.Element {    override val key: CoroutineContext.Key<*> get() = Key    companion object Key : CoroutineContext.Key<AnyContextElement>}class OtherContextElement : CoroutineContext.Element {    override val key: CoroutineContext.Key<*> get() = Key    companion object Key : CoroutineContext.Key<OtherContextElement>}// Создаем собственный контекст с элементом AnyContextElement("name1") var customContext = EmptyCoroutineContext + AnyContextElement("name1")println(customContext[AnyContextElement]?.name) // Вывод: name1// Добавили к контексту ещё одни элементcustomContext += OtherContextElement()// Заменили в контексте AnyContextElement другим объектомcustomContext +=  AnyContextElement("name2")println(customContext[AnyContextElement]?.name) // Вывод: name2

CoroutineScope

CoroutineScope является техническим интерфейсом, хранящем в своем контексте объекты, определяющую работу корутин.

public interface CoroutineScope {    public val coroutineContext: CoroutineContext}public fun CoroutineScope(context: CoroutineContext): CoroutineScope =    ContextScope(if (context[Job] != null) context else context + Job())

CoroutineScope интересен двумя вещами:

• содержит coroutineContext, который объединяет объекты, определяющие поведение корутин (но не только), в частности, хотя бы объект типа Job

• содержит extension-функции для запуска дочерних корутин — launch и async

Текущий контекст можно получить через вызов функции currentCoroutineContext().

Любая корутина может быть запущена только в рамках какого-то CoroutineScope.

Точкой входа в механизм корутин является функция runBlocking.

fun main() = runBlocking {    // объект this указывает на текущий CoroutineScope и можно вызывать     // extension-функции    launch {        ...    }    }

При создании новой корутины создается новый coroutineContext на базе родительского coroutineContext. При этом создается новый объект Job, дочерний по отношению к Job родительской корутины. (**)

val scope = CoroutineScope(EmptyCoroutineContext)val job1 = scope.launch { ... }val job2 = scope.launch { ... }val job3 = scope.launch { ... }

** в runtime объект, доступный через вызов coroutineContext[Job], и есть текущая корутина; конкретный класс объекта корутины наследуется от класса AbstractCoroutine, реализующего интерфейсы Job и CoroutineScope

Объект Job в coroutineContext обеспечивает реализацию механизма Structured Concurrency и, в частности, позволяет останавливать выполнение дочерних корутин.

val scope = CoroutineScope(EmptyCoroutineContext)val job1 = scope.launch { ... }val job2 = scope.launch { ... }val job3 = scope.launch { ... }scope.cancel() // Вызывается scope.coroutineContext[Job].cancel()

Родительский Job не будет завершен пока не выполнятся все дочерние корутины.

Важно понимать что принцип Structured Concurrency относится к корутинам, запущенным в рамках одной иерархии корутин (одного сoroutineContext).

Structured Concurrency на примерах

Пример 1

// runBlocking - точка входа в мир корутин; создает свой scoperunBlocking {  // this - ссылка на scope runBlocking  // корутины запускаются в scope runBlocking  launch { while(true) {  delay(2000)  } } // this.launch  launch { while(true) {  delay(3000)  } }  launch { while(true) {  delay(3000)  } }  // ВИСИМ !!!}

Корутины, запускаемые через launch корутины являются дочерними по отношению к корутине выполняющей lambda-функцию, переданную в runBlocking. Родительская корутина ожидает завершения работы дочерних корутин.

Пример 2

runBlocking {  val scope = CoroutineScope(EmptyCoroutineContext)  scope.launch { while(true) {  delay(2000)  } }  scope.launch { while(true) {  delay(3000)  } }  scope.launch { while(true) {  delay(3000)  } }  // выходим из runBlocking не дожидаясь завершения корутин}

В данном примере в методе runBlocking создается новый scope. Новый scope и scope runBlocking не связаны между собой.

Пример 3

runBlocking {  val scope = CoroutineScope(EmptyCoroutineContext) + coroutineContext  scope.launch { while(true) {  delay(2000)  } }  scope.launch { while(true) {  delay(3000)  } }  scope.launch { while(true) {  delay(3000)  } }  // ВИСИМ!!!}

В примере 3 несмотря на то, что создается новый scope, используется общий с runBlocking контекст. Корутина, выполняемая в runBlocking является родительской по отношению к корутинам запущенным через scope.launch и ожидает завершения дочерних корутин.

Пример 4

val scope1 = CoroutineScope(EmptyCoroutineContext)val job = scope1.launch {    val scope2 = CoroutineScope(EmptyCoroutineContext)    val scope2Job = scope2.launch {  ... }    val job1 = launch {  ...  }    val job2 = launch {  ...  }    val job3 = launch {  ...  }}job.cancel() 

В примере 4 при вызове job.cancel будут остановлены корутины job1, job2, job3; Корутина scope2Job продолжит работу.

Пример 5

runBlocking {    doFn() // запускается на scope runBlocking    // Висим!!!}suspend fun doFn() {    val currentJob = currentCoroutineContext()[Job]    (currentJob as? CoroutineScope)?.launch {        while(true) {  delay(2000)  }    }}

Пример 5 демонстрирует, что объект, представляющий текущую корутину можно получить через контекст и то, что конкретный класс объекта корутины реализует интерфейс CoroutineScope и связан с контекстом runBlocking.

Пример 6

runBlocking {    launch {        doFn()    }    // ВИСИМ !!!}suspend fun doFn() = withContext(EmptyCoroutineContext) {  launch {    while(true) {  delay(2000)  }  }}

В примере 6 несмотря на то, что в withContext передается новый контекст(EmptyCoroutineContext), но в реализации withContext контексты суммируются и полученная в результате корутина принадлежит иерархии корутин runBlocking.

Пример 7

suspend fun doFn() = withContext(Job()) {    launch {        while(true) {  delay(2000)  }    }}

Исходя из вышеизложенного становится понятно почему при попытке передать в builder-функцию объект Job или контекст, содержаший Job можно увидеть предупреждение:

Passing ‘CoroutineContext’ with a ‘Job’ to ‘withContext’ builder can lead to structured concurrency violations

Пример 8 — демонстрирует доступ к иерархии корутин

scope.coroutineContext[Job]?.children?.forEach { job ->   // launched job   println(job)   // grandchildren   job.children.forEach {       println(job)   }}

Заключение

В основе механизма управления корутин лежит принцип получивший название Structured Concurrency.

Применительно к Кotlin подход реализуется следующим образом:

1. корутины упорядочиваются в древовидную иерархию по отношению родитель-потомок

2. родительская корутина дожидается завершения дочерних корутин

3. остановка родительской корутины приводит к остановке дочерних корутин

4. за хранение объектов, определяющий работу корутин отвечает контекст корутины — coroutineContext

5. текущий контекст можно получить через метод currentCoroutineContext()

6. управлять корутинами можно через объект типа Job, который всегда есть в контексте корутины и который можно получить coroutineContext[Job]

7. при создании корутины происходит объединение контекстов — контекста родительской корутины и дополнительного контекста, созданного на основе параметров, переданных в builder-функцию

8. корректная иерархия корутин может быть выстроена только если при создании дочерней корутины дополнительный контекст не содержит элемент Job; в контексте дочерней корутины будет содержаться Job, дочерний по отношению к Job родительского контекста

Ссылки и благодарности

• Coroutine scope and structured concurrency

• Roman Elizarov — Structured concurrency

• Шаблон проектирования: Composite

• Structured concurrency paradigm

• Kotlin Coroutines под капотом: CoroutineContext и CoroutineScope

• Custom Kotlin Coroutine Context Uses Cases

• Kotlin Coroutines

• Coroutines Job Structures

• Sústrik, Martin: Structured Concurrency