Elixir: Задачи и агенты

Для упрощения работы с процессами, в Elixir есть модули Task и Agent. Для начала познакомимся с Task.

Преимущества модуля Task в отличие от создания процессов напрямую через spawn заключаются в улучшенном выводе ошибок, а так же соответствующим API для обработки таких задач:

Task.start(fn -> raise "boom" end)
# => {:ok, #PID<0.8444.0>}

# => 16:20:15.164 [error] Task #PID<0.8444.0> started from #PID<0.105.0> terminating
# => ** (RuntimeError) boom
# =>     iex:368: (file)
# =>     (elixir 1.15.0) src/elixir.erl:374: anonymous fn/4 in :elixir.eval_external_handler/1
# => Function: #Function<43.3316493/0 in :erl_eval.expr/6>
# =>     Args: []

Task.start_link(fn -> raise "boom" end)
# => {:ok, #PID<0.8446.0>}
# проверим и очистим почтовый ящик процесса
flush()
# => {:EXIT, #PID<0.8446.0>,
# =>   {%RuntimeError{message: "boom"},
# =>     [
# =>       {:elixir_eval, :__FILE__, 1, [file: ~c"iex", line: 372]},
# =>       {:elixir, :"-eval_external_handler/1-fun-2-", 4,
# =>       [file: ~c"src/elixir.erl", line: 374, error_info: %{module: Exception}]}
# =>     ]}}

Помимо улучшенного отчета об ошибке в модуле Task есть API для асинхронной обработки задач с помощью функций Task.async() и Task.await():

tasks = 0..5 |> Enum.map(fn x -> Task.async(fn -> x * 2 end) end)
# => [
# =>   %Task{
# =>     mfa: {:erlang, :apply, 2},
# =>     owner: #PID<0.105.0>,
# =>     pid: #PID<0.8447.0>,
# =>     ref: #Reference<0.73723457.2325282820.199341>
# =>   },
# =>   %Task{
# =>     mfa: {:erlang, :apply, 2},
# =>     owner: #PID<0.105.0>,
# =>     pid: #PID<0.8448.0>,
# =>     ref: #Reference<0.73723457.2325282820.199342>
# =>   },
# =>   .........
# => ]

tasks |> Enum.map(fn task -> Task.await(task) end)
# => [0, 2, 4, 6, 8, 10]

В модуле есть и другие функции, с которыми потом можно самостоятельно ознакомиться.

Теперь рассмотрим модуль Agent подробнее. Если нужно создать процесс, который хранит состояние и это состояние необходимо передавать другим процессам или состояние используется одним процессом в течении некоторого времени - в этом случае подходит Agent. Примеры:

defmodule Counter do
  use Agent

  def start_link(initial_state \\ 0) do
    Agent.start_link(fn -> initial_state end, name: __MODULE__)
  end

  def current_value do
    Agent.get(__MODULE__, fn state -> state end)
  end

  def inc do
    Agent.update(__MODULE__, fn state -> state + 1 end)
  end

  def dec do
    Agent.update(__MODULE__, fn state -> state - 1 end)
  end
end

Counter.start_link(2)
# => {:ok, #PID<0.115.0>}

Counter.current_value()
# => 2

Counter.inc()
# => :ok
Counter.current_value()
# => 3

Counter.dec()
# => ok
Counter.current_value()
# => 2

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

В функции start_link происходит глобальная регистрация процесса-агента и его связывание с процессом из которого вызвана функция, подобно функции spawn_link. Про регистрацию процессов будет рассказано в следующих упражнениях.

В функции current_value происходит обращение к внутреннему состоянию процесса и возврат этого состояния наружу.

В функциях inc и dec изменяется состояние процесса-агента, увеличиваясь или уменьшаясь на единицу соответственно.

Модуль Agent предполагает разделение на клиентскую и серверную часть. Функции, переданные в качестве аргументов для модуля Agent вызываются внутри агента (серверная часть). В ином случае вызов кода считается клиентской частью. Рассмотрим подробнее на примере:

# вычисление произойдет на стороне процесс-агента, то есть на серверной части
def do_something(agent) do
  Agent.get(agent, fn state -> run_slow_code(state) end)
end

# вычисление произойдет на стороне процесса, вызвавшего функцию, то есть на клиентской части
def do_something(agent) do
  Agent.get(agent, fn state -> state end) |> run_slow_code()
end

Первая функция блокирует работу агента. Вторая функция копирует все состояние на клиента и затем выполняет операцию на клиенте. При этом следует учитывать, достаточно ли велики данные, чтобы выполнить их обработку на сервере, или их можно быстро переслать клиенту. Другой фактор - нужно ли обрабатывать данные атомарно: получение состояния и вызов run_slow_code(state) вне агента означает, что состояние агента может быть обновлено в это время. Это особенно важно в случае обновлений, поскольку вычисление нового состояния на клиенте, а не на сервере может привести к условиям гонки, если несколько клиентов будут пытаться обновить одно и то же состояние на разные значения.

В модуле Agent есть и другие функции, которые потом можно самостоятельно изучить.

После объявления модуля агента, в нем еще появляется функция child_spec/1, про ее смысл поговорим дальше, в рамках дерева супервизии процессов.

В этот раз допишите функции агента Accumulator, используя модуль Task и опираясь на модуль Calculator:

• add прибавить к состоянию аккумулятора переданное число;
• sub вычесть состояние аккумулятора на переданное число;
• mul умножить состояние аккумулятора на переданное число;
• div разделить состояние аккумулятора на переданное число;
• reset сбрасывает состояние Accumulator в ноль;
• current возвращает нынешнее состояние Accumulator.

Accumulator.start_link(0)

Accumulator.add(10) # => :ok
Accumulator.current() # => 10

Accumulator.sub(2) # => :ok
Accumulator.current() # => 8

Accumulator.mul(2) # => :ok
Accumulator.current() # => 16

Accumulator.div(4) # => :ok
Accumulator.current() # => 4

Accumulator.reset() # => :ok
Accumulator.current() # => 0