wktdev
Looking for clarity around using agent
In the following code I use agent to “save state” of a list and then I update the list.
(I know elixir does not have objects but naming my functions as such helps me reason about the code )
defmodule M do
def create_object() do
{:ok, pid} = Agent.start_link(fn -> [1, 2, 3] end)
pid
end
def update_object(pid, new_data) do
Agent.update(pid, fn (state) -> state ++ new_data end)
pid
end
def get_object(pid) do
IO.inspect Agent.get(pid, &(&1))
end
end
M.create_object() |> M.update_object([4, 5]) |> M.get_object()
The following code has the same result but does not use agent. What can the code that does use agent do that the following code can not? Asking this question is my attempt to see what I am not understanding.
defmodule M do
def create_object() do
obj = [1, 2, 3]
obj
end
def update_object(obj, new_data) do
obj = obj ++ new_data
obj
end
def get_object(obj) do
IO.inspect obj
obj
end
end
M.create_object() |> M.update_object([4, 5]) |> M.get_object()
Most Liked
hubertlepicki
The main difference is that Agent (or GenServer) runs in it’s own process.
This means that knowing the pid of Agent (or it’s registered name) you can retrieve/save state from different concurrent processes.
You can use Agent to say implement a dynamic configuration for the application. All HTTP requests are handled in their own processses, but they would be able to access this shared configuration. Then, you could update the configuration say from iex shell and it would be picked up by all other processes now on.
peerreynders
Hubert’s reply is correct
I’d like to explore some aspects in more depth and detail.
First of all many of us are still scratching our head in terms of finding valid use cases for Agents as evidenced by this topic:
Discussion about uses for Agent Processes
Hubert’s use case of changing the configuration state of a running system through the shell is a viable one - though I would imagine that sending a function to update the state is inherently more risky than simply replacing the old state with a new, known to be consistent, state.
Then there is the choice of your example - it is inherently sequential. I recommend that you watch:
Erlang Master Class 2: Video 1 - Turning sequential code into concurrent code
Your example forces a particular ordering on the sequence of operations which doesn’t take advantage of the capabilities of the Agent - there are no independent, concurrent parts in your computation.
Your use of an Agent can be roughly expressed as a GenServer like this
defmodule M do
use GenServer
# instead of fn -> [1, 2, 3] end
def create_object() do
{:ok, pid} = GenServer.start_link(__MODULE__, [1,2,3])
pid
end
def update_object(pid, new_data) do
GenServer.call(pid,{:update, new_data})
pid
end
def get_object(pid) do
IO.inspect GenServer.call(pid,:get)
pid
end
# GenServer callbacks
def init(state) do
# TODO initialization logic
{:ok, state}
end
# instead of `fn (state) -> state ++ new_data end`
def handle_call({:update, new_data}, _from, state) do
{:reply, :ok, state ++ new_data}
end
# instead of `&(&1)`
def handle_call(:get, _from, state) do
{:reply, state, state}
end
end
M.create_object() |> M.update_object([4,5]) |> M.get_object()
Agents are often used to introduce the concept of processes because the code looks initially much less arcane than GenServer code. In your example the client code provides the function that is updating the Agent’s state. In my GenServer based code I fixed the “meaning” of “update” in a module function to appending new_data to the GenServer state.
The point I’m trying to make is that Agent can be viewed as a GenServer turned-inside-out. With a GenServer the functionality to change process state is fixed within the callback module - with an Agent the computations (functionality) to change process state are provided from outside of the process.
The whole point for an Agent:
Agent.update(pid, fn (state) -> state ++ new_data end)
or a GenServer
GenServer.call(pid,{:update, new_data})
is that numerous (tens, hundreds, thousands, … of) other processes can concurrently append data to the list managed by process pid without sharing state.
In a typical conventional multi-threaded program that list would often be shared so that each thread could append its own data to the shared list - so the list would have to be explicitly protected by locks, mutexes, etc.
Now Agent.update and GenServer.call are synchronous calls - so they will block until the target process receives the message and sends a reply. For asynchronous processing there is Agent.cast and GenServer.cast
def update_object(pid, new_data) do
GenServer.cast(pid,{:update, new_data})
pid
end
...
# instead of `fn (state) -> state ++ new_data end`
def handle_cast({:update, new_data}, state) do
{:noreply, state ++ new_data}
end
Now when update_object/2 returns there is no guarantee that the list in pidhas been updated yet. But the code still works - because get_object/1 is still synchronous and because here messages are processed in the order of arrival, i.e. {:update,new_data} before :get, the list will be updated before the state is returned.
peerreynders
A lot can be learned by updating outdated code.
defmodule Chat.Server do
defp reply(pid, reply) do
send pid, {self(), reply}
end
defp cast(pid, message) do
send pid, {self(), message}
# noreply expected
end
# Send 'message' to all clients except 'sender'
defp broadcast(room, message, sender \\ :undefined) do
targets =
case is_pid sender do
false ->
room
_ ->
List.delete(room, sender)
end
Enum.each targets, fn(pid) -> cast(pid, {:message, message}) end
end
def loop(room) do
receive do
{pid, :join} ->
broadcast room, "Some user with pid #{inspect pid} joined"
reply(pid, :ok)
loop([pid|room])
{pid, {:say, message}} ->
broadcast room, "#{inspect pid}" <> message, pid
reply(pid, :ok)
loop(room)
{pid, :leave} ->
reply(pid, :ok)
new_room = List.delete(room, pid)
broadcast new_room, "User with pid #{inspect pid} left"
loop(new_room)
{_pid, :stop} ->
IO.puts "#{inspect self()} Server terminating"
:ok
end
end
end
defmodule Chat.Interface do
# Send 'message' to 'server' and wait for a reply
# Notice how this function is declared using 'defp' meaning
# it's private and can only be called inside this module
defp call(server, message) do
send server, {self(), message}
receive do
{^server, reply} ->
reply
after
1000 ->
IO.puts "Connection to room timed out"
:timeout
end
end
defp cast(server, message) do
send server, {self(), message}
# noreply expected
end
# Receive a pending message from 'server' and print it
def flush(server) do
receive do
# The caret '^' is used to match against the value of 'server',
# it is a basic filtering based on the sender
{ ^server, {:message, message} } ->
IO.puts message
flush(server)
# no more messages to flush
after
0 ->
:ok
end
end
# In all of the following functions 'server' stands for the server's pid
def join(server) do
call(server, :join)
end
def say(server, message) do
call(server, {:say, message} )
end
def leave(server) do
call(server, :leave)
end
# Part of the interface - though not the "Chat" interface
def stop(server) do
cast(server, :stop)
end
end
server_pid = spawn fn() -> Chat.Server.loop [] end
# register shell as a client
Chat.Interface.join server_pid
# Spawn another process as client
spawn fn() ->
Chat.Interface.join server_pid
Chat.Interface.say server_pid, "Hi!"
Chat.Interface.leave server_pid
end
# And another one
spawn fn() ->
Chat.Interface.join server_pid
Chat.Interface.say server_pid, "What's up?"
Chat.Interface.leave server_pid
end
:timer.sleep 500
# messages accumulated for the shell
Chat.Interface.flush server_pid
Chat.Interface.leave server_pid
Chat.Interface.stop server_pid
:timer.sleep 500
IO.puts"Done"
works even on http://elixirplayground.com/
Some user with pid #PID<0.58.0> joined
Some user with pid #PID<0.59.0> joined
#PID<0.58.0>Hi!
User with pid #PID<0.58.0> left
#PID<0.59.0>What's up?
User with pid #PID<0.59.0> left
#PID<0.57.0> Server terminating
Done
GenServer version
defmodule Chat.Server do
use GenServer
# Send 'message' to all clients except 'sender'
defp broadcast(room, message, sender \\ :undefined) do
targets =
case is_pid sender do
false ->
room
_ ->
List.delete(room, sender)
end
Enum.each targets, fn(pid) -> GenServer.cast(pid, {:message, message}) end
end
## GenServer callbacks
def terminate(_reason, _room) do
IO.puts "#{inspect self()} Server terminating"
:ok
end
# Note:
# Always use "from" as an opaque data type;
# don’t assume it is a tuple, as its representation might change in future releases.
# p.89 Designing for Scalability with Erlang/OTP (2016)
#
def handle_call({:join, pid}, _from, room) do
broadcast room, "Some user with pid #{inspect pid} joined"
{:reply, :ok, [pid|room]}
end
def handle_call({:say, pid, message}, _from, room) do
broadcast room, "#{inspect pid}" <> message, pid
{:reply, :ok, room}
end
def handle_call({:leave, pid}, _from, room) do
new_room = List.delete(room, pid)
broadcast new_room, "User with pid #{inspect pid} left"
{:reply, :ok, new_room}
end
end
defmodule Chat.Interface do
# In all of the following functions 'server' stands for the server's pid
def join(server) do
GenServer.call(server, {:join, self()})
end
def say(server, message) do
GenServer.call(server, {:say, self(), message})
end
def leave(server) do
GenServer.call(server, {:leave, self()})
end
# Part of the interface - though not the "Chat" interface
def stop(server) do
GenServer.stop(server)
end
# Receive a pending message from 'server' and print it
def flush(server) do
receive do
{:"$gen_cast", {:message, message }} ->
IO.inspect message
flush(server)
# no more messages to flush
after
0 ->
:ok
end
end
end
{:ok, server_pid} = GenServer.start Chat.Server, []
# register shell as a client
Chat.Interface.join server_pid
# Spawn another process as client
spawn fn() ->
Chat.Interface.join server_pid
Chat.Interface.say server_pid, "Hi!"
Chat.Interface.leave server_pid
end
# And another one
spawn fn() ->
Chat.Interface.join server_pid
Chat.Interface.say server_pid, "What's up?"
Chat.Interface.leave server_pid
end
:timer.sleep 500
# messages accumulated for the shell
Chat.Interface.flush server_pid
Chat.Interface.leave server_pid
Chat.Interface.stop server_pid
:timer.sleep 500
IO.puts"Done"
Elixir playground doesn’t support GenServer.stop.
And finally the Agent version … it’s wrong in so many ways.
defmodule Chat.App do
## Agent.update runs the following functions in the Agent's process
## via the anonymous functions returned by the "handle_x" functions
# Send 'message' to all clients except 'sender'
defp broadcast(room, message, sender \\ :undefined) do
targets =
case is_pid sender do
false ->
room
_ ->
List.delete(room, sender)
end
Enum.each targets,
fn(pid) -> send pid, {self(), {:message, message}} end
end
defp handle_join(pid) do
fn(room) ->
broadcast room, "Some user with pid #{inspect pid} joined"
[pid|room]
end
end
defp handle_say(pid, message) do
fn(room) ->
broadcast room, "#{inspect pid}" <> message, pid
room
end
end
defp handle_leave(pid) do
fn(room) ->
new_room = List.delete(room, pid)
broadcast new_room, "User with pid #{inspect pid} left"
new_room
end
end
## the following functions run in the client's process
def join(agent) do
Agent.update(agent, handle_join(self()))
end
def say(agent, message) do
Agent.update(agent, handle_say(self(),message))
end
def leave(agent) do
Agent.update(agent, handle_leave(self()))
end
# Receive pending messages from 'agent' and print them
def flush(agent) do
receive do
# The caret '^' is used to match against the value of 'server',
# it is a basic filtering based on the sender
{ ^agent, {:message, message} } ->
IO.puts message
flush(agent)
# no more messages to flush
after
0 ->
:ok
end
end
end
{:ok, agent_pid} = Agent.start fn -> [] end
# register shell as a client
Chat.App.join agent_pid
# Spawn another process as client
spawn fn() ->
Chat.App.join agent_pid
Chat.App.say agent_pid, "Hi!"
Chat.App.leave agent_pid
end
# And another one
spawn fn() ->
Chat.App.join agent_pid
Chat.App.say agent_pid, "What's up?"
Chat.App.leave agent_pid
end
:timer.sleep 500
# messages accumulated for the shell
Chat.App.flush agent_pid
Chat.App.leave agent_pid
Agent.stop agent_pid
:timer.sleep 500
IO.puts"Done"
The following is more in keeping with how an Agent is actually supposed to be used - as a state container. As a result the chat is now peer-to-peer - there is no “server” as such. However the clients do not keep the “room” as part of their own state - that is farmed out to the agent and manipulated through the anonymous functions created by the various handle_x functions.
However it still remains a strange way of using an agent.
defmodule Chat.App do
#"handle_x" return anonymous functions
# that are sent to the agent to run in the agent process
defp handle_join(pid) do
fn(room) ->
{room, [pid|room]}
end
end
defp handle_say() do
fn(room) -> room end
end
defp handle_leave(pid) do
fn(room) ->
new_room = List.delete(room, pid)
{new_room, new_room}
end
end
## the following functions run in the client's process
## now only the clients are sending messages
def join(agent) do
pid = self()
old_room = Agent.get_and_update(agent, handle_join(pid))
broadcast old_room, "Some user with pid #{inspect pid} joined"
end
def say(agent, message) do
pid = self()
room = Agent.get(agent, handle_say())
broadcast room, "#{inspect pid}" <> message, pid
end
def leave(agent) do
pid = self()
new_room = Agent.get_and_update(agent, handle_leave(pid))
broadcast new_room, "User with pid #{inspect pid} left"
end
# Send 'message' to all clients except 'sender'
defp broadcast(room, message, sender \\ :undefined) do
targets =
case is_pid sender do
false ->
room
_ ->
List.delete(room, sender)
end
Enum.each targets,
fn(pid) -> send pid, {self(), {:message, message}} end
end
# Receive pending messages from 'agent' and print them
def flush() do
receive do
# The caret '^' is used to match against the value of 'server',
# it is a basic filtering based on the sender
{ _pid, {:message, message} } ->
IO.puts message
flush()
# no more messages to flush
after
0 ->
:ok
end
end
end
{:ok, agent_pid} = Agent.start fn -> [] end
# register shell as a client
Chat.App.join agent_pid
# Spawn another process as client
spawn fn() ->
Chat.App.join agent_pid
Chat.App.say agent_pid, "Hi!"
Chat.App.leave agent_pid
end
# And another one
spawn fn() ->
Chat.App.join agent_pid
Chat.App.say agent_pid, "What's up?"
Chat.App.leave agent_pid
end
:timer.sleep 500
# messages accumulated for the shell
Chat.App.flush
Chat.App.leave agent_pid
Agent.stop agent_pid
:timer.sleep 500
IO.puts"Done"
Popular in Questions
Other popular topics
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #code-sync
- #javascript
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #websockets
- #supervisor
- #advent-of-code
- #elixirconf-us
- #distillery
- #processes
- #forms
- #api
- #metaprogramming
- #security
- #hex









