Событийный цикл, каналы и таймауты

Файберы мы переключали руками. Поверх них Janet даёт планировщик, который делает это сам. Весь урок держится на одном правиле: переключение происходит только в точках ожидания. Из этого правила выведем и поведение каналов, и работу дедлайнов, и отмену задач.

ev/spawn и планировщик

ev/spawn ставит задачу в очередь и немедленно возвращает управление:

(def order @[])

(ev/spawn
  (ev/sleep 0.01)
  (array/push order :background))

(array/push order :main)
(ev/sleep 0.05)

order                # @[:main :background]

Порядок здесь показателен: ev/spawn не выполняет тело сразу, а планирует его. Основной код продолжается, и фоновая задача получает управление только когда основная сама уступит, на ev/sleep или любой другой операции ожидания.

Это не потоки. Всё выполняется в одном системном потоке. Переключение происходит только в точках ожидания, поэтому гонок за данные в привычном смысле нет: между двумя точками ожидания твой код выполняется атомарно. Это кооперативная многозадачность, ровно как в событийном цикле Node.js.

Обратная сторона та же: вычислительная задача без точек ожидания заблокирует всё. Для настоящей параллельности нужны отдельные системные потоки, но это тема за пределами урока.

Каналы

Задачи общаются через каналы:

(def ch (ev/chan 10))     # канал с буфером на 10 значений
(ev/give ch :value)       # положить
(ev/take ch)              # взять

Канал без аргумента небуферизованный: ev/give блокируется, пока кто-нибудь не сделает ev/take. Так синхронизируют производителя с потребителем. Модель та же, что у каналов Go, мы разбирали её в уроке про горутины и каналы.

Заметь, как здесь работает правило урока: ev/give и ev/take это и есть точки ожидания. Именно на них планировщик переключает задачи, поэтому каналы не просто передают данные, они задают ритм всей конкурентности.

Ловушка: закрытие канала отбрасывает буфер

Сначала посмотри на код:

(def ch (ev/chan 10))
(ev/give ch :a)
(ev/give ch :b)
(ev/count ch)        # 2, оба на месте

(ev/chan-close ch)
(ev/take ch)         # nil, значения потеряны
(ev/count ch)        # 2, счётчик всё ещё показывает два

А теперь потрогай то же самое руками: положи несколько значений и закрой канал, не забрав их:

Это тихая потеря данных, которую легко не заметить. Типичный код, где производитель кладёт значения и закрывает канал, а потребитель читает до nil, соберёт лишь часть значений. Или вообще ничего, если производитель успеет закрыть канал раньше первого чтения.

Три способа не потерять данные

Небуферизованный канал. ev/give ждёт ev/take, поэтому к моменту закрытия всё уже прочитано:

(def ch (ev/chan))                        # без буфера
(ev/spawn
  (each i [1 2 3] (ev/give ch i))
  (ev/chan-close ch))

(def acc @[])
(while (def v (ev/take ch)) (array/push acc v))
acc                  # @[1 2 3]

Значение-сентинел вместо закрытия:

(ev/spawn
  (each i [1 2 3] (ev/give ch i))
  (ev/give ch :done))

Фиксированное число чтений, когда количество результатов известно заранее:

(def ch (ev/chan 10))
(each i [1 2 3]
  (ev/spawn (ev/sleep (* 0.001 i)) (ev/give ch (* i 10))))

(def acc @[])
(repeat 3 (array/push acc (ev/take ch)))
(sort acc)           # @[10 20 30]

Выбор и таймауты

ev/select ждёт первый готовый канал из нескольких:

(def fast (ev/chan 1))
(def slow (ev/chan 1))

(ev/spawn (ev/sleep 0.001) (ev/give fast :quick))
(ev/spawn (ev/sleep 0.2) (ev/give slow :slowly))

(ev/select fast slow)
# (:take <канал> :quick)

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

Проигравшую задачу можно снять с планировщика. ev/cancel доставляет файберу ошибку прямо в его точку ожидания:

(def task (ev/spawn (ev/sleep 10) (print "не напечатается")))
(ev/cancel task "хватит")

Задача просыпается в своей точке ожидания уже с ошибкой, тело дальше не выполняется, статус становится :error. Это ровно то, что понадобится в упражнении e13.5: победитель найден, проигравшего отменяем.

ev/with-deadline ограничивает время выполнения блока:

(ev/with-deadline 1 (ev/sleep 0.001) :made-it)   # :made-it

(try
  (ev/with-deadline 0.05 (ev/sleep 10) :no-luck)
  ([err] err))                                    # "deadline expired"

По истечении срока блок прерывается ошибкой, поэтому его оборачивают в try.

Дедлайн это не таймер прерывания. Под капотом он устроен как та же отмена: ошибка доставляется в точке ожидания. Вычислительный цикл без таких точек дедлайн переживёт:

(ev/with-deadline 0.05
  (while true))   # висит навсегда: внутри ни одной точки ожидания

Правило урока работает и против тебя: планировщик получает управление только там, где задача его отдаёт, и прервать невежливую задачу ему нечем.

Таймауты в тестах обязательны. Любой тест, который ждёт канал или сетевую операцию, должен иметь предел ожидания. Без него зависший тест не падает, а висит, и подвешивает CI до общего таймаута сборки. Это правило мы вводили в разделе про тестирование, и здесь оно особенно уместно.

Что запомнить

• ev/spawn планирует задачу, а не выполняет её сразу.
• Всё в одном системном потоке, переключение только в точках ожидания.
• Гонок между точками ожидания нет, но задача без ожиданий блокирует всех.
• ev/chan с аргументом это буфер, без аргумента синхронный канал.
• ev/chan-close отбрасывает буфер молча, а ev/count при этом врёт.
• Безопасные идиомы: небуферизованный канал, сентинел, фиксированное число чтений.
• ev/select берёт первый готовый канал, ev/cancel снимает задачу, доставляя ей ошибку в точке ожидания.
• ev/with-deadline ограничивает время и падает ошибкой, но сработать может только в точке ожидания.

Упражнения

Дальше

Конкурентность закрыта. Следующий урок возвращается к основам языка и разбирает управление потоком выполнения целиком: все формы ветвления, сопоставление с образцом, универсальный цикл и нелокальные переходы.