Doing things only once in Elixir
Tags: elixir, stampedingherd
Published: 2022-10-07 12:45:06

[Listen very carefully, I shall say zis only once](https://www.youtube.com/watch?v=m2V1EAerRWw)

It often so happens that multiple processes need the same, potentially expensive function to be executed. A common solution is to cache the computed value for future callers, but on a cold cache (or when a cached value expires) it could still happen that the function is executed multiple times. Such a spike in resource usage is known as the stampeding herd problem. If only the animals would proceed in an orderly fashion.

Elixir's `GenServer` provides the `handle_call` callback, which is guaranteed to execute synchronously. As such, if several processes make a call to a `GenServer`, they would execute sequentially, and the value could even be cached for the benefit of the next caller. Problem solved.

However, anything that runs within the `handle_call` or any other callback is blocking, potentially making the `GenServer` appear unresponsive and turn it into a bottleneck. Ever heard of Python's GIL :)

But what if the `GenServer` could dish out tasks to be executed as separate processes and keep the execution time of the callback to a minimum? To make this work, we need to cover two concepts.

## `handle_call` doesn't have to reply immediately

You are probably familiar with the anatomy of a typical `handle_call` implementation:

``` elixir
def handle_call(request, from, state) do
  result, new_state = calculate_state(request, state )
  {:reply, result, new_state}
end
```

The `:reply` in the tuple means that the result will be sent immediately to the caller (the `from`). But `handle_call` may also return `{:noreply, new_state}` in which case the caller will be left hanging until it receives a reply or times out.

To send a reply to a caller after having initially returned `{:noreply, new_state}` we have to use `GenServer.reply/2`, which needs the `from` and some kind of result. This can be called from within any callback implementation.

## `Task` is built on message passing

It's good advice to only call `Task.async` if you are also going to await the result. Typically you would do so with `Task.await` or some of the other functions available in the `Task` module. But under the hood these functions rely on receiving the result from the task by means of message passing. That is, they make use of `Kernel.SpecialForms.receive/1` to wait for the result to roll in.

When you call `Task.async` from within a `GenServer` callback, you could await it with `Task.await` as usual, but for the duration of the computation of the task, the callback would be blocking, preventing the `GenServer` from handling any other calls. Depending on what you want from your `GenServer` this may defeat the purpose altogether.

Fortunately, within a `GenServer` you can await the success or failure of a task as you would await any other message - with a `handle_info` callback implementation:

``` elixir
def handle_info({ref, result}, state) do
  handle_task_success(ref, result, state)
end

def handle_info({:DOWN, ref, _, _, reason}, state) do
  handle_task_failure(ref, reason, state)
end
```

Make sure to guard that `ref` is a reference, otherwise you might catch unintended calls.

## A solution

So here is a broad overview of a solution to the stampeding herd problem. Processes that are interested in the results of a function, can call a `GenServer` to provide them with the result. The `GenServer` returns the result if it has it, or spawns a `Task` to compute it and keeps track of the caller, or if it has spawned a `Task` in the past, it doesn't do so again, but still keeps track of the caller. On receiving the response, it notifies all the callers using `GenServer.reply`.

Here is the whole thing together:

``` elixir
defmodule GenHerder do
  use GenServer

  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, [], Keyword.put(opts, :name, __MODULE__))
  end

  @impl true
  def init(_opts) do
    {:ok, %{}}
  end

  @impl true
  def handle_call(request, from, state) do
    case state[request] do
      nil ->
        task =
          Task.Supervisor.async_nolink(__MODULE__.TaskSupervisor, fn ->
            handle_request(request)
          end)

        {:noreply, Map.put(state, request, {:task, task, [from]})}

      {:task, task, froms} ->
        {:noreply, Map.put(state, request, {:task, task, [from | froms]})}

      {:result, result} ->
        {:reply, result, state}
    end
  end

  defp handle_request(_request) do
    :only_once_kenobi
  end

  @impl true
  def handle_info({ref, result}, state) when is_reference(ref) do
    handle_task_success(ref, result, state)
  end

  @impl true
  def handle_info({:DOWN, ref, _, _, reason}, state) do
    handle_task_failure(ref, reason, state)
  end

  defp handle_task_success(ref, result, state) do
    # The task succeeded so we can cancel the monitoring and discard the DOWN message
    Process.demonitor(ref, [:flush])

    {request, _task_and_froms} =
      Enum.find(state, fn
        {_request, {:task, task, _forms}} -> task.ref == ref
        _ -> false
      end)

    {{:task, _task, froms}, state} = Map.pop(state, request)

    state = Map.put(state, request, {:result, result})

    # Send the result to everyone that asked for it
    for from <- froms do
      GenServer.reply(from, result)
    end

    {:noreply, state}
  end

  defp handle_task_failure(ref, reason, state) do
    {request, _task_and_froms} =
      Enum.find(state, fn
        {_request, {:task, task, _forms}} -> task.ref == ref
        _ -> false
      end)

    {{:task, _task, froms}, state} = Map.pop(state, request)

    # Send the result to everyone that asked for it
    for from <- froms do
      GenServer.reply(from, {:error, reason})
    end

    {:noreply, state}
  end
end
```

You have to start it with something like this in a supervision tree:

``` elixir
children = [
          {Task.Supervisor, name: GenHerder.TaskSupervisor},
          GenHerder
        ]

Supervisor.start_link(children, Keyword.put(opts, :strategy, :one_for_one))
```

And then you can call it with:

``` elixir
GenServer.call(GenHerder, request, timeout)
```

Obviously, you can add nice wrappers around it. But a simpler way might be to use my package on hexpm. Call `mix hex.info gen_herder` for more info.

It implements the above with some sugar and also adds optional expiry of results.