Elixir by Example

Backpressure

Edit any cell and run it in your browser Download source

Backpressure means a slow consumer slows its producers down instead of letting work pile up in memory. GenServer.call/3 gives you a little of it for free: the caller waits for the reply, so it can't get ahead. But call only stops a producer while its own request is in flight. To cap how much work is waiting in total, the server has to hold callers back on purpose.

The tool is a deferred reply. A handle_call/3 doesn't have to answer right away: it can return {:noreply, state}, keep the caller's from, and answer later with GenServer.reply/2. Until then the caller stays blocked inside its call.

A bounded buffer built that way: push/2 is answered at once while there is room, and held back while the buffer is full. Each pop/1 frees a slot, so the oldest waiting producer gets in and is released.

defmodule Buffer do
  use GenServer

  def start_link(capacity), do: GenServer.start_link(__MODULE__, capacity)
  def push(buffer, item, timeout \\ :infinity), do: GenServer.call(buffer, {:push, item}, timeout)
  def pop(buffer), do: GenServer.call(buffer, :pop)
  def info(buffer), do: GenServer.call(buffer, :info)

  @impl true
  def init(capacity), do: {:ok, %{capacity: capacity, items: :queue.new(), waiting: :queue.new()}}

  @impl true
  def handle_call({:push, item}, from, state) do
    if :queue.len(state.items) < state.capacity do
      {:reply, :ok, %{state | items: :queue.in(item, state.items)}}
    else
      # Full: park the caller. No reply means it stays
      # blocked.
      {:noreply, %{state | waiting: :queue.in({from, item}, state.waiting)}}
    end
  end

  def handle_call(:pop, _from, state) do
    case :queue.out(state.items) do
      {:empty, _} ->
        {:reply, :empty, state}

      {{:value, item}, items} ->
        {:reply, {:ok, item}, admit_waiting(%{state | items: items})}
    end
  end

  def handle_call(:info, _from, state) do
    blocked = for {_from, item} <- :queue.to_list(state.waiting), do: item
    {:reply, %{items: :queue.to_list(state.items), blocked: blocked}, state}
  end

  # A slot is free: move the oldest waiting item in and
  # release its producer.
  defp admit_waiting(state) do
    case :queue.out(state.waiting) do
      {:empty, _} ->
        state

      {{:value, {from, item}}, waiting} ->
        GenServer.reply(from, :ok)
        %{state | items: :queue.in(item, state.items), waiting: waiting}
    end
  end
end

The script waits for the other processes to reach a known state before printing, so the output doesn't depend on timing.

defmodule Wait do
  def until(check) do
    if check.(), do: :ok, else: Process.sleep(1) && until(check)
  end
end

A producer pushes five items into a buffer with room for two. It gets two in and then blocks on the third.

{:ok, buffer} = Buffer.start_link(2)
parent = self()

producer =
  spawn(fn ->
    for item <- 1..5 do
      :ok = Buffer.push(buffer, item)
    end

    send(parent, :producer_done)
  end)

Wait.until(fn -> Buffer.info(buffer).blocked == [3] end)
IO.inspect(Buffer.info(buffer), label: "full")
full: %{blocked: [3], items: [1, 2]}

Process.info/2 shows where the producer is: waiting inside gen:do_call, the receive that every call uses for its reply.

{:current_function, current} = Process.info(producer, :current_function)
IO.inspect(current, label: "producer is in")
producer is in: {:gen, :do_call, 4}

Every pop lets exactly one more item in. The producer moves at the consumer's pace, and the buffer never holds more than two.

for _ <- 1..2 do
  IO.inspect(Buffer.pop(buffer), label: "pop")
  Wait.until(fn -> length(Buffer.info(buffer).blocked) == 1 end)
  IO.inspect(Buffer.info(buffer), label: "now")
end

for _ <- 1..3, do: IO.inspect(Buffer.pop(buffer), label: "pop")

receive do
  :producer_done -> IO.puts("producer finished")
end
pop: {:ok, 1}
now: %{blocked: [4], items: [2, 3]}
pop: {:ok, 2}
now: %{blocked: [5], items: [3, 4]}
pop: {:ok, 3}
pop: {:ok, 4}
pop: {:ok, 5}
producer finished

Several producers queue up behind a full buffer in arrival order, one waiting entry each. That's the whole cost of backpressure: one blocked process per producer, not an unbounded pile of messages.

{:ok, buffer} = Buffer.start_link(1)
:ok = Buffer.push(buffer, :a)

for item <- [:b, :c, :d] do
  spawn(fn -> Buffer.push(buffer, item) end)
  Wait.until(fn -> List.last(Buffer.info(buffer).blocked) == item end)
end

IO.inspect(Buffer.info(buffer), label: "three producers waiting")
three producers waiting: %{blocked: [:b, :c, :d], items: [:a]}

Pitfall: a call timeout doesn't cancel the request. When a blocked push gives up, the caller exits with a timeout (caught here), but its item is still parked in the server, and the next free slot takes it. The producer thinks it failed; the item goes through anyway.

{:ok, buffer} = Buffer.start_link(1)
:ok = Buffer.push(buffer, :first)

try do
  Buffer.push(buffer, :gave_up, 50)
catch
  :exit, {:timeout, _} -> IO.puts("push timed out after 50 ms")
end

IO.inspect(Buffer.pop(buffer), label: "pop")
IO.inspect(Buffer.info(buffer), label: "after the timeout")
push timed out after 50 ms
pop: {:ok, :first}
after the timeout: %{blocked: [], items: [:gave_up]}

If the producer is going to retry, make that safe: give each item an id and drop duplicates, or have the server check whether the caller is still alive (monitor it) before admitting its item.

Pitfall: cast skips all of this. A producer that casts never waits, so a bounded buffer behind a cast API is bounded in name only: the overflow just moves into the server's mailbox (see Mailboxes and Slow Consumers).

Pitfall: the default call timeout is 5 seconds. A producer that is supposed to wait for a slow consumer needs :infinity or a timeout that fits, otherwise ordinary backpressure looks like a crash.

The same idea, the consumer setting the pace, is what GenStage and Broadway build on: consumers ask for a number of events, and producers never send more than was asked for.

Improve this page on GitHub