ExDataSketch.Server (ExDataSketch v0.10.0)

Copy Markdown View Source

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

estimate_opts()

@type estimate_opts() :: [{:window, :all}]

flush_opts()

@type flush_opts() :: [
  interval: pos_integer() | :infinity,
  callback: (struct() -> term()) | nil
]

Periodic return-and-reset configuration, matching ExDataSketch.Broadway.PeriodicAggregator.

sketch_type()

@type sketch_type() :: ExDataSketch.sketch_type() | module()

Either a registry atom (:hll) or a sketch module (ExDataSketch.HLL).

snapshot_opts()

@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).

start_opts()

@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
]

window_opts()

@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

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

estimate(server, opts \\ [])

@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

flush(server)

@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

merge(server, partial_sketch)

@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

reset(server)

@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

sketch(server)

@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

start_link(opts)

@spec start_link(start_opts()) :: GenServer.on_start()

Starts a server process.

Options

  • :sketch -- required. A registry atom (:hll, looked up via ExDataSketch.sketches/0) or a sketch module directly. Must resolve to a mergeable family (see ExDataSketch.Window.new/3) when :window is also given.
  • :sketch_opts -- options forwarded to the sketch module's new/1 (default: []).
  • :name -- optional, forwarded to GenServer.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

update(server, item)

@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

update_many(server, items)

@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

update_sync(server, item)

@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