🎓 A1Learner
🔍
0 of 46 lessons completed

Elixir Project: Distributed KV

A distributed key-value store is a store that spreads keys across several worker processes. Each worker holds only its own keys, and a routing rule decides which worker holds each key.

Two workers hold separate keys

Each worker keeps its own map. A key stored on one worker is missing on the other.

defmodule DistKV do
  use GenServer
  def start_link(opts), do: GenServer.start_link(__MODULE__, %{}, opts)
  def put(pid, key, value), do: GenServer.call(pid, {:put, key, value})
  def get(pid, key), do: GenServer.call(pid, {:get, key})
  def all(pid), do: GenServer.call(pid, :all)
  def init(state), do: {:ok, state}
  def handle_call({:put, key, value}, _from, state), do: {:reply, :ok, Map.put(state, key, value)}
  def handle_call({:get, key}, _from, state), do: {:reply, Map.get(state, key), state}
  def handle_call(:all, _from, state), do: {:reply, state, state}
end
{:ok, first} = DistKV.start_link([])
{:ok, second} = DistKV.start_link([])
IO.inspect(DistKV.put(first, :a, 1))
IO.inspect(DistKV.get(first, :a))
IO.inspect(DistKV.get(second, :a))
IO.inspect(DistKV.all(first))
GenServer.stop(first)
GenServer.stop(second)

A hash rule routes each key to one shard

The hash rule gives every key a shard number. The same key always routes to the same shard.

defmodule DistKV do
  use GenServer
  def start_link(opts), do: GenServer.start_link(__MODULE__, %{}, opts)
  def put(pid, key, value), do: GenServer.call(pid, {:put, key, value})
  def get(pid, key), do: GenServer.call(pid, {:get, key})
  def init(state), do: {:ok, state}
  def handle_call({:put, key, value}, _from, state), do: {:reply, :ok, Map.put(state, key, value)}
  def handle_call({:get, key}, _from, state), do: {:reply, Map.get(state, key), state}
end
{:ok, shard0} = DistKV.start_link([])
{:ok, shard1} = DistKV.start_link([])
workers = [shard0, shard1]
shard = fn key -> :erlang.phash2(key, 2) end
IO.inspect(shard.(:a))
IO.inspect(shard.(:b))
IO.inspect(shard.(:c))
DistKV.put(Enum.at(workers, shard.(:a)), :a, 10)
DistKV.put(Enum.at(workers, shard.(:b)), :b, 20)
DistKV.put(Enum.at(workers, shard.(:c)), :c, 30)
IO.inspect(DistKV.get(Enum.at(workers, shard.(:a)), :a))
IO.inspect(DistKV.get(Enum.at(workers, shard.(:b)), :b))
IO.inspect(DistKV.get(Enum.at(workers, shard.(:c)), :c))
GenServer.stop(shard0)
GenServer.stop(shard1)

A killed worker restarts under a supervisor

The supervisor watches its worker and starts a new one after a kill. Polling proves the new worker is different from the old one.

defmodule DistKV do
  use GenServer
  def start_link(opts), do: GenServer.start_link(__MODULE__, %{}, opts)
  def put(pid, key, value), do: GenServer.call(pid, {:put, key, value})
  def get(pid, key), do: GenServer.call(pid, {:get, key})
  def init(state), do: {:ok, state}
  def handle_call({:put, key, value}, _from, state), do: {:reply, :ok, Map.put(state, key, value)}
  def handle_call({:get, key}, _from, state), do: {:reply, Map.get(state, key), state}
end
{:ok, sup} = DynamicSupervisor.start_link([])
{:ok, old} = DynamicSupervisor.start_child(sup, DistKV)
DistKV.put(old, :a, 1)
Process.exit(old, :kill)
poll = fn f, sup_arg, old_arg, n ->
  pids = Enum.map(Supervisor.which_children(sup_arg), fn {_id, pid, _type, _mods} -> pid end)
  cond do
    n <= 0 -> :timeout
    pids != [] and hd(pids) != old_arg -> hd(pids)
    true -> Process.sleep(5)
      f.(f, sup_arg, old_arg, n - 1)
  end
end
new_pid = poll.(poll, sup, old, 200)
IO.inspect(new_pid != old)
IO.inspect(is_pid(new_pid))
IO.inspect(DistKV.get(new_pid, :a))
Supervisor.stop(sup)

Example

defmodule DistKV do
  use GenServer
  def start_link(opts), do: GenServer.start_link(__MODULE__, %{}, opts)
  def put(pid, key, value), do: GenServer.call(pid, {:put, key, value})
  def get(pid, key), do: GenServer.call(pid, {:get, key})
  def all(pid), do: GenServer.call(pid, :all)
  def init(state), do: {:ok, state}
  def handle_call({:put, key, value}, _from, state), do: {:reply, :ok, Map.put(state, key, value)}
  def handle_call({:get, key}, _from, state), do: {:reply, Map.get(state, key), state}
  def handle_call(:all, _from, state), do: {:reply, state, state}
end
{:ok, _reg} = Registry.start_link(keys: :unique, name: DistRegistry)
{:ok, _p0} = DistKV.start_link(name: {:via, Registry, {DistRegistry, 0}})
{:ok, _p1} = DistKV.start_link(name: {:via, Registry, {DistRegistry, 1}})
[{pid0, _}] = Registry.lookup(DistRegistry, 0)
[{pid1, _}] = Registry.lookup(DistRegistry, 1)
workers = [pid0, pid1]
shard = fn key -> :erlang.phash2(key, 2) end
counter = :counters.new(1, [])
DistKV.put(Enum.at(workers, shard.(:a)), :a, 10)
:counters.add(counter, 1, 1)
DistKV.put(Enum.at(workers, shard.(:b)), :b, 20)
:counters.add(counter, 1, 1)
DistKV.put(Enum.at(workers, shard.(:c)), :c, 30)
:counters.add(counter, 1, 1)
IO.inspect(:counters.get(counter, 1))
IO.inspect(DistKV.get(Enum.at(workers, shard.(:a)), :a))
IO.inspect(DistKV.get(Enum.at(workers, shard.(:b)), :b))
IO.inspect(DistKV.get(Enum.at(workers, shard.(:c)), :c))
IO.inspect(DistKV.all(pid0))
IO.inspect(DistKV.all(pid1))
GenServer.stop(pid0)
GenServer.stop(pid1)

Key points

  • Shard: a worker that holds one part of the whole key set.
  • Routing: the rule that maps a key to its shard.
  • Registry: a lookup that finds a worker by a name instead of a pid.
  • Restart: a fresh worker starts empty after its old worker is killed.
  • Counter: a shared count that records how many writes the store handled.

In this example, a Registry names two DistKV shards, a hash rule routes three keys to those shards, and a counter records the three writes. The lookups then show each key on its own shard and the counter total.

Note

a restarted worker does not recover keys it held in memory.

Try it Yourself

Edit the code, then press Run.

Running is simulated

Exercise

A key hashes to shard 1 in a two-shard DistKV store. Where is the key stored?