wktdev

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

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

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

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"

Where Next?

Popular in Questions Top

chokchit
** (DBConnection.ConnectionError) connection not available and request was dropped from queue after 2733ms. You can configure how long re...
New
vertexbuffer
Hello, can anybody help here..? I have a list of players and I what to delete an element, but every for loop the list is reverting to ori...
New
New
WestKeys
Currently suffering from paralysis by [HTTP client] analysis. This is rather unusual in Elixirland as there tends to be consensus on the ...
New
skosch
To my knowledge, put_in, Map.update etc. all have the one limitation of not automatically creating intermediate keys when needed (for exa...
New
jononomo
I am trying to figure out how Mix knows whether the environment is test, dev, or prod – where is this set? Thanks.
New
shahryarjb
Hello, I get Persian date from my client and convert it to normal calendar like this: def jalali_string_to_miladi_english_number(persi...
New
joeerl
Hello again - after a longish gap I’ve decided I really must dig into Elixir and see what’s been happening here - so I have a few questio...
New
sergio_101
I am VERY much an elixir newbie. I have taken one elixir course and one phoenix course on Udemy. During that course, I saw the instructor...
New
ashish173
I am using Ecto timestamps with postgres, I can see the timestamps() use the :naive_dateime but for my use case I wanted to store the ti...
New

Other popular topics Top

Harrisonl
We have an ECS cluster with 4 services, where each task joins a single cluster, via discovery ECS discovery service. Currently when I de...
New
danschultzer
None of the current solutions worked well for me, so I went ahead and built a user management system from scratch. This project took far...
548 29703 241
New
AstonJ
We’ve put together this wiki for Phoenix LiveView - please feel free to add any info you feel is worth including. What is Phoenix LiveV...
New
skosch
To my knowledge, put_in, Map.update etc. all have the one limitation of not automatically creating intermediate keys when needed (for exa...
New
johnnyicon
Hi all, I’ve just started learning Elixir and Phoenix Framework, so please pardon my n00bness at this stage. I’m trying to use Postgres...
New
stefanchrobot
What’s the safe way to decode a JSON string into a struct? I want to avoid calling String.to_atom. Jason.decode can give me a map with st...
New
gausby
I asked this very same question on twitter and got some interesting feedback, but I thought it would be a good question to ask here as we...
1207 39523 209
New
Qqwy
Original source of discussion: This topic on the Pragmatic Programmers’ Functional Web Development with Elixir, OTP, and Phoenix forum. ...
New
Brian
What is the proper way to load a module from a file in to IEX? In the python world, doing something like this pretty standard: from ....
New
svb
Hi! Currently I want to submit a form by pressing the Enter key. However, since my input field is of type “textarea” this is just adds a...
New

We're in Beta

About us Mission Statement