Агент

Агенты — это простое абстрагирование состояния.

Часто в Elixir возникает необходимость в совместном использовании или хранении состояния, к которому нужно получить доступ из разных процессов или одним и тем же процессом в разное время.

Модуль Agent предоставляет базовую реализацию сервера, которая позволяет получать и обновлять состояние через простой API.

Примеры

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

defmodule Mix.TasksServer do
  use Agent

  def start_link do
    Agent.start_link(fn -> MapSet.new end, name: __MODULE__)
  end

  @doc "Checks if the task has already executed"
  def executed?(task, project) do
    item = {task, project}
    Agent.get(__MODULE__, fn set ->
      item in set
    end)
  end

  @doc "Marks a task as executed"
  def put_task(task, project) do
    item = {task, project}
    Agent.update(__MODULE__, &MapSet.put(&1, item))
  end

  @doc "Resets the executed tasks and returns the previous list of tasks"
  def take_all() do
    Agent.get_and_update(__MODULE__, fn set ->
      {Enum.into(set, []), MapSet.new}
    end)
  end
end

Агенты обеспечивают разделение между клиентским и серверным API (аналогично GenServers). В частности, анонимные функции, переданные в Agent, выполняются внутри агента (сервера). Это различие важно, потому что вы можете захотеть избежать дорогостоящих операций внутри агента, так как они фактически заблокируют агент до тех пор, пока запрос не будет выполнен.

Рассмотрим эти два примера:

# Compute in the agent/server
def get_something(agent) do
  Agent.get(agent, fn state -> do_something_expensive(state) end)
end

# Compute in the agent/client
def get_something(agent) do
  Agent.get(agent, &(&1)) |> do_something_expensive()
end

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

Наконец, обратите внимание, что use Agent определяет функцию child_spec/1, позволяющую поместить определённый модуль в дерево управления. Сгенерированный child_spec/1 может быть настроен с помощью следующих параметров:

:id — идентификатор спецификации дочернего элемента, по умолчанию совпадает с текущим модулем
:start — способ запуска дочернего процесса (по умолчанию вызов __MODULE__.start_link/1)
:restart — время, когда дочерний элемент должен быть перезапущен, по умолчанию :permanent
:shutdown — способ завершения работы дочернего элемента

Например:

use Agent, restart: :transient, shutdown: 10_000

См. документацию Supervisor для получения дополнительной информации.

Регистрация имени

Агент подчиняется тем же правилам регистрации имени, что и GenServers. Подробнее об этом читайте в документации GenServer.

Несколько слов о распределённых агентах

Важно учитывать ограничения распределённых агентов. Агенты предоставляют два API: один, работающий с анонимными функциями, и другой, ожидающий явного модуля, функции и аргументов.

В распределённой настройке с несколькими узлами API, принимающий анонимные функции, работает только в том случае, если у вызывающей стороны (клиента) и агента одна и та же версия вызываемого модуля.

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

Лучшее решение — просто использовать API с явным указанием модуля, функции и аргументов при работе с распределёнными агентами.

Горячая замена кода

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

{:update, :sample, {:advanced, {Enum, :into, [%{}]}}}

Состояние агента будет добавлено в заданный список аргументов ([%{}]) в качестве первого аргумента.

Резюме

Типы

agent(): Ссылка на агента
name(): Имя агента
on_start(): Возвращаемые значения функций start*
state(): Состояние агента

Функции

cast(agent, fun): Выполняет операцию отправки (fire and forget) состояния агента
cast(agent, module, fun, args): Выполняет операцию отправки (fire and forget) состояния агента
get(agent, fun, timeout \\ 5000): Получает значение агента с помощью указанной анонимной функции
get(agent, module, fun, args, timeout \\ 5000): Получает значение агента с помощью указанной функции
get_and_update(agent, fun, timeout \\ 5000): Получает и обновляет состояние агента в одной операции с помощью указанной анонимной функции
get_and_update(agent, module, fun, args, timeout \\ 5000): Получает и обновляет состояние агента в одной операции с помощью указанной функции
start(fun, options \\ []): Запускает процесс агента без связей (вне дерева управления)
start(module, fun, args, options \\ []): Запускает агента без связей с указанным модулем, функцией и аргументами
start_link(fun, options \\ []): Запускает агент, связанный с текущим процессом, с указанной функцией
start_link(module, fun, args, options \\ []): Запускает агент, связанный с текущим процессом
stop(agent, reason \\ :normal, timeout \\ :infinity): Синхронно останавливает агент с указанным reason
update(agent, fun, timeout \\ 5000): Обновляет состояние агента с помощью указанной анонимной функции
update(agent, module, fun, args, timeout \\ 5000): Обновляет состояние агента с помощью указанной функции

Типы

agent()

agent() :: pid() | {atom(), node()} | name()

Ссылка на агента

name()

name() :: atom() | {:global, term()} | {:via, module(), term()}

Имя агента

on_start()

on_start() ::
  {:ok, pid()} |
  {:error, {:already_started, pid()} | term()}

Возвращаемые значения функций start*

state()

state() :: term()

Состояние агента

Функции

cast(agent, fun)

cast(agent(), (state() -> state())) :: :ok

Выполняет операцию отправки (fire and forget) состояния агента.

Функция fun отправляется в agent, которая вызывает функцию, передавая состояние агента. Значение возвращаемое функцией fun становится новым состоянием агента.

Обратите внимание, что cast возвращает :ok немедленно, независимо от того, существует ли agent (или узел, на котором он должен существовать).

cast(agent, module, fun, args)

cast(agent(), module(), atom(), [term()]) :: :ok

Выполняет операцию отправки (fire and forget) состояния агента.

Аналогично cast/2, но вместо анонимной функции ожидается модуль, функция и аргументы. Состояние добавляется в качестве первого аргумента к заданному списку аргументов.

get(agent, fun, timeout \\ 5000)

get(agent(), (state() -> a), timeout()) :: a when a: var

Получает значение агента с помощью указанной анонимной функции.

Функция fun отправляется в agent, которая вызывает функцию, передавая состояние агента. Результат вызова функции возвращается из этой функции.

timeout — целое число больше нуля, которое задаёт количество миллисекунд, разрешённых для выполнения функции агентом и возврата результата, или атом :infinity, чтобы ждать неопределённо долго. Если результат не получен в течение указанного времени, вызов функции завершается ошибкой, и вызывающая сторона завершает работу.

Примеры

iex> {:ok, pid} = Agent.start_link(fn -> 42 end)
iex> Agent.get(pid, fn state -> state end)
42

get(agent, module, fun, args, timeout \\ 5000)

get(agent(), module(), atom(), [term()], timeout()) :: any()

Получает значение агента с помощью указанной функции.

Аналогично get/3, но вместо анонимной функции ожидается модуль, функция и аргументы. Состояние добавляется в качестве первого аргумента к заданному списку аргументов.

get_and_update(agent, fun, timeout \\ 5000)

get_and_update(agent(), (state() -> {a, state()}), timeout()) :: a when a: var

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

Функция fun отправляется в agent, которая вызывает функцию, передавая состояние агента. Функция должна вернуть кортеж из двух элементов: первый — значение для возврата (то есть значение «получить»), а второй — новое состояние агента.

Примеры

iex> {:ok, pid} = Agent.start_link(fn -> 42 end)
iex> Agent.get_and_update(pid, fn state -> {state, state + 1} end)
42
iex> Agent.get(pid, fn state -> state end)
43

get_and_update(agent, module, fun, args, timeout \\ 5000)

get_and_update(agent(), module(), atom(), [term()], timeout()) :: any()

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

Аналогично get_and_update/3, но ожидаются модуль, функция и аргументы вместо анонимной функции. Состояние добавляется в качестве первого аргумента в предоставленный список аргументов.

start(fun, options \\ [])

start((() -> term()), GenServer.options()) :: on_start()

Запускает процесс агента без связей (вне дерева надзора).

См. start_link/2 для получения дополнительной информации.

Примеры

iex> {:ok, pid} = Agent.start(fn -> 42 end)
iex> Agent.get(pid, fn(state) -> state end)
42

start(module, fun, args, options \\ [])

start(module(), atom(), [any()], GenServer.options()) :: on_start()

Запускает агента без связей с указанным модулем, функцией и аргументами.

См. start_link/4 для получения дополнительной информации.

start_link(fun, options \\ [])

start_link((() -> term()), GenServer.options()) :: on_start()

Запускает агента, связанного с текущим процессом, с указанной функцией.

Это часто используется для запуска агента как части дерева надзора.

После создания агента, указанная функция fun вызывается, а ее возвращаемое значение используется в качестве состояния агента. Обратите внимание, что start_link/2 не возвращает значение, пока не вернется указанная функция.

Параметры

Параметр :name используется для регистрации, как описано в документации модуля.

Если параметр :timeout присутствует, агенту разрешено потратить не более указанного количества миллисекунд на инициализацию, иначе он будет завершен, и функция start вернет {:error, :timeout}.

Если параметр :debug присутствует, будет вызвана соответствующая функция в модуле :sys.

Если параметр :spawn_opt присутствует, его значение будет передано в качестве параметров подлежащему процессу, как в Process.spawn/4.

Возвращаемые значения

Если сервер успешно создан и инициализирован, функция возвращает {:ok, pid}, где pid — PID сервера. Если агент с указанным именем уже существует, функция возвращает {:error, {:already_started, pid}} с PID этого процесса.

Если заданная функция-обработчик завершается ошибкой, функция возвращает {:error, reason}.

Примеры

iex> {:ok, pid} = Agent.start_link(fn -> 42 end)
iex> Agent.get(pid, fn state -> state end)
42

iex> {:error, {exception, _stacktrace}} = Agent.start(fn -> raise "oops" end)
iex> exception
%RuntimeError{message: "oops"}

start_link(module, fun, args, options \\ [])

start_link(module(), atom(), [any()], GenServer.options()) :: on_start()

Запускает агента, связанного с текущим процессом.

Аналогично start_link/2, но вместо анонимной функции ожидаются модуль, функция и аргументы; fun в module будет вызван с указанными аргументами args для инициализации состояния.

stop(agent, reason \\ :normal, timeout \\ :infinity)

stop(agent(), reason :: term(), timeout()) :: :ok

Синхронно останавливает агента с заданным reason.

Возвращает :ok, если агент завершается с указанным причиной. Если агент завершается по другой причине, вызов завершится ошибкой.

Эта функция сохраняет семантику OTP относительно отчётов об ошибках. Если причина отлична от :normal, :shutdown или {:shutdown, _}, будет записан отчёт об ошибке.

Примеры

iex> {:ok, pid} = Agent.start_link(fn -> 42 end)
iex> Agent.stop(pid)
:ok

update(agent, fun, timeout \\ 5000)

update(agent(), (state() -> state()), timeout()) :: :ok

Обновляет состояние агента с помощью заданной анонимной функции.

Функция fun отправляется агенту agent, который вызывает функцию, передавая состояние агента. Возвращаемое значение fun становится новым состоянием агента.

Эта функция всегда возвращает :ok.

timeout — целое число больше нуля, которое указывает, сколько миллисекунд разрешено агенту выполнить функцию и вернуть результат, или атом :infinity, чтобы ждать неограниченно. Если результат не получен в течение заданного времени, вызов функции завершается ошибкой, и вызывающий код завершает выполнение.

Примеры

iex> {:ok, pid} = Agent.start_link(fn -> 42 end)
iex> Agent.update(pid, fn state -> state + 1 end)
:ok
iex> Agent.get(pid, fn state -> state end)
43

update(agent, module, fun, args, timeout \\ 5000)

update(agent(), module(), atom(), [term()], timeout()) :: :ok

Обновляет состояние агента с помощью заданной функции.

Аналогично update/3, но вместо анонимной функции ожидаются модуль, функция и аргументы. Состояние добавляется в качестве первого аргумента в предоставленный список аргументов.