Elixir by Example

Catching up new nodes

Builds on Process Groups with :pg

Run it with elixir --name kv09@127.0.0.1 kv-09-catching-up.exs Download source

Step 8 ended with a gap: a node that joins later starts empty, because replication only covers writes made while it was a member. This step lets a new replica catch up before it serves anything: it copies a snapshot of the data from a replica that already has it.

The order matters. The newcomer first joins the :kv_replicas group, so every write from then on is sent to it too, and only then asks for the snapshot. Done the other way round, a write landing between the copy and the join would reach neither.

The demo:

  1. Three founding replicas (this node, a and b) take two writes.
  2. Node c joins and copies its data from b. It reads both keys without ever having seen the writes.
  3. A write made after c joined reaches c like any other replica.
start_peer = fn name ->
  {:ok, pid, node} = :peer.start_link(%{name: name, host: ~c"127.0.0.1", longnames: true})
  :ok = :erpc.call(node, :code, :add_paths, [:code.get_path()])
  {pid, node}
end

{_peer_a, a} = start_peer.(:kv09_a)
{_peer_b, b} = start_peer.(:kv09_b)
founders = [node(), a, b]

The replica from step 8, with one change: start/1 takes the node to copy from (nil for a founding replica). It joins the group in init/1, then catches up in handle_continue/2, which runs before the replica handles any message. Writes that arrive while the snapshot is on its way wait in the mailbox and are applied after it, so the snapshot can't overwrite them.

{:module, KV, bytecode, _} =
  defmodule KV do
    use GenServer

    def start(source) do
      {:ok, _pid} = GenServer.start(__MODULE__, source, name: __MODULE__)
      :ok
    end

    def put(key, value) do
      for replica <- :pg.get_members(:kv_replicas) do
        :ok = GenServer.call(replica, {:put, key, value})
      end

      :ok
    end

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

    @impl true
    def init(source) do
      :ok = :pg.join(:kv_replicas, self())
      {:ok, %{}, {:continue, {:catch_up, source}}}
    end

    @impl true
    def handle_continue({:catch_up, nil}, map), do: {:noreply, map}

    def handle_continue({:catch_up, source}, _map) do
      {:noreply, GenServer.call({__MODULE__, source}, :snapshot)}
    end

    @impl true
    def handle_call({:put, k, v}, _from, map), do: {:reply, :ok, Map.put(map, k, v)}
    def handle_call({:get, k}, _from, map), do: {:reply, Map.get(map, k), map}
    def handle_call(:snapshot, _from, map), do: {:reply, map, map}
  end

load = fn n ->
  {:module, KV} = :erpc.call(n, :code, :load_binary, [KV, ~c"nofile", bytecode])
  {:ok, _} = :erpc.call(n, :pg, :start, [:pg])
end

{:ok, _} = :pg.start_link()
for n <- [a, b], do: load.(n)
for n <- founders, do: :ok = :erpc.call(n, KV, :start, [nil])

Group membership spreads between nodes in the background, so before writing we wait until this node sees the expected number of replicas.

wait_for_replicas = fn again, count ->
  if length(:pg.get_members(:kv_replicas)) == count,
    do: :ok,
    else: Process.sleep(10) && again.(again, count)
end

:ok = wait_for_replicas.(wait_for_replicas, 3)
:ok = KV.put("lang", "elixir")
:ok = KV.put("vm", "beam")

Node c joins, copying from b. KV.start/1 returns once c has joined the group; the catch-up runs right after, and c's first get waits behind it in the mailbox.

{_peer_c, c} = start_peer.(:kv09_c)
load.(c)
:ok = :erpc.call(c, KV, :start, [b])
IO.inspect(:erpc.call(c, KV, :get, ["lang"]), label: "c reads lang")
IO.inspect(:erpc.call(c, KV, :get, ["vm"]), label: "c reads vm")
c reads lang: "elixir"
c reads vm: "beam"

From now on c is an ordinary replica: once this node sees it in the group, writes reach it directly.

:ok = wait_for_replicas.(wait_for_replicas, 4)
:ok = KV.put("otp", "29")
IO.inspect(:erpc.call(c, KV, :get, ["otp"]), label: "c reads otp")
c reads otp: "29"

Pitfall: there is still a gap. :pg membership reaches other nodes a moment after c joins. A write started on another node in that moment skips c, and if b had not applied it yet when it took the snapshot, c never gets it. Closing that gap takes a protocol where joining and copying are one step, which is what Mnesia does: the next step.

Pitfall: the snapshot is the whole map, sent in one message. Fine for thousands of keys; for millions, copy in chunks and keep the source responsive between them. Meanwhile every write waits for the newcomer's reply, so a long catch-up slows writes for everyone.

Improve this page on GitHub