Главная мысль урока: агент это ссылка, которая меняется в своём темпе, по одному действию из очереди, и не заставляет никого ждать.
future,promiseиdelayотвечают на другой вопрос: как получить значение, которого ещё нет. У каждого из четырёх свой ответ на «кто считает, в каком потоке и сколько раз».
У сервера «Хроник Пограничья» четыре заботы, и ни одна не ложится на атом:
- летописец записывает каждое событие мира в файл, и герой не должен ждать, пока диск провернётся;
- рейтинг тысячи героев считается секунду, а ответ игроку нужен сразу, рейтинг подождёт;
- гонец уехал за ответом к соседнему королевству, и квест ждёт, пока ответ доставят;
- карта мира тяжёлая, грузить её стоит один раз и только если кто-то её открыл.
В прошлом уроке ref менял несколько ссылок синхронно: dosync возвращается, когда всё уже поменялось. Здесь всё асинхронно. Первая забота решается агентом, остальные три тремя примитивами для значения из будущего.
Почти весь код урока это листинги для babashka: в ClojureScript нет потоков, а значит нет агентов, future и promise. Живым в песочнице будет только delay.
Агент: почтовый ящик летописца
У летописца есть почтовый ящик. Герои бросают туда записки и уходят по своим делам. Летописец достаёт записки по одной и вносит в хронику. Две записки он не читает одновременно, порядок записок от одного героя сохраняется, а ждать его никто не обязан.
Это агент. Записка это функция: агент применит её к своему текущему значению, и результат станет новым значением.
(def chronicle (agent []))
(send chronicle conj "Ирма вошла в Пограничье")
(send chronicle conj "Тим нашёл меч")
(println "сразу после send:" @chronicle) ; скорее всего ещё пусто
(await chronicle) ; ждём, пока очередь опустеет
(println "после await:" @chronicle)
(assert (= ["Ирма вошла в Пограничье" "Тим нашёл меч"] @chronicle))
(shutdown-agents)send вернулся мгновенно, не дожидаясь conj. Чтение @chronicle тоже не ждёт: оно покажет значение на сейчас, каким бы оно ни было. Для тестов и скриптов есть await, он блокирует поток, пока агент не выполнит всё, что ему отправили до этого момента.
В последней строке shutdown-agents. Потоки агентов и future живут в пуле, и без этой строки скрипт на JVM подождёт ещё минуту, прежде чем завершиться.
Агент похож на актора из Erlang, но с двумя отличиями. Состояние агента читается снаружи в любой момент через @. И в ящик кладут не сообщение, которое актор сам разбирает, а функцию, которую агент применяет. Подробнее про акторы в уроке про инструменты конкурентности.
send и send-off
У агентов два пула потоков, и выбор между send и send-off это выбор пула:
send | send-off | |
|---|---|---|
| Пул | фиксированный, ядра плюс два | растёт по требованию |
| Для чего | короткие вычисления в памяти | ожидание: диск, сеть, Thread/sleep |
| Чем плохо перепутать | блокирующие действия займут все потоки, и вычисления встанут | лишние потоки на быстрые задачи |
Летописец пишет на диск, значит его путь send-off:
(def chronicle-file (java.io.File/createTempFile "chronicle" ".log"))
(defn write-line [written line]
(spit chronicle-file (str line "\n") :append true)
(inc written))
(def scribe (agent 0)) ; значение агента: сколько строк записано
(doseq [event ["Ирма вошла" "Тим нашёл меч" "Ой победила волка"]]
(send-off scribe write-line event))
(await scribe)
(print (slurp chronicle-file))
(assert (= 3 @scribe))
(shutdown-agents)Здесь видно, зачем агенту очередь. Три send-off пришли почти одновременно, но строки в файле не перемешались: агент выполняет действия строго по одному. Файл стал ресурсом, у которого один хозяин.
Агент работает и вместе с транзакциями: send внутри dosync придерживается до коммита и уходит ровно один раз, сколько бы раз ни повторилось тело. Пример есть в уроке про STM. Так в Clojure принято выносить эффекты из транзакций.
Когда агент падает
Действие бросило исключение. Значение агента не меняется, а сам агент переходит в состояние сбоя: новые send бросают исключение, пока агента не перезапустят.
(def scribe (agent 0))
(send scribe (fn [_] (throw (ex-info "диск полон" {}))))
(Thread/sleep 50)
(println "ошибка:" (ex-message (agent-error scribe)))
(println (try (send scribe inc)
(catch Exception e (ex-message e)))) ; Agent is failed, needs restart
(restart-agent scribe 0 :clear-actions true)
(send scribe inc)
(await scribe)
(assert (= 1 @scribe))
(assert (nil? (agent-error scribe)))
(shutdown-agents)agent-error возвращает исключение, на котором агент упал. restart-agent задаёт новое значение и снимает сбой, :clear-actions true выбрасывает действия, накопившиеся в очереди.
Для летописца падать из-за одной строки глупо. Агенту можно дать обработчик ошибок, и тогда он переходит в режим :continue: ошибка уходит в обработчик, агент работает дальше.
(def failures (atom 0))
(def scribe (agent 0 :error-handler (fn [_agent _e] (swap! failures inc))))
(send scribe inc)
(send scribe (fn [_] (throw (ex-info "диск полон" {}))))
(send scribe inc)
(await scribe)
(println "строк:" @scribe "сбоев:" @failures "режим:" (error-mode scribe))
(assert (= 2 @scribe))
(assert (= 1 @failures))
(shutdown-agents)future: посчитать в другом потоке
Игрок открыл таблицу рекордов. Подсчёт рейтинга идёт секунду, и держать игрока всё это время незачем. future запускает тело в другом потоке и сразу возвращает ссылку на результат. Разыменовать её можно когда угодно, @ подождёт, если результат ещё не готов.
(defn compute-rating []
(Thread/sleep 300) ; притворимся, что считаем долго
[{:hero/name "Ой" :score 27} {:hero/name "Ирма" :score 18}])
(def rating (future (compute-rating)))
(println "считается?" (not (realized? rating)))
(println "ждём 50 мс:" (deref rating 50 :not-ready)) ; таймаут и запасное значение
(println "ждём сколько надо:" (map :hero/name @rating))
(assert (realized? rating))
;; Исключение из тела всплывает при deref, обёрнутым в ExecutionException.
(def broken (future (throw (ex-info "нет данных" {}))))
(println (try @broken (catch Exception e (ex-message (ex-cause e)))))
(shutdown-agents)Форма с тремя аргументами (deref ссылка таймаут-в-мс запасное-значение) работает с future и promise. Её стоит писать везде, где ожидание видит пользователь: сервер ответит «рейтинг обновляется» вместо того, чтобы висеть.
Результат future вычисляется один раз и запоминается: второй @rating вернёт тот же вектор мгновенно. Отменить вычисление можно через future-cancel, после этого @ бросит CancellationException.
Аналог из JavaScript это Promise, который начинает работу сразу при создании. Отличие в том, что @ блокирует поток, а await в JS отдаёт управление циклу событий. В JVM потоков много, и блокировка одного из них обычно не страшна.
promise: значение, которое кто-то доставит
Гонец уехал к соседям. Квест не знает, кто и когда привезёт ответ. Он знает только, что ответ однажды положат в условленное место. promise и есть это место: пустая ячейка, в которую один раз кладут значение через deliver.
(def answer (promise))
(future
(Thread/sleep 100) ; гонец в пути
(deliver answer {:kingdom "Восточная марка" :reply :alliance}))
(println "квест ждёт ответа...")
(println "ответ:" (:reply @answer))
(deliver answer {:reply :war}) ; второй гонец опоздал
(assert (= :alliance (:reply @answer)))
(shutdown-agents)Разница с future в том, кто считает. future сам запускает вычисление и сам кладёт результат. promise ничего не вычисляет, он связывает того, кто ждёт, с тем, кто когда-нибудь доставит. Поэтому promise удобен на стыке с колбэками: колбэк HTTP-клиента вызывает deliver, а код выше пишется так, будто ответ уже пришёл.
delay: посчитать один раз и только если нужно
Карта мира грузится долго. Половина игроков её не открывает. delay откладывает вычисление до первого обращения, выполняет его в том потоке, который обратился, и запоминает результат. Это работает и в ClojureScript, так что попробуй прямо здесь:
Строка «грузим карту…» напечаталась один раз, при первом @. Второй @ взял готовый результат. Если бы карта не понадобилась, она бы не загрузилась вовсе.
Если delay разыменуют из нескольких потоков сразу, тело всё равно выполнится один раз, остальные потоки подождут. Поэтому delay годится для ленивой инициализации общих ресурсов: соединения с базой, тяжёлой конфигурации.
Четыре ответа рядом
| Кто считает | В каком потоке | Сколько раз | Когда начинается | |
|---|---|---|---|---|
| агент | сам агент, по одному действию | пул send или send-off | по разу на каждое действие | при send |
future | тело future | пул потоков | один | сразу при создании |
promise | кто угодно снаружи | поток того, кто вызвал deliver | одна доставка | когда доставят |
delay | тело delay | поток первого @ | один | при первом @ |
Агент выбивается из ряда: он хранит меняющееся состояние, как атом и ref. Остальные три хранят одно значение, которое однажды появится и больше не меняется. Это значения, а не идентичности, если вспомнить урок про модель идентичности.
На временной оси разница видна так:
время ─────────────────────────────────────────────────────────►
future [создан]══ считает в пуле ══[готов] @ → сразу
promise [создан]......... ждёт .........[deliver] @ → сразу
delay [создан]............ спит ............[@]══ считает ══[готов]
агент [send]═[действие 1]═[действие 2]═[действие 3] @ → что есть сейчасpmap и его цена
Последний инструмент урока самый соблазнительный. pmap выглядит как map, но применяет функцию параллельно, пачками по числу ядер плюс два. Когда функция медленная, выигрыш виден сразу:
(defn slow-score [hero]
(Thread/sleep 100) ; запрос к внешнему сервису
(* 2 (:hero/hp hero)))
(def party (for [i (range 8)] {:hero/hp i}))
(defn elapsed-ms [f]
(let [start (System/nanoTime)]
(f)
(/ (- (System/nanoTime) start) 1e6)))
(def sequential (elapsed-ms #(doall (map slow-score party))))
(def parallel (elapsed-ms #(doall (pmap slow-score party))))
(println "map:" (Math/round sequential) "мс, pmap:" (Math/round parallel) "мс")
(assert (= (map slow-score party) (pmap slow-score party)))
(assert (< parallel (/ sequential 2)))
(shutdown-agents)Восемь вызовов по сто миллисекунд: map тратит восемьсот, pmap около ста. Но на каждый элемент pmap заводит future, а это поток из пула, очередь и синхронизация. Если функция дешёвая, например inc, накладные расходы перевесят работу, и pmap окажется медленнее обычного map. Правило: pmap для дорогих независимых вызовов, где каждый занимает хотя бы миллисекунды. Для длинных цепочек дешёвых преобразований Clojure предлагает reducers и трансдьюсеры.
И ещё одна деталь: pmap ленивый, как map. Без doall или потребителя он посчитает только первую пачку.
Проверь себя
Задание
Дальше
Агенты и future закрывают блок про состояние. Следующий блок про полиморфизм без классов, и начинается он с мультиметодов, на которых в «Хрониках» держатся правила боя. А общение потоков через каналы вместо общих ссылок ждёт тебя в уроке про core.async, там же сравнение с горутинами Go.
домашка