defmodule HLClock do @moduledoc """ Hybrid Logical Clock Provides globally-unique, monotonic timestamps. Timestamps are bounded by the clock synchronization constraint, max_drift. Implementation assumes that timestamps are (at a minimum) regularly sent; a clock at rest will eventually be unable to generate timestamps due to assumed bounds on the logical clock relative to physical time. In order to account for physical time drift within the system, timestamps should regularly be exchanged between nodes. Generate a timestamp at one node via HLClock.send_timestamp/1; at the other node, call HLClock.recv_timestamp/2 with the received timestamp from the first node. Inspired by https://www.cse.buffalo.edu/tech-reports/2014-04.pdf """ alias HLClock.Timestamp @doc """ clock constructor requires the node_id, a millisecond clock fn and a maximum drift parameter in milliseconds """ def new(pt0 \\ default_time(), node_id \\ 0) do Timestamp.new(pt0, 0, node_id) end @doc """ Generate a single HLC Timestamp for sending to other nodes or local causality tracking """ def send_timestamp(%Timestamp{}=old, pt \\ default_time()) do nt = max(old.time, pt) nc = advance(old, nt) with :ok <- handle_drift(old.time, nt) do Timestamp.new(nt, nc, old.node_id) end end @doc """ Given the current timestamp for this node and a provided remote timestamp, perform the merge of both logical time and logical counters. Returns the new current timestamp for the local node """ def recv_timestamp(local, remote, pt \\ default_time()) do max_pt = max(pt, max(remote.time, local.time)) with {:ok, node_id} <- compare_node_ids(local.node_id, remote.node_id), :ok <- handle_drift(remote.time, pt, :remote_drift_violation), :ok <- handle_drift(max_pt, pt), log <- merge_logical(max_pt, local, remote) do Timestamp.new(max_pt, log, node_id) end end @doc """ Determines if the clock's timestamp "happened before" a different timestamp """ def before?(c1, c2) do Timestamp.less?(c1, c2) end @doc """ Ensure that System.os_time is returning in milliseconds """ def default_time(), do: System.os_time(:milliseconds) @doc """ Configurable clock synchronization parameter, ε. """ def max_drift(), do: Application.get_env(:hlclock, :max_drift_millis, 300_000) defp compare_node_ids(local_id, remote_id) when local_id == remote_id, do: {:error, :duplicate_node_id} defp compare_node_ids(local_id, _), do: {:ok, local_id} defp merge_logical(max_pt, local, remote) do cond do max_pt == local.time && max_pt == remote.time -> max(local.counter, remote.counter) + 1 max_pt == local.time -> local.counter + 1 max_pt == remote.time -> remote.counter + 1 true -> 0 end end defp handle_drift(l, pt, err \\ :clock_drift_violation) do cond do drift?(l, pt) -> {:error, err} true -> :ok end end defp drift?(l, pt) do abs(l - pt) > max_drift() end defp advance(old, new_time) do cond do old.time == new_time -> old.counter + 1 true -> 0 end end end