Асинхронное программирование

Задачу (Task) можно представить как дескриптор для некоторой единицы вычислительной работы, которую необходимо выполнить. Ее жизненный цикл состоит из этапов создания, запуска, выполнения и завершения. Задачи создаются путем вызова конструктора Task для функции без аргументов, которую необходимо выполнить, или с помощью макроса @task:

@task x эквивалентно Task(()->x).

Эта задача ожидает пять секунд, а затем выводит done. Однако она еще не начала выполняться. Ее можно запустить в любой момент, вызвав функцию schedule:

Если вы попробуете сделать это в REPL, то увидите, что функция schedule немедленно возвращает управление. Причина в том, что она просто добавляет t во внутреннюю очередь задач, подлежащих выполнению. Затем в REPL выводится следующее приглашение к вводу данных. Пока ожидается ввод данных с клавиатуры, могут выполняться другие задачи, поэтому запускается задача t. t вызывает функцию sleep, которая устанавливает таймер и прекращает выполнение. Если запланированы другие задачи, они могут выполняться в это время. Через пять секунд таймер срабатывает и снова запускает задачу t, которая выводит done. После этого задача t завершается.

Функция wait блокирует выполнение вызвавшей ее задачи до тех пор, пока не завершится какая-либо другая задача. Например, если ввести,

а не просто вызвать schedule, следующее приглашение к вводу появится через пять секунд. Причина в том, что REPL ожидает завершения задачи t, прежде чем продолжить работу.

Часто бывает нужно создать задачу и сразу запланировать ее к выполнению. Для этого предусмотрен макрос @async: вызов @async x эквивалентен вызову schedule(@task x).

В ряде ситуаций единицы работы не связаны с вызовами функций: среди задач, подлежащих выполнению, нет явных вызывающих или вызываемых объектов. Примером может служить сценарий «источник-потребитель», когда одна сложная процедура генерирует значения, а другая — использует их. Потребитель не может просто вызвать функцию-источник для получения значения, так как источник, возможно, еще не создал все значения и не готов их предоставить. При использовании задач источник и потребитель могут выполняться столько, сколько нужно, обмениваясь данными по мере необходимости.

Для решения этой проблемы в Julia имеется механизм каналов (Channel). Channel — это очередь ожидания, работающая по принципу «первым поступил — первым обслужен»: в нее могут записывать данные или считывать данные из нее сразу несколько задач.

Давайте определим задачу-источник, которая создает значения посредством вызова метода put!. Для получения значений необходимо запланировать выполнение потребителя в новой задаче. Для запуска задачи, привязанной к каналу, можно использовать специальный конструктор Channel, который принимает в качестве аргумента функцию с одним аргументом. Затем можно многократно принимать значения из канала с помощью метода take!:

Источник (producer) словно бы способен многократно возвращать управление. Между вызовами put! выполнение источника приостанавливается, и управление получает потребитель.

Возвращаемый канал Channel можно использовать как итерируемый объект в цикле for. В этом случае переменная цикла принимает все полученные значения. Цикл завершается после закрытия канала.

Обратите внимание: нам не пришлось явным образом закрывать канал в источнике. Причина в том, что при привязке объекта Channel к объекту Task время открытия канала будет зависеть от привязанной задачи. Канал автоматически закрывается при завершении задачи. К одной задаче можно привязать несколько каналов, и наоборот.

Конструктор Task принимает функцию без аргументов, в то время как метод Channel, создающий привязанный к задаче канал, принимает функцию с одним аргументом типа Channel. Стандартным подходом является параметризация источника. В этом случае требуется применение частичной функции для создания анонимной функции без аргументов или с одним аргументом.

Для объектов Task это можно делать напрямую или с помощью вспомогательного макроса:

Для организации более сложных шаблонов распределения работы в сочетании с конструкторами Task и Channel можно использовать функции bind и schedule, явным образом связывая набор каналов с набором задач-источников и задач-потребителей.

Операции с задачами основаны на низкоуровневом примитиве yieldto. Вызов yieldto(task, value) приостанавливает текущую задачу, выполняет переключение на задачу, указанную в аргументе task, и предписывает последнему вызову yieldto этой задачи вернуть указанное в value значение. Обратите внимание: yieldto — это единственная операция, которая должна использовать порядок выполнения в стиле задач; вместо вызова и возврата управления мы просто переключаемся на другую задачу. Вот почему задачи также называют «симметричными сопрограммами»: переключение на каждую задачу и из нее происходит с использованием одного и того же механизма.

yieldto — это очень эффективная функция, но при использовании задач она редко вызывается напрямую. Разберем возможные причины. Если вы переключаетесь из текущей задачи, то, вероятно, через какое-то время может потребоваться снова переключиться на нее. Но для определения нужного момента переключения, а также задачи, которая отвечает за такое переключение, требуется строгое согласование действий. Например, методы put! и take! блокируют операции, состояние которых в контексте использования каналов сохраняется, чтобы можно было определить потребители. Именно отсутствие необходимости отслеживать задачу-потребитель вручную делает метод put! проще в использовании по сравнению с низкоуровневой функцией yieldto.

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

Чаще всего переключения задач происходят в результате ожидания событий, например запросов ввода-вывода, и производятся планировщиком, входящим в состав модуля Base в Julia. Планировщик ведет очередь задач, которые могут быть запущены, и выполняет цикл событий, который перезапускает задачи при наступлении внешних событий, например поступлении сообщения.

Базовая функция для ожидания события — wait. Функцию wait реализует ряд объектов. Например для объекта Process функция wait ожидает завершения его выполнения. Функция wait часто неявная; например, wait может выполняться внутри вызова read для ожидания доступности данных.

Во всех этих случаях функция wait в конечном итоге работает с объектом Condition, который отвечает за постановку задач в очередь и их перезапуск. Когда задача вызывает функцию wait при наступлении условия Condition, она помечается как не подлежащая запуску, добавляется в очередь условия и переключается на планировщик. После этого планировщик выбирает другую задачу для запуска или блокирует ожидание внешних событий. Если все в порядке, обработчик события в конечном итоге вызывает функцию notify для условия, так что ожидающие его выполнения задачи снова становятся доступны для запуска.

Задача, созданная явным образом путем вызова Task, изначально неизвестна планировщику. Это позволяет при необходимости управлять задачами вручную с помощью yieldto. Однако, когда такая задача ожидает событие, она все же перезапускается автоматически при его наступлении, как и следовало ожидать.