Normandy.Coordination.StatefulContext (normandy v2.2.0)

View Source

GenServer-backed shared context for multi-agent systems.

StatefulContext provides a concurrent, process-based key-value store using GenServer with ETS backing.

Features

  • Concurrent access from multiple processes
  • Reads from ETS: each read makes one short GenServer.call for the table, then looks the key up in ETS. The context process does only short work (an insert, a delete, a notification), so a read waits behind at most that; update/4 runs its function in the caller, never in the context process
  • Atomic updates: the table is :protected, so only the context process writes it; update/4 commits with a compare-and-set (see there)
  • Optional pub/sub notifications for changes
  • Process supervision compatible

Example

# Start context process
{:ok, pid} = StatefulContext.start_link(name: :my_context)

# Store and retrieve data
:ok = StatefulContext.put(pid, "key", "value")
{:ok, "value"} = StatefulContext.fetch(pid, "key")

# Use namespaced keys
:ok = StatefulContext.put(pid, {"agent_1", "status"}, "active")

# Subscribe to changes
:ok = StatefulContext.subscribe(pid, self())

Summary

Types

key()
subscriber()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

delete(server, key)

Deletes a key from the context.

fetch(server, key)

Looks up key: {:ok, value} or :error.

get(server, key, default)

Retrieves a value with a default if not found.

get_table(server)

Returns the ETS table, for reads (:ets.lookup/2, :ets.select/2, ...).

has_key?(server, key)

Checks if a key exists in the context.

keys(server)

Returns all keys in the context.

put(server, key, value)

Stores a value in the context.

start_link(opts \\ [])

Starts a StatefulContext GenServer.

subscribe(server, subscriber_pid)

Subscribes a process to change notifications.

to_map(server)

Returns all data in the context as a map.

unsubscribe(server, subscriber_pid)

Unsubscribes a process from change notifications.

update(server, key, initial, fun)

Updates a value using a function.

Types

key()

@type key() :: String.t() | {String.t(), String.t()}

subscriber()

@type subscriber() :: pid()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

delete(server, key)

@spec delete(GenServer.server(), key()) :: :ok

Deletes a key from the context.

Example

:ok = StatefulContext.delete(pid, "key")

fetch(server, key)

@spec fetch(GenServer.server(), key()) :: {:ok, term()} | :error

Looks up key: {:ok, value} or :error.

Asks the context process for its table (one short GenServer.call), then reads the key from ETS.

Example

{:ok, value} = StatefulContext.fetch(pid, "key")
:error = StatefulContext.fetch(pid, "missing")

get(server, key, default)

@spec get(GenServer.server(), key(), term()) :: term()

Retrieves a value with a default if not found.

Example

value = StatefulContext.get(pid, "key", "default")

get_table(server)

@spec get_table(GenServer.server()) :: :ets.tid()

Returns the ETS table, for reads (:ets.lookup/2, :ets.select/2, ...).

The table is :protected: only the context process writes it, so a write from any other process raises ArgumentError. Write through put/3, update/4 and delete/2.

Example

table = StatefulContext.get_table(pid)
:ets.lookup(table, "key")

has_key?(server, key)

@spec has_key?(GenServer.server(), key()) :: boolean()

Checks if a key exists in the context.

Example

true = StatefulContext.has_key?(pid, "key")

keys(server)

@spec keys(GenServer.server()) :: [String.t()]

Returns all keys in the context.

Example

keys = StatefulContext.keys(pid)
#=> ["key1", "agent_1:status"]

put(server, key, value)

@spec put(GenServer.server(), key(), term()) :: :ok

Stores a value in the context.

Writes go through the context process for consistency; reads look the key up in ETS (after one short call for the table).

Example

:ok = StatefulContext.put(pid, "key", "value")
:ok = StatefulContext.put(pid, {"agent_1", "data"}, %{result: 42})

start_link(opts \\ [])

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

Starts a StatefulContext GenServer.

Options

Example

{:ok, pid} = StatefulContext.start_link(name: :shared_context)

subscribe(server, subscriber_pid)

@spec subscribe(GenServer.server(), subscriber()) :: :ok

Subscribes a process to change notifications.

The subscriber will receive messages of the form: {:context_changed, key, old_value, new_value}

Example

:ok = StatefulContext.subscribe(pid, self())

to_map(server)

@spec to_map(GenServer.server()) :: map()

Returns all data in the context as a map.

Example

data = StatefulContext.to_map(pid)
#=> %{"key1" => "value1", "agent_1:status" => "active"}

unsubscribe(server, subscriber_pid)

@spec unsubscribe(GenServer.server(), subscriber()) :: :ok

Unsubscribes a process from change notifications.

Example

:ok = StatefulContext.unsubscribe(pid, self())

update(server, key, initial, fun)

@spec update(GenServer.server(), key(), term(), (term() -> term())) :: :ok

Updates a value using a function.

If the key doesn't exist, uses the initial value.

fun runs in the calling process, not in the context process, so a slow fun blocks no other caller and a fun that raises raises in the caller. The result is committed only if the key still holds the value fun was given; otherwise fun runs again on the new value. So fun can run more than once and must have no side effects. After 10 lost races (a key written more often than fun takes) fun runs once inside the context process, serialized with every other call, so every update finishes. There a fun that raises crashes the context, and a fun that calls this same context (put/3, fetch/2, update/4, get_table/1, ...) exits with :calling_self and crashes it too.

The caller reads the key from ETS directly (get_table/1), so update/4 works only against a context on the caller's node.

Example

:ok = StatefulContext.update(pid, "counter", 0, fn count -> count + 1 end)