Elixir by Example

Atomics, Counters and persistent_term

Edit any cell and run it in your browser Download source

Sometimes all you share between processes is a number: a count of requests, a peak, an "already done" flag. A GenServer works, but every update is a message and a wait. :atomics gives you integers that any process can change directly, with no process and no messages.

:atomics.new(n, opts) makes an array of n 64-bit integers, all zero. Indexes start at 1.

counts = :atomics.new(2, signed: true)
:atomics.put(counts, 1, 10)
:atomics.add(counts, 1, 5)
:atomics.sub(counts, 2, 3)
IO.inspect({:atomics.get(counts, 1), :atomics.get(counts, 2)}, label: "values")
values: {15, -3}

The array is a reference. Pass it to other processes like any value; they all change the same integers. A hundred tasks adding 1 each:

hits = :atomics.new(1, [])

1..100
|> Enum.map(fn _ -> Task.async(fn -> :atomics.add(hits, 1, 1) end) end)
|> Task.await_many()

IO.inspect(:atomics.get(hits, 1), label: "hits")
hits: 100

Act on the value you changed

Say at most 3 jobs may run at once. The tempting version reads, checks, then adds:

if :atomics.get(slots, 1) < 3, do: :atomics.add(slots, 1, 1)

Two processes can both read 2, both pass the check and both add: 4 jobs running. Between the read and the write, anything can happen. (A real race can't be shown reliably, so it isn't run here.)

add_get/3 adds and returns the new value in one step that no other process can split. Each caller gets a different number, so exactly 3 see a value up to 3. The rest take their add back with sub/3.

slots = :atomics.new(1, [])

try_start = fn ->
  if :atomics.add_get(slots, 1, 1) <= 3 do
    :started
  else
    :atomics.sub(slots, 1, 1)
    :rejected
  end
end

results = 1..10 |> Enum.map(fn _ -> Task.async(try_start) end) |> Task.await_many()
IO.inspect(Enum.count(results, &(&1 == :started)), label: "started")
IO.inspect(Enum.count(results, &(&1 == :rejected)), label: "rejected")
IO.inspect(:atomics.get(slots, 1), label: "running")
started: 3
rejected: 7
running: 3

sub_get/3 is the same for subtracting, and exchange/3 writes a value and returns the old one.

IO.inspect(:atomics.exchange(slots, 1, 0), label: "exchange returned")
IO.inspect(:atomics.get(slots, 1), label: "now")
exchange returned: 3
now: 0

compare_exchange: change it only if unchanged

compare_exchange(ref, i, expected, new) writes new only if the value is still expected. It returns :ok on success, or the value it found. A one-time flag: ten processes race to run setup, and exactly one wins.

flag = :atomics.new(1, [])

winners =
  1..10
  |> Enum.map(fn _ -> Task.async(fn -> :atomics.compare_exchange(flag, 1, 0, 1) end) end)
  |> Task.await_many()
  |> Enum.count(&(&1 == :ok))

IO.inspect(winners, label: "setup ran")
setup ran: 1

For anything add can't express, loop: read, compute, compare_exchange, and if someone changed the value in between, read again and retry. Keeping the largest value ever seen:

defmodule Peak do
  def record(ref, value) do
    current = :atomics.get(ref, 1)

    cond do
      value <= current -> current
      :atomics.compare_exchange(ref, 1, current, value) == :ok -> value
      true -> record(ref, value)
    end
  end
end

peak = :atomics.new(1, [])

[17, 4, 42, 8, 23]
|> Enum.map(fn n -> Task.async(fn -> Peak.record(peak, n) end) end)
|> Task.await_many()

IO.inspect(:atomics.get(peak, 1), label: "peak")
peak: 42

Atomics aren't owned by a process, unlike an ETS table. This array is made by a process that then exits, and it keeps working. It's freed once nothing references it anymore.

parent = self()
spawn(fn -> send(parent, {:made, :atomics.new(1, [])}) end)
made = receive do: ({:made, ref} -> ref)
:atomics.add(made, 1, 7)
IO.inspect(:atomics.get(made, 1), label: "creator gone")
creator gone: 7

:counters, for statistics

:counters is a cut-down :atomics for counting: add, sub, get and put, always signed. Its extra is :write_concurrency: adds from many schedulers at once get faster, but a get that runs during other adds may see them in any order. No add is lost, and a read sees every add that finished before it.

stats = :counters.new(2, [:write_concurrency])
[ok, err] = [1, 2]

record = fn n ->
  index = if rem(n, 10) == 0, do: err, else: ok
  :counters.add(stats, index, 1)
end

1..50
|> Enum.map(fn n -> Task.async(fn -> record.(n) end) end)
|> Task.await_many()

IO.inspect({:counters.get(stats, ok), :counters.get(stats, err)}, label: "ok, errors")
ok, errors: {45, 5}

info/1 returns a map with size and an approximate memory in bytes.

IO.inspect(:counters.info(stats).size, label: "size")
size: 2

The default, [:atomics], keeps reads consistent. Pick :counters for metrics that are mostly bumped and rarely read; pick :atomics when you need add_get, compare_exchange or unsigned values.

:persistent_term, read-mostly globals

:persistent_term stores any term under a key for the whole node. Reads are very fast: the value is not copied into the reading process, unlike an ETS lookup. Keys are shared by all code on the node, so start yours with your app or module name.

key = {:atomics_page, :limits}
:persistent_term.put(key, %{max_jobs: 3, timeout: 5_000})
IO.inspect(:persistent_term.get(key).max_jobs, label: "max_jobs")
IO.inspect(:persistent_term.get({:atomics_page, :none}, :default), label: "missing")
max_jobs: 3
missing: :default

The catch is on writes. Replacing or erasing a value makes every process on the node scan its heap for the old one, and any process still using it gets a full garbage collection that copies it in. With many processes, the whole node can become less responsive until they're done. So it's for config-like data that is set at boot and almost never changes.

A process still holding the old value keeps it working after the erase; the VM copies it over.

holder =
  spawn(fn ->
    limits = :persistent_term.get(key)
    send(parent, :got_it)
    receive do: (:show -> send(parent, {:held, limits}))
  end)

receive do: (:got_it -> :ok)
IO.inspect(:persistent_term.erase(key), label: "erased")
IO.inspect(:persistent_term.get(key, :gone), label: "after erase")
send(holder, :show)
receive do: ({:held, limits} -> IO.inspect(limits.max_jobs, label: "holder still has max_jobs"))
erased: true
after erase: :gone
holder still has max_jobs: 3

Which shared state when?

  • GenServer or Agent: state with rules, where updates must happen one at a time in order.
  • ETS: many keys, read and written by many processes, owned by one.
  • :atomics / :counters: plain integers that many processes bump: counts, gauges, flags.
  • :persistent_term: a few values read on every request and set about once per boot.

Atomics wrap around instead of growing like Elixir integers. One past the largest value is the smallest:

wrap = :atomics.new(1, signed: true)
:atomics.put(wrap, 1, 9_223_372_036_854_775_807)
IO.inspect(:atomics.add_get(wrap, 1, 1), label: "signed max + 1")

unsigned = :atomics.new(1, signed: false)
IO.inspect(:atomics.sub_get(unsigned, 1, 1), label: "unsigned 0 - 1")
signed max + 1: -9223372036854775808
unsigned 0 - 1: 18446744073709551615

Pitfall: atomics and counters wrap around silently at 64 bits, as shown above. An unsigned gauge that is decremented once too often becomes a huge number, not -1.

Pitfall: an index outside 1..n raises ArgumentError. Index 0 is never valid.

Pitfall: a :persistent_term.put/2 in a hot path (per request, per message) replaces the value each time and can stall the whole node. Store rarely-changing data there; put changing data in ETS or atomics.

Pitfall: an atomics or counters reference only works on the node that created it. Send it to another node and it points at nothing there.

Improve this page on GitHub