Ferricstore.Raft.WARaftBackend.Batcher (ferricstore v0.4.2)

Namespace-window batcher for the WARaft replacement backend.

The normal WARaft write path stays direct. This process is used only when a namespace explicitly configures a window_ms larger than the default, so the common SET/DEL hot path does not pay a GenServer hop just to discover there is nothing to coalesce.

Summary

Types

hot_slot()

slot()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

flush(shard_index, timeout \\ 30000)

name(shard_index)

start_link(shard_index, opts \\ [])

stop(shard_index)

write(shard_index, prefix, command, window_ms)

write_batch(shard_index, commands)

write_delete_batch(shard_index, keys)

write_delete_batch_async(shard_index, keys, from, sync_token \\ nil)

write_put_batch(shard_index, entries)

write_put_batch_async(shard_index, entries, from, sync_token \\ nil)

Types

hot_slot()

@type hot_slot() :: %{
  groups: [{GenServer.from(), [term()]}],
  count: non_neg_integer(),
  timer_ref: reference() | nil,
  timer_token: reference() | nil,
  window_ms: non_neg_integer(),
  created_mono: integer()
}

slot()

@type slot() :: %{
  commands: [tuple()],
  froms: [GenServer.from()],
  count: non_neg_integer(),
  timer_ref: reference() | nil,
  timer_token: reference() | nil,
  window_ms: pos_integer(),
  created_mono: integer()
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

flush(shard_index, timeout \\ 30000)

@spec flush(non_neg_integer(), timeout()) :: :ok | {:error, term()}

name(shard_index)

@spec name(non_neg_integer()) :: atom()

start_link(shard_index, opts \\ [])

@spec start_link(
  non_neg_integer(),
  keyword()
) :: GenServer.on_start()

stop(shard_index)

@spec stop(non_neg_integer()) :: :ok

write(shard_index, prefix, command, window_ms)

@spec write(non_neg_integer(), binary(), tuple(), pos_integer()) :: term()

write_batch(shard_index, commands)

@spec write_batch(non_neg_integer(), [tuple()]) :: term()

write_delete_batch(shard_index, keys)

@spec write_delete_batch(non_neg_integer(), [binary()]) :: term()

write_delete_batch_async(shard_index, keys, from, sync_token \\ nil)

@spec write_delete_batch_async(
  non_neg_integer(),
  [binary()],
  GenServer.from(),
  Ferricstore.Raft.WARaftBackend.SyncGate.token() | nil
) :: :ok | {:direct, term()}

write_put_batch(shard_index, entries)

@spec write_put_batch(non_neg_integer(), [{binary(), binary(), non_neg_integer()}]) ::
  term()

write_put_batch_async(shard_index, entries, from, sync_token \\ nil)

@spec write_put_batch_async(
  non_neg_integer(),
  [{binary(), binary(), non_neg_integer()}],
  GenServer.from(),
  Ferricstore.Raft.WARaftBackend.SyncGate.token() | nil
) :: :ok | {:direct, term()}