Elixir by Example

Limit each client

Builds on Rate LimitingETS

Edit any cell and run it in your browser Download source

KV now serves many clients, and one of them is a script stuck in a loop, writing as fast as it can. It isn't malicious, just broken, and every write it makes is time the store doesn't spend on everyone else.

This step puts a limit on each client: at most limit writes per second. The counts live in an ETS table with :ets.update_counter/4, as on Rate Limiting, so checking the limit never waits on a process. Writes that pass go to the store; writes over the limit get {:error, :rate_limited} at once.

The store is step 6's: a GenServer owns the table and serializes writes, and reads go straight to ETS. Reads aren't limited here: they're cheap, and they don't queue behind anything.

defmodule KV.Store do
  use GenServer

  @table __MODULE__

  def start_link(_), do: GenServer.start_link(__MODULE__, nil, name: __MODULE__)

  def put(key, value), do: GenServer.call(__MODULE__, {:put, key, value})

  def get(key) do
    case :ets.lookup(@table, key) do
      [{^key, value}] -> {:ok, value}
      [] -> :error
    end
  end

  @impl true
  def init(nil) do
    :ets.new(@table, [:set, :protected, :named_table, read_concurrency: true])
    {:ok, nil}
  end

  @impl true
  def handle_call({:put, key, value}, _from, state) do
    :ets.insert(@table, {key, value})
    {:reply, :ok, state}
  end
end

The limiter: a fixed window per client. The clock is passed in, so the demo controls time.

defmodule KV.Limiter do
  @table __MODULE__

  def start, do: :ets.new(@table, [:set, :public, :named_table, write_concurrency: true])

  def check(client, limit, now) do
    slot = {client, div(now, 1_000)}
    count = :ets.update_counter(@table, slot, {2, 1}, {slot, 0})
    if count <= limit, do: :ok, else: {:error, :rate_limited}
  end

  # Old windows are useless once the clock has moved
  # past them.
  def sweep(now) do
    current = div(now, 1_000)
    :ets.select_delete(@table, [{{{:_, :"$1"}, :_}, [{:<, :"$1", current}], [true]}])
  end
end

The API clients use: check the limit, then write.

defmodule KV do
  @limit 5

  def put(client, key, value, now) do
    with :ok <- KV.Limiter.check(client, @limit, now) do
      KV.Store.put(key, value)
    end
  end

  defdelegate get(key), to: KV.Store
end

{:ok, _} = KV.Store.start_link(nil)
KV.Limiter.start()

In the same second, the broken script sends 20 writes and a well-behaved client sends 2. The script gets 5 through; the other client isn't affected at all.

loop = for n <- 1..20, do: KV.put(:loop_script, "counter", n, 10_000)
IO.inspect(Enum.frequencies(loop), label: "loop script")

fine = for n <- 1..2, do: KV.put(:billing, "invoice:#{n}", :paid, 10_000)
IO.inspect(fine, label: "billing")
loop script: %{:ok => 5, {:error, :rate_limited} => 15}
billing: [:ok, :ok]

The store has the loop script's fifth write, the last one let through.

IO.inspect(KV.get("counter"), label: "counter")
counter: {:ok, 5}

A second later, a new window: the script gets its next five.

again = for n <- 21..30, do: KV.put(:loop_script, "counter", n, 11_000)
IO.inspect(Enum.frequencies(again), label: "loop script, next second")
loop script, next second: %{:ok => 5, {:error, :rate_limited} => 5}

The sweep deletes the windows that are over.

IO.inspect(KV.Limiter.sweep(11_000), label: "old windows deleted")
old windows deleted: 2

Limiting stops one client from hogging the store. It doesn't help when every client is fine and the store is still too slow, for example because each write must reach a slow disk. That's the next step: backpressure.

Improve this page on GitHub