Elixir/OTP : Basics of Agents

Agents are abstractions built on top of GenServers, that makes it easier to create and use server processes that handle state. It abstracts away all the client-server segregation, contract callbacks and other lower level details in GenServer to provide a much cleaner and simpler API for storing, fetching and updating state.

Starting an Agent

An Agent can be started using the Agent.start/2 or Agent.start_link/2 functions. They take a no-arg anonymous function as the first argument and an optional keyword list containing options as the second argument. The anonymous function will be run in a spawned agent server process and its return value will be used as the initial state of the Agent. The difference between the start and start_link functions is that the latter creates a bidirectional process link between the caller process and the agent server process. Process linking is used for creation of supervision trees which will be discussed in detail in another article.

The second argument, which is an optional keyword list, supports all the options that a GenServer supports in its GenServer.start or GenServer.start_link function. Some of them are :name that takes in a name to register the Agent with, :timeout which is the maximum amount of time in milliseconds that the passed-in anonymous function can run to set the initial state, :debug that takes in flags to debug the agent process and :spawn_opt that takes in values to manipulate the process flags. More detailed information about these options can be obtained from here.

Once an Agent has been started successfully, the tuple {:ok, agent_pid} will be returned from the start or start_link function. The returned pid or a registered name atom can be used to identify and contact the agent server process.

{:ok, agent_pid} = Agent.start(fn -> [1] end, name: AgentServer)
{:ok, #PID<0.116.0>}

Process.whereis(AgentServer)
#PID<0.116.0>

:sys.get_state(agent_pid)
[1]

The Agent module also contains alternative functions for all APIs which take in module, function and a list of arguments instead of a single anonymous function. The provided function defined in the provided module will be executed in the spawned process with the provided arguments and the return value of this call will be set as the initial state of the agent.

{:ok, agent_pid} = Agent.start(Map, :new, [[one: 1]], name: AgentServer)
{:ok, #PID<0.117.0>}

Process.whereis(AgentServer)
#PID<0.117.0>

:sys.get_state(agent_pid)
%{one: 1}

The Agent functions with the module, function and arguments version are used mainly when the agent process and the client/caller process are not part of the same node. Using anonymous functions in distributed agents may cause issues when the agent node and the caller node contain different versions of the caller module code, especially during hot code swapping.

Fetching state

The current state of an agent can be fetched by using the Agent.get/3 function that takes in the agent pid or the registered name as its first argument, an anonymous function as its second argument and an optional timeout in milliseconds as its third argument whose default value is 5000. The anonymous function will be executed in the spawned agent process and the return value of the anonymous function will be returned to the caller process. Fetching state in agents are internally synchronous call requests, where the caller process will be blocked until it receives the result back from the agent process. If the agent process takes more time to return the result than the provided timeout value, the caller process will crash with an exit, if not properly caught.

The anonymous function passed in as the second argument, has one parameter and the agent process will pass the current state as its argument and invoke it. Hence any operation that needs to be done on the state can be performed within the anonymous function and the result can be returned from it.

{:ok, agent_pid} = Agent.start(fn -> %{one: 1} end, name: AgentServer)
{:ok, #PID<0.118.0>}

key = :one
Agent.get(AgentServer, fn state -> state[key] end)
1

The whole state can also be returned as it is from the agent to the caller process, and any processing can be done on the returned state within the caller process.

key = :one
Agent.get(AgentServer, fn state -> state end) |> Map.get(key)
1

When the state of the agent process is large, it is not efficient to return the whole state to the caller process as these processes don’t share memory. The terms sent via messages are copied and created new in the receiving processes, except for large binaries. Hence, this would lead to increased memory usage in the application if the agent state is large.

On the other hand, any process can handle only one message at a time and the rest of the messages will be in the message queue waiting to be picked up for processing. If a time-intensive operation is performed on the state within the agent process instead of within the caller process, then until the operation is complete, other requests to the agent will have to wait, leading to performance issues. Hence it is essential to consider the size of the state and the intensity of the operation to determine whether to copy the state onto the caller process and perform the processing there or to directly perform the processing on the state within the agent process.

Similar to the start and start_link functions, the get function also has an alternate version, Agent.start/5 which supports module, function and arguments instead of the anonymous function.

Updating state

The state of an agent can be updated using the Agent.update/3 function that takes in the agent pid or the registered name as its first argument, a single-arg anonymous function as its second argument and an optional timeout in milliseconds as its third argument whose default value is 5000. The anonymous function will be executed in the spawned agent process and the return value of the anonymous function will be set as the agent’s state, after which :ok is returned to the caller process. Updating state in agents are also synchronous call requests, where the caller process will be blocked until it receives :ok from the agent process. If the agent process takes more time than the provided timeout to update the state and return :ok, the caller process will crash with an exit, if not caught.

The anonymous function passed in as the second argument, has one parameter. The agent process will pass the current state as an argument and invoke the anonymous function. The return value of the anonymous function is used as the new state for the agent and then :ok is returned to the caller process.

{:ok, agent_pid} = Agent.start(fn -> %{one: 1} end, name: AgentServer)
{:ok, #PID<0.119.0>}

Agent.update(AgentServer, fn state -> Map.put(state, :two, 2) end)
:ok

Agent.get(AgentServer, &(&1))
%{one: 1, two: 2}

The function above always returns :ok and hence Agent module offers another variant, Agent.get_and_update/3, that updates the state and also returns a value to the caller instead of :ok. All the arguments are the same as the update function, but the anonymous function used here must return a two element tuple instead of the new state. The first element of the tuple should be the value that must be returned back to the caller and the second element must be the new state that must be set as the agent’s state.

{:ok, agent_pid} = Agent.start(fn -> %{x: 1} end, name: AgentServer)
{:ok, #PID<0.119.0>}

Agent.get_and_update(AgentServer, fn state -> 
    val = Map.get(state, :x, 0)
    {val + 1, Map.put(state, :x, val + 1)}
  end)
2

Agent.get(AgentServer, &(&1))
%{x: 2}

Both functions mentioned above for updating the state of an agent are synchronous. Agent module offers an asynchronous way to update state using the Agent.cast/2 function. It takes in the agent pid or a registered name as its first argument and a single-arg anonymous function as its second argument. The anonymous function is invoked in the spawned agent process with the current state as its argument, and its return value is set as the new state of the agent. But the Agent.cast function call returns :ok immediately to the caller process, instead of blocking it until the update is complete.

{:ok, agent_pid} = Agent.start(fn -> %{one: 1} end, name: AgentServer)
{:ok, #PID<0.119.0>}

Agent.cast(AgentServer, fn state -> Map.put(state, :two, 2) end)
:ok # returned immediately

Agent.get(AgentServer, &(&1))
%{one: 1, two: 2}

All the three functions mentioned above have their alternate versions, Agent.update/5, Agent.get_and_update/5 and Agent.cast/4, that take in module, function and arguments instead of an anonymous function.

Stopping the agent

An agent can be stopped explicitly by using the Agent.stop/3 function. It internally calls the GenServer.stop/3 function and takes three arguments such as the agent pid or the registered name as the first argument, an optional reason as the second argument whose default value is :normal and an optional timeout as its third argument whose default value is :infinity. It terminates the agent associated with the first argument with reason as the second argument. If a valid timeout value in milliseconds is provided, then the agent must process all remaining messages and perform proper clean up within the provided time. The agent process, if still alive, will be abruptly killed after the provided time.

{:ok, agent_pid} = Agent.start(fn -> %{one: 1} end, name: AgentServer)
{:ok, #PID<0.120.0>}

Agent.stop(AgentServer)
:ok

Process.whereis(AgentServer)
nil

If the Agent.stop/3 function call provides an explicit reason that is not :normal, :shutdown or {:shutdown, term}, then an error message will be logged for the agent’s termination. If the reason :kill is used, then the agent will be terminated abruptly.

KeyValue store example

defmodule KeyValueStore do
  def start(), do: start(%{})
  def start(state) do 
    Agent.start(fn -> state || %{} end, name: __MODULE__)
    :ok
  end

  def get(key) do
    Agent.get(__MODULE__, fn state -> state[key] end)
  end

  def put(key, value) do
    Agent.update(__MODULE__, fn state -> Map.put(state, key, value) end)
  end

  def delete(key) do
    Agent.cast(__MODULE__, fn state -> Map.delete(state, key) end)
  end
end
---------------------------------------------------------------------------
KeyValueStore.start()
:ok

KeyValueStore.put(:one, 1)
:ok

KeyValueStore.put(:two, 2)
:ok

KeyValueStore.get(:two)
2

KeyValueStore.delete(:two)
:ok

KeyValueStore.get(:two)
nil

The code above creates a simple KeyValueStore module that uses an agent to store a map and respond to requests. All the Agent api functions such as Agent.start, Agent.get, Agent.update and Agent.cast are abstracted inside module-specific functions such as start, get/1, put/2 and delete/1. You can compare the above implementation with this GenServer implementation to understand how much the Agent module abstracts away to provide a much simpler api.

use Agent

The use macro is used to inject code from a source module into destination modules during compilation. When use Agent is used in modules that internally implements an Agent, the Agent module injects an overridable default implementation for a function called child_spec/1 into the destination module.

defmodule Test do
  use Agent
end
---------------------------------------------------------------------------
defmodule Test do # injected code during compilation
  def child_spec(arg) do
    default = %{ id: __MODULE__, start: {__MODULE__, :start_link, [arg]} }
    Supervisor.child_spec(default, unquote(Macro.escape([])))
  end
end

This function is used as part of the supervision feature which will be discussed in detail in another article.