ExAthena.RequestQueue (ExAthena v0.17.0)

Copy Markdown View Source

ETS-backed GenServer semaphore that limits concurrent in-flight requests per provider.

Each provider has an independent slot cap (see ExAthena.Config.request_queue_max_depth/1). with_slot/3 is the primary entry point: it acquires a slot, runs the fun, and releases on every exit path. acquire/2 blocks the caller until a slot is available; release/1 frees the slot and wakes the next queued caller (FIFO). depth/1 reads the ETS counter directly, never blocking the caller.

Local inference servers (ollama / llama.cpp / exo) can only serve 1–3 requests concurrently; the gate keeps concurrent agent loops from overwhelming them and gives hosts queue visibility (waiting_count/1, :on_wait callbacks).

The queue is enabled by default. Disable via:

config :ex_athena, :request_queue, enabled: false

When the GenServer is not running (feature disabled), acquire/1 and release/1 are no-ops returning :ok and with_slot/3 just runs the fun.

Crash safety

Both sides of the semaphore are monitored:

  • Waiters — if a blocked caller exits before its slot is granted, the monitor fires and removes the dead entry from the queue.
  • Holders — if a caller that holds a slot dies without releasing (e.g. brutally killed mid-stream), the monitor fires and the slot is reclaimed, granting the next waiter. This covers exits that skip after blocks.

Waiter ordering

FIFO. Each waiter entry carries a reserved priority field (currently always 0) so priority scheduling can be added later without reshaping the queue.

Summary

Functions

acquire(provider, timeout \\ 5000)

Acquire a slot for provider. Blocks the caller until a slot is available or timeout milliseconds elapse (default 5 000 ms; :infinity supported).

cancel_acquire(provider)

Cancel any pending acquire for the calling process on provider.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

depth(provider)

Return the number of active (acquired) slots for provider.

release(provider)

Release a previously acquired slot for provider. Returns :ok.

start_link(opts \\ [])
waiting_count(provider)

Return the number of callers currently queued (blocked) for provider.

with_slot(provider, fun, opts \\ [])

Acquire a slot, run fun, release the slot on every exit path.

Functions

acquire(provider, timeout \\ 5000)

@spec acquire(atom(), timeout()) :: :ok

Acquire a slot for provider. Blocks the caller until a slot is available or timeout milliseconds elapse (default 5 000 ms; :infinity supported).

The calling process is monitored while it holds the slot — if it dies without releasing, the slot is reclaimed automatically.

Returns :ok when a slot is granted. When the GenServer is not running (feature disabled), returns :ok immediately as a no-op.

cancel_acquire(provider)

@spec cancel_acquire(atom()) :: :ok

Cancel any pending acquire for the calling process on provider.

Called automatically by with_slot/3 when an acquire timeout fires so that stale waiting entries do not prevent depth from decrementing on the next release/1.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

depth(provider)

@spec depth(atom()) :: non_neg_integer()

Return the number of active (acquired) slots for provider.

Reads directly from the ETS table; never blocks the caller and never goes through the GenServer mailbox. Returns 0 when the queue is disabled (ETS table does not exist).

release(provider)

@spec release(atom()) :: :ok

Release a previously acquired slot for provider. Returns :ok.

If callers are queued for this provider, the next one in line is unblocked and inherits the slot (depth stays unchanged). When the GenServer is not running, returns :ok immediately as a no-op.

start_link(opts \\ [])

waiting_count(provider)

@spec waiting_count(atom()) :: non_neg_integer()

Return the number of callers currently queued (blocked) for provider.

Hosts poll this to render queue-pressure indicators. Returns 0 when the queue is disabled.

with_slot(provider, fun, opts \\ [])

@spec with_slot(atom() | module() | nil, (-> result), keyword()) ::
  result | {:error, :request_queue_timeout}
when result: term()

Acquire a slot, run fun, release the slot on every exit path.

This is the canonical way to gate a provider call. Telemetry ([:ex_athena, :request_queue, :wait | :acquired | :released | :timeout]) is emitted around the slot lifecycle.

Passes straight through (no acquire) when:

  • provider is nil or a module (custom provider modules are not gated),
  • the queue: false option is given,
  • the queue feature is disabled in config.

Options

  • :queue — false bypasses the gate for this call (default true).
  • :timeout — milliseconds (or :infinity) to wait for a slot before returning {:error, :request_queue_timeout} (default 5_000).
  • :on_wait — optional fun/1 invoked with :waiting just before a blocking acquire and {:acquired, waited_ms} once the slot is granted. Not invoked at all when a slot is free — hosts use this to show a "waiting on GPU" state only when there is actually a wait.