DEV Community: Resende

Exponential Backoff with Elixir

Resende — Tue, 11 Apr 2023 11:05:11 +0000

When we rely on external services, they may not always be available, may take a long time to respond, or may not be able to fulfil our requests. This can be frustrating, but if it's a one-time request and we only need to query one host, we can simply keep trying until we get a response.

However, if we have multiple providers for the same service and need to query them repeatedly, such as in a scenario where we're tracking prices for different franchises of a store and maintaining a leaderboard of the best prices, we can't rely on the same strategy of continually polling a single service. In these cases, we need a more sophisticated approach, such as exponential backoff, to manage the requests and handle any failures more efficiently.

In some cases, the stores that we're polling may not have the specific item we're looking for, or they may have an unstable server or poor internet connection. This can make it challenging to avoid wasting time on unsuccessful requests.

So how can we prevent permanent polling, active wait, blocking code, and scenarios like this:

you - are you ready yet ?

service - NO.

you - What about now?

service - ..... for the last time NO.

you - And now?

service - still NO.

you - Hey service could you give me ...?

service - Iup here you go!

Getting started

Today I'm gonna talk about exponential back-off, and how to deal with this kind of problem in a very simple way.

Better yet I'm gonna show you some examples in Elixir that make this whole process a lot simpler with OTP.

Exponential backoff is like taking a deep breath and counting to 10 before trying again when you don't succeed at something on the first try.
But instead of counting to 10, you wait a little longer each time you try again, giving yourself and others around you a chance to catch their breath and avoid chaos.

We will use GenServers so the final result ends up in a very simple and elegant implementation.

The premises are:

You make the request:
- The request went through. Great, do the next request in 5 seconds.
- The request didn't go through, we'll try again in 25 seconds (5^2).

Now we could do this to infinity but let's do a smarter implementation with an allowlist and blocklist for the stores that return n unsuccessful requests so we don't have an infinite waiting time for the next request

You make the request :
- The request went through. Great, do the next request in 5 seconds.
- The request didn't go through, we'll try again in 25 seconds (5^2) and you get a ticket (1).
- The request didn't go through and you already have 3 tickets, you're going on the blocklist.

Cool so how do I implement this in Elixir?

Well, it's pretty easy!

We can have this structure :

GenServer for the Stores
GenServer for the Allowlist
GenServer for polling each store

GenServer for the stores

In this, we have the main GenServer that will create the allowlist, and each of the polling store processes.

If you don't know, a GenServer is implemented in two parts: the client API and the server callbacks. If you want to learn more check this blog post on concurrency mechanisms in Elixir that I did a while back.

At the public API level, we only need a way to get the prices. This is the objective of this module, get prices.

defmodule Project.Stores do
 def prices do
    GenServer.call(__MODULE__, :prices)
 end
 # ....
end

On the private callback side, we need to support the public price call (of course) and the refresh callback to update the allow list.

The price call is gonna query the allowlist for the prices of the stores that are currently available and answering to our queries.

defmodule Project.Stores do
  # ...
  # private API
  def handle_call(:prices, _from, %{allowlist: allowlist} = state),
    do: {:reply, Stores.Allowlist.list(allowlist), state}

  def handle_info(:refresh, %{period: period, allowlist:   allowlist} = state) do
    case API.list_stores() do
      {:ok, stores} ->
        start_store_polling(stores, allowlist)

      error ->
        error
    end

    schedule_work(period)

    {:noreply, state}
  end
  # ...
end

Finally, we can have private functions like starting the polling GenServer for each store and a function to periodically wake up the process by sending a message to self.

This message will be caught by the handle_info on the private callback allowing for a refresh of the stores.

defmodule Project.Stores do
  # ...
  defp start_store_polling(stores, allowlist) do
    for store <- stores do
      Stores.Polling.start_link(store, allowlist)
    end
  end

  defp schedule_work(period), do: Process.send_after(self(), :refresh, period)
end

GenServer for the allowlist

This allowlist will both keep all the current stores that are responding as well as their prices and a quarantine list for stores that may be removed if they keep failing to respond. With this we need the public API to support adding a store, adding its price, removing a store, quarantining a store, and finally listing the stores and prices.

defmodule Project.Stores.Allowlist do
  use GenServer

  def start_link,
    do: GenServer.start_link(__MODULE__, :ok)

  def init(:ok),
    do: {:ok, %{allowlist: %{}, quarantine: MapSet.new()}}

  def add(pid, store, price),
    do: GenServer.cast(pid, {:add, store, price})

  def remove(pid, store),
    do: GenServer.cast(pid, {:remove, store})

  def quarantine(pid, store),
    do: GenServer.cast(pid, {:quarantine, store})

  def list(pid),
    do: GenServer.call(pid, :list)
  # ...
end

The private API is self-describing, the cast adds, removes, lists, and changes a store from the allow list to the quarantine list. There's nothing outside of the ordinary here.

defmodule Project.Stores.Allowlist do
  # ...
  # private API 
  def handle_cast({:add, store, price}, state) do
    quarantine = MapSet.delete(state.quarantine, store)
    allowlist = Map.put(state.allowlist, store, price)

    {:noreply, %{state | quarantine: quarantine, allowlist: allowlist}}
  end

  def handle_cast({:remove, store}, state) do
    quarantine = MapSet.delete(state.quarantine, store)
    allowlist = Map.delete(state.allowlist, store)

    {:noreply, %{state | quarantine: quarantine, allowlist: allowlist}}
  end

  def handle_cast({:quarantine, store}, state) do
    allowlist = Map.delete(state.allowlist, store)
    quarantine = MapSet.put(state.quarantine, store)

    {:noreply, %{state | quarantine: quarantine, allowlist: allowlist}}
  end

  def handle_call(:list, _from, state) do
    {:reply, Map.values(state.allowlist), state}
  end
end

GenServer for polling a store

The polling store GenServer itself is where all the fun stuff happens. It keeps track of all the requests and the number of failed attempts to communicate to the store and loads constant values like the number of max_attempts allowed and the timeout/base_timeout.

defmodule Project.Stores.Polling do
  use GenServer

  import Project.Config, only: [config!: 2]

  alias Project.Stores.Allowlist

  def start_link(store, allowlist) do
    name = process_name(store)

    state = %{
      store: store,
      allowlist: allowlist,
      attempts: 0,
      max_attempts: config!(__MODULE__, :max_attempts),
      timeout: 2_000,
      base_timeout: 2_000
    }

    GenServer.start_link(__MODULE__, state, name: name)
  end
  # ...
end

On the init function we pass the state that we received from start_link and call the private function schedule_work that will send a message to this process, and make it do some work (the name is accurate).

defmodule Project.Stores.Polling do
  # ...
  def init(state) do
    schedule_work(state.base_timeout)

    {:ok, state}
  end
  # ...
end

When the process receives a message to do some work it queries the store for the current price. Depending on the result, it's handled differently. This is especially easy with tuples in Elixir using the :ok and :error symbols on the first element.

defmodule Project.Stores.Polling do
  # ...
  def handle_info(:work, state) do
    case query_price(state.store) do
      {:ok, price} ->
        handle_successful_polling(state, price)

      {:error, _reason} ->
        handle_missing_polling(state)
    end
  end
  # ...
end

If the polling is successful we add the store and the price to the allowlist GenServer.
We will reschedule the polling with the base timeout and update the process state to set the timeout to the base timeout with 0 retry attempts.

defmodule Project.Stores.Polling do
  # ...
  defp handle_successful_polling(state, price) do
    Allowlist.add(state.allowlist, state.store, price)
    schedule_work(state.base_timeout)

    new_state = %{state | timeout: state.base_timeout, attempts: 0}

    {:noreply, new_state}
  end
  # ...
end

If the process misses the poll and it's the last attempt we are going to remove it from the Allowlist and update the state to mark the attempts as the max attempts.

defmodule Project.Stores.Polling do
  # ...
  defp handle_missing_polling(
         %{attempts: attempts, max_attempts: max_attempts} = state
       )
       when attempts == max_attempts - 1 do
    Allowlist.remove(state.allowlist, state.store)

    {:stop, :normal, %{state | attempts: max_attempts}}
  end
  # ...
end

If it's not the last attempt and the process misses the polling we quarantine that store, exponentiate the timeout and increase the attempt counter.
After that, we set the new state with the timeout and attempts while scheduling the next poll with the new timeout.

defmodule Project.Stores.Polling do
  # ...
  defp handle_missing_polling(state) do
    Allowlist.quarantine(state.allowlist, state.store)

    timeout = exponential_backoff(state.timeout)
    attempts = state.attempts + 1
    new_state = %{state | timeout: timeout, attempts: attempts}

    schedule_work(timeout)

    {:noreply, new_state}
  end
  # ...
end

At the end of the file, we define the helper functions for this module,
exponential backoff, schedule_work, and process name.

The exponential backoff function is a wrapper around the power function from Erlang with a round of the value, the schedule work function is a send-after to the same process, and the process_name uses the module and store to give a unique name to each process.

defmodule Project.Stores.Polling do
  # ...
  defp exponential_backoff(timeout), do: :math.pow(timeout, 2) |> round()

  defp schedule_work(timeout), do: Process.send_after(self(), :work, timeout)

  defp process_name(store), do: :"#{__MODULE__}-#{store}"
end

Conclusion

This was a fairly technical blog post but I truly hope you learned something useful.
Although the example presented in this post is somewhat specific, with small changes it's possible to adapt it to your specific use case.

Hope this will help you in your next project if an exponential backoff is needed, maybe when you are dependent on some service that rate limits requests with PlugAttack as described in this blog post.

Thank you for reading and if you have any questions or doubts feel free to reach out on Twitter at @Resende_666.

GenServer, Agent, Task

Resende — Tue, 04 May 2021 20:00:09 +0000

The first thing people tell you about Elixir is the scalability with lightweight threads of execution (called processes) that are isolated and exchange information via messages, the concurrency model with the actor model that allows you to program a concurrent program like a sequential one and the fault tolerance with supervisors which describe how to restart parts of your system when things go wrong.

However, I see a lot of newcomers that don't make use of these features and don't know how, because the concepts are a little different than what they are used to. This blog post is trying to introduce 3 important concepts that can be useful when programming Elixir.

Concurrency model

Elixir uses the Erlang virtual machine (BEAM), that uses actors to describe concurrency, in other words, a contained process that communicates with other processes through message passing

Process send and receive

This communication with message passing is composed of two main functions, send and receive, and we can send a message based on its Process Identifier (PID).

The send function does not block the sender process, it puts the message on the receiver process mailbox and continues.

The receive block will grab a message from the mailbox and try to pattern match it against the patterns defined in the receive block. If the process mailbox is empty it will block until a new message arrives.

Taken from the Elixir documentation:

defmodule Example do
  def listen do
    receive do
      {:ok, "hello"} -> IO.puts("World")
    end

    listen()
  end
end

iex> pid = spawn(Example, :listen, [])
#PID<0.108.0>

iex> send pid, {:ok, "hello"}
World
{:ok, "hello"}

iex> send pid, :ok
:ok

You can notice that the listen function processes messages for an infinite duration using recursion. This is possible because of "Tail Call Optimization" (TCO) and it ensures that if the last thing a function does is the invocation of another function (or itself), then there won't be a stack push. Instead, a simple jump will occur.

We’ve looked at the Elixir abstractions for concurrency but sometimes we need greater control and for that, we turn to the OTP behaviors that Elixir is built on.

We're gonna talk about Supervisors, GenServers, Agents, and Tasks.

Supervisor

Supervises processes with configuration which determines if/when/how a process that it is supervising is restarted. These should be used, either via explicit or dynamic or task supervisors, to supervise almost all of your processes. You almost always want your worker processes supervised in some way.

Taken from the elixir doc we can see how to set up a supervisor with a child process called stack where the init parameter is a list with the atom :hello:

children = [
  # The Stack is a child started via Stack.start_link([:hello])
  %{
    id: Stack,
    start: {Stack, :start_link, [[:hello]]}
  }
]

# Now we start the supervisor with the children and a strategy
{:ok, pid} = Supervisor.start_link(children, strategy: :one_for_one)

# After started, we can query the supervisor for information
Supervisor.count_children(PID)

GenServer

A generic server. That's it, ok? bye... ok let's expand.
An OTP server is a module with the GenServer behavior that implements a set of callbacks. At its most basic level, a GenServer is a single process that runs a loop that handles one message per iteration passing along an updated state.
Use it for anything where you have a need to represent something that has a state at a point of time and can be queried. This could represent a user in a chat, a char in a game, a game map, a shopping cart, etc.

The most important functions to know when using GenServer are:

start_link
init
call
handle_call
cast
handle_cast
send
handle_info

Let's do a very brief and simple explanation:

The start_link function starts a GenServer.
The init function initiates the state of the GenServer and starts calls.
call and handle_call are used for synchronous calls to the GenServer, normally where we need a return value from the GenServer.
cast and handle_cast are used in asynchronous calls to the GenServer, where we don't need a return value.
handle_info must be used for all other messages a server may receive that are not sent via call or cast, including regular messages sent with send.

So let's see a simple example of a queue with a GenServer:

defmodule Queue do
  use GenServer

  ### Public API 

  def start_link(state \\ []) do
    GenServer.start_link(__MODULE__, state, name: __MODULE__)
  end

  def queue, do: GenServer.call(__MODULE__, :queue)
  def dequeue, do: GenServer.call(__MODULE__, :dequeue)
  def enqueue(value), do: GenServer.cast(__MODULE__, {:enqueue, value})

  ### Private API
  def init(state), do: {:ok, state}

  def handle_call(:dequeue, _from, [value | state]) do
    {:reply, value, state}
  end

  def handle_call(:dequeue, _from, []), do: {:reply, nil, []}

  def handle_call(:queue, _from, state), do: {:reply, state, state} 

  def handle_cast({:enqueue, value}, state) do
    {:noreply, state ++ [value]}
  end

end

If we test it in iex:

iex> Queue.start_link([1, 2, 3])
{:ok, #PID<0.90.0>}
iex> Queue.dequeue
1
iex> Queue.dequeue
2
iex> Queue.queue
[3]
iex(5)> Queue.enqueue(5)
:ok
iex(6)> Queue.queue()
[3, 5]

Most of the time you will be using this abstraction because it's really powerful for a range of problems.

Agent

Like the Elixir documentation says "Agents are a simple abstraction around state", its an abstraction around GenServer in which the only function is to retrieve and update its state.
It's very useful for sharing data between other processes where you can query the data from the Agent or update it from multiple processes with total atomic and secure operations.
It has a very simple API and easy to use.
If you need more than sharing state then you are probably looking for a GenServer and not an Agent.
A GenServer is more autonomous over its internal state than an Agent is.

So let's do a simple Queue with Agent:

defmodule Queue do
  def start_link(state \\ []) do
    Agent.start_link(fn -> state end, name: __MODULE__)
  end

  def enqueue(item) do
    Agent.update(__MODULE__, fn(state) -> state ++ [item] end)
  end

  def dequeue() do
    Agent.get_and_update(__MODULE__, fn([item | state]) -> {item, state} end)
  end

  def queue() do
    Agent.get(__MODULE__, & &1)
  end
end

Let's test it on iex:

iex(1)> Queue.start_link([1,2,3])
{:ok, #PID<0.155.0>}
iex(2)> Queue.dequeue()
1
iex(3)> Queue.dequeue()
2
iex(4)> Queue.queue()
[3]
iex(5)> Queue.enqueue(5)
:ok
iex(6)> Queue.queue()
[3, 5]

In a simple case like a queue where we are only storing, updating and pulling state from the process an Agent version will be simpler and have the same behaviour that the GenServer one, but like all abstraction, if you want to deviate from what it gives it's gonna be very difficult to do what you want. You will have to use a GenServer for the complete control of its internal state and the handle_ callbacks.

Task

A process designed to handle a one-off task before shutting down and returning whatever results. Has a special task supervisor which makes it easy to dynamically spawn and monitor tasks.

We can make use of the async-await pattern with this kind of process that allows you to write sequential code and running it concurrently. async creates a new process, links it and monitors it. Once the task action finishes, a message will be sent to the caller with the result.
With await you can read this message sent by the task.

We can also implement cool patterns of batch processing like parallel maps with tasks:

defmodule Parallel do
  def map(collection, func) do
    collection
    |> Enum.map(&(Task.async(fn -> func.(&1) end)))
    |> Enum.map(&Task.await/1)
  end
end

Remember Elixir makes a distinction between anonymous functions and named functions, where the former must be invoked with a dot (.) between the variable name and parentheses. The capture operator bridges this gap by allowing named functions to be assigned to variables and passed as arguments in the same way we assign, invoke, and pass anonymous functions.

Good for jobs that can be processed without blocking the current process and taking advantage of the lightweight processes from BEAM.

Some examples in iex:

iex(9)> task = Task.async(fn -> Integer.is_even(1) end)
%Task{
  owner: #PID<0.152.0>,
  pid: #PID<0.169.0>,
  ref: #Reference<0.164594496.859832322.56552>
}
iex(10)> result = Task.await(task)
false

I hope you will incorporate these types of tools in your daily Elixir programming and that they will help you out solving cool problems!

Stay safe and be happy! 👋