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

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

https://kotlinlang.org/docs/coroutines-basics.html

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

При конкурентном параллелизме общая задача разбивается на небольшие блоки-задания, управление выполнением которых берет на себя планировщик среды исполнения (не 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 - точка входа в мир корутин; создает свой scope
runBlocking {
  // 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 подход реализуется следующим образом:

корутины упорядочиваются в древовидную иерархию по отношению родитель-потомок
родительская корутина дожидается завершения дочерних корутин
остановка родительской корутины приводит к остановке дочерних корутин
за хранение объектов, определяющий работу корутин отвечает контекст корутины - coroutineContext
текущий контекст можно получить через метод currentCoroutineContext()
управлять корутинами можно через объект типа Job, который всегда есть в контексте корутины и который можно получить coroutineContext[Job]
при создании корутины происходит объединение контекстов - контекста родительской корутины и дополнительного контекста, созданного на основе параметров, переданных в builder-функцию
корректная иерархия корутин может быть выстроена только если при создании дочерней корутины дополнительный контекст не содержит элемент Job; в контексте дочерней корутины будет содержаться Job, дочерний по отношению к Job родительского контекста

Structured Concurrency

CoroutineContext

CoroutineScope

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

Заключение

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

Kotlin: про Structured Concurrency, CoroutineContext и CoroutineScope