A supervised, named, concurrently-updatable sketch process with optional windowing and snapshotting.
ExDataSketch.Server is a GenServer wrapping a single sketch (or
ExDataSketch.Window), so an application can hold a running counter --
"distinct users right now", "distinct users in the last 5 minutes" -- as
an ordinary named process instead of hand-writing state management,
concurrency, and crash recovery.
Quick Example
{:ok, _pid} = ExDataSketch.Server.start_link(
name: :uniques,
sketch: :hll,
sketch_opts: [p: 14],
window: [every: :timer.minutes(1), keep: 5],
snapshot: [to: {ExDataSketch.Storage.ETS, :sketches, "uniques"}, every: :timer.seconds(30)]
)
ExDataSketch.Server.update(:uniques, user_id)
ExDataSketch.Server.estimate(:uniques)Backpressure
update/2 and update_many/2 are GenServer.cast/2 calls: they return
immediately and do not wait for the update to be applied. Under
sustained load faster than the server can process, its mailbox grows
without bound unless :max_queue is configured. This is a deliberate
trade for throughput -- callers that need a guarantee the update was
applied before continuing use update_sync/2 instead, a GenServer.call/2
that blocks until processed (and, as an ordinary call, provides natural
backpressure: a caller cannot get more than one update_sync/2 ahead of
the server).
When :max_queue is set, update/2 and update_many/2 drop the update
(rather than applying it) whenever the server's own mailbox is at or
above that threshold, emitting [:ex_data_sketch, :server, :drop]. This
bounds memory growth from a backed-up mailbox at the cost of the data
loss update/2 already implies under overload -- update_sync/2 is never
subject to :max_queue.
Windowing
Pass :window (the same options ExDataSketch.Window.new/3 accepts:
:every, :keep, :time_fn) to have the server hold a
ExDataSketch.Window instead of a bare sketch. estimate/1 then answers
"in the last keep * every" instead of "ever". See guides/windowing.md
for what that means precisely (a tumbling ring, not an exact sliding
window) and which sketch families can be windowed.
Add track_all_time: true to :window to also maintain a second,
un-windowed sketch alongside the window (same :sketch_opts, so the same
fixed size as any other sketch of that family -- not a growing or
unbounded structure). estimate(server, window: :all) reads it;
snapshot, if configured, always persists the window, not this second
sketch, which restarts empty after a crash even when enabled (see
baoulo/plans/0.10.0_phase4_design_review.md section 4.2 for the
rationale).
A windowed server does not support merge/2 in this release --
ExDataSketch.Window has no primitive to merge an already-built sketch
into its current slot (only raw-item update/2,3 and update_many/2).
Calling merge/2 on a windowed server raises
ExDataSketch.Errors.UnsupportedOperationError. This is tracked as
follow-up work in baoulo/plans/plan-0.10.0.md section 9, not an
oversight.
Snapshotting
Pass :snapshot (:to, an {backend_module, ref, key} triple naming
any ExDataSketch.Storage backend; :every, a millisecond interval or
:infinity) to persist the server's current state periodically and on
graceful shutdown. On start, the server attempts to load from the same
location first (crash recovery); if nothing is found, or loading fails
for any other reason, it starts from a fresh sketch either way --
[:ex_data_sketch, :server, :restore] fires either way, with found in
its metadata saying which happened.
Worst-case data loss from an ordinary (non-:kill) process termination
is bounded by :every -- a graceful stop or supervisor-initiated shutdown
additionally snapshots on terminate/2 before exiting, so that case loses
nothing. An untrappable Process.exit(pid, :kill) never runs
terminate/2 at all (this is a BEAM guarantee, not something any
process can opt out of), so in that specific case the periodic :every
interval is the only protection -- this is the "kill" scenario the
plan's own test targets.
Flushing
Pass :flush (:interval, a millisecond interval or :infinity;
:callback, an optional (struct() -> term())) for the
return-and-reset-on-a-timer pattern ExDataSketch.Broadway.PeriodicAggregator
uses: on each interval, the callback (if any) receives the current
sketch, [:ex_data_sketch, :server, :flush] fires, and the server resets
to a fresh sketch. PeriodicAggregator is a thin wrapper around a
:flush-configured Server -- see its moduledoc for the relationship.
:flush and :window are independent options; combining them is not
rejected, but resets the whole window (every slot) on each flush and is
not a use case this module specifically targets.
Summary
Types
Periodic return-and-reset configuration, matching ExDataSketch.Broadway.PeriodicAggregator.
Either a registry atom (:hll) or a sketch module (ExDataSketch.HLL).
Where and how often to persist state. :every may be :infinity (snapshot on graceful stop only).
The same options ExDataSketch.Window.new/3 accepts, plus :track_all_time.
Functions
Returns a specification to start this module under a supervisor.
Returns the server's current headline estimate.
Returns the server's current state and resets it to a fresh, empty one,
on demand (independent of any configured :flush interval).
Merges an already-built partial sketch into the server's current state.
Resets the server to a fresh, empty state (the window and, if configured, the all-time accumulator, when windowed).
Returns the server's current raw state: a %ExDataSketch.Window{} if
:window was configured, otherwise the bare sketch struct.
Starts a server process.
Updates the server with a single item.
Updates the server with every item in an enumerable.
Updates the server with a single item, blocking until it has been applied.
Types
@type estimate_opts() :: [{:window, :all}]
@type flush_opts() :: [ interval: pos_integer() | :infinity, callback: (struct() -> term()) | nil ]
Periodic return-and-reset configuration, matching ExDataSketch.Broadway.PeriodicAggregator.
@type sketch_type() :: ExDataSketch.sketch_type() | module()
Either a registry atom (:hll) or a sketch module (ExDataSketch.HLL).
@type snapshot_opts() :: [ to: {module(), term(), ExDataSketch.Storage.key()}, every: pos_integer() | :infinity ]
Where and how often to persist state. :every may be :infinity (snapshot on graceful stop only).
@type start_opts() :: [ sketch: sketch_type(), sketch_opts: keyword(), name: GenServer.name(), window: window_opts() | nil, snapshot: snapshot_opts() | nil, flush: flush_opts() | nil, max_queue: non_neg_integer() | :infinity ]
@type window_opts() :: [ every: pos_integer(), keep: pos_integer(), time_fn: (-> integer()), track_all_time: boolean() ]
The same options ExDataSketch.Window.new/3 accepts, plus :track_all_time.
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec estimate(GenServer.server(), estimate_opts()) :: number()
Returns the server's current headline estimate.
With no :window configured, this is ExDataSketch.estimate/1 on the
underlying sketch. With :window configured, it defaults to the windowed
reading (ExDataSketch.Window.estimate/1); pass window: :all to read
the un-windowed, all-time accumulator instead, which requires the server
to have been started with window: [..., track_all_time: true].
Raises ExDataSketch.Errors.UnsupportedOperationError for window: :all
when track_all_time: true was not set, or for any estimate/1,2 call
against a family with no single-value reading (see
ExDataSketch.estimate/1).
Examples
iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
iex> :ok = ExDataSketch.Server.update_sync(pid, "user_1")
iex> ExDataSketch.Server.estimate(pid) > 0.0
true
iex> {:ok, pid} = ExDataSketch.Server.start_link(
...> sketch: :hll, sketch_opts: [p: 10],
...> window: [every: 60_000, keep: 5, track_all_time: true]
...> )
iex> :ok = ExDataSketch.Server.update_sync(pid, "user_1")
iex> ExDataSketch.Server.estimate(pid) > 0.0 and ExDataSketch.Server.estimate(pid, window: :all) > 0.0
true
@spec flush(GenServer.server()) :: struct()
Returns the server's current state and resets it to a fresh, empty one,
on demand (independent of any configured :flush interval).
Unlike the automatic timer-driven flush (see the module documentation's
"Flushing" section), a manual call here does not invoke the :flush
callback -- only the timer does. Both emit
[:ex_data_sketch, :server, :flush].
Examples
iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
iex> :ok = ExDataSketch.Server.update_sync(pid, "user_1")
iex> flushed = ExDataSketch.Server.flush(pid)
iex> ExDataSketch.HLL.estimate(flushed) > 0.0
true
iex> ExDataSketch.Server.estimate(pid)
0.0
@spec merge( GenServer.server(), struct() ) :: :ok
Merges an already-built partial sketch into the server's current state.
Raises ExDataSketch.Errors.UnsupportedOperationError if the server is
windowed -- see the module documentation's "Windowing" section.
Examples
iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
iex> partial = ExDataSketch.HLL.new(p: 10) |> ExDataSketch.HLL.update("a")
iex> :ok = ExDataSketch.Server.merge(pid, partial)
iex> ExDataSketch.Server.estimate(pid) > 0.0
true
@spec reset(GenServer.server()) :: :ok
Resets the server to a fresh, empty state (the window and, if configured, the all-time accumulator, when windowed).
Examples
iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
iex> :ok = ExDataSketch.Server.update_sync(pid, "user_1")
iex> :ok = ExDataSketch.Server.reset(pid)
iex> ExDataSketch.Server.estimate(pid)
0.0
@spec sketch(GenServer.server()) :: struct()
Returns the server's current raw state: a %ExDataSketch.Window{} if
:window was configured, otherwise the bare sketch struct.
Examples
iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
iex> match?(%ExDataSketch.HLL{}, ExDataSketch.Server.sketch(pid))
true
@spec start_link(start_opts()) :: GenServer.on_start()
Starts a server process.
Options
:sketch-- required. A registry atom (:hll, looked up viaExDataSketch.sketches/0) or a sketch module directly. Must resolve to a mergeable family (seeExDataSketch.Window.new/3) when:windowis also given.:sketch_opts-- options forwarded to the sketch module'snew/1(default:[]).:name-- optional, forwarded toGenServer.start_link/3's:name(an atom,{:global, term}, or{:via, module, term}).:window-- optional. See the module documentation's "Windowing" section.:snapshot-- optional. See the module documentation's "Snapshotting" section.:flush-- optional. See the module documentation's "Flushing" section.:max_queue-- optional non-negative integer or:infinity(default::infinity). See the module documentation's "Backpressure" section.
Examples
iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
iex> is_pid(pid)
true
@spec update(GenServer.server(), term()) :: :ok
Updates the server with a single item.
Asynchronous (GenServer.cast/2) -- returns immediately, does not wait
for the update to be applied, and can be dropped under :max_queue
pressure. See the module documentation's "Backpressure" section, and
update_sync/2 for a guaranteed alternative.
Examples
iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
iex> ExDataSketch.Server.update(pid, "user_1")
:ok
@spec update_many(GenServer.server(), Enumerable.t()) :: :ok
Updates the server with every item in an enumerable.
Asynchronous, like update/2 -- see the module documentation's
"Backpressure" section.
Examples
iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
iex> ExDataSketch.Server.update_many(pid, ["a", "b", "c"])
:ok
@spec update_sync(GenServer.server(), term()) :: :ok
Updates the server with a single item, blocking until it has been applied.
Never dropped by :max_queue -- callers that need a guarantee the
update was applied use this instead of update/2.
Examples
iex> {:ok, pid} = ExDataSketch.Server.start_link(sketch: :hll, sketch_opts: [p: 10])
iex> :ok = ExDataSketch.Server.update_sync(pid, "user_1")
iex> ExDataSketch.Server.estimate(pid) > 0.0
true