AudioProxy.Semaphore (audio_proxy v0.7.0)

Copy Markdown View Source

The render-slot budget: at most AP_MAX_CONCURRENCY renders at once, with at most AP_QUEUE_SIZE waiting behind them.

ffmpeg is CPU-bound, so the number of encoders running at once is the one resource this proxy has to ration. Everything else — the coalescing registry, the bounded buffers, the kill discipline — assumes a slot was granted first. A request that cannot get one and cannot wait for one is a 429, which is the only place §5's Retry-After comes from.

The server never blocks

request/2 is a GenServer.call that answers immediately with one of three outcomes:

  • :granted — a slot was free and the caller now holds it
  • :queued — the caller is in the wait queue and will receive {AudioProxy.Semaphore, :granted} when its turn comes
  • {:error, {:queue_full, retry_after}} — no slot, no room to wait

Grants happen inside release/1 and DOWN handling, never in a callback that waits for something. That is what lets a caller queue for minutes without the semaphore being unable to answer anyone else — including the releases that are the only way the queue ever moves.

acquire/1 is the blocking convenience on top: request/2 plus a receive with the caller's own timeout. AudioProxy.RenderCoordinator uses request/2 directly, because a coordinator that blocked waiting for a slot could not answer the joins that are coalescing onto it.

Admission classes

The wait queue is ordered rather than flat: interactive > high > normal > low. A freed slot goes to the oldest waiter of the highest non-empty class, and FIFO holds within a class, so classes/0 is a tie-breaking order on top of arrival order rather than a replacement for it.

interactive is the default, which is what makes the whole mechanism invisible: a caller that says nothing is in the one class nothing can outrank and nothing can displace, queued behind exactly the callers that were already in front of it. A workload where nobody speaks a class is plain FIFO, and AudioProxy.SemaphorePropertyTest compares it against a FIFO model to keep it that way. Nothing in this repository passes :class today; the classes exist for callers that have background work to defer — cache warming, batch rendering — and want it to yield to a live listener.

When the queue is full, an arrival that outranks something queued displaces the newest waiter of the lowest non-interactive class present: the newest has waited least, so displacing it wastes the least sunk waiting. The victim receives {AudioProxy.Semaphore, {:displaced, retry_after}} — a reply distinct from queue-full, and retryable the same way. An arrival that outranks nothing queued is refused with {:queue_full, retry_after}, exactly as before.

Starvation of the lower classes is the contract, not a defect. There is no aging: under sustained interactive load, low waits indefinitely and is displaced first. That is safe because deferred work is not lost — a render that never got its background slot is still rendered lazily the moment someone asks for it directly — and it is visible, because every event carries per-class depth. A scheduler that let batch work overtake a listener would be trading the one latency the proxy is judged on for throughput nobody is waiting on.

One slot per process, released by exit if not by hand

A holder is monitored, and its DOWN releases the slot — so a crashed render costs a slot for as long as the monitor takes to fire, and no longer. A queued waiter is monitored too, and its DOWN drops it from the queue, so a client that gives up while waiting does not get handed a slot nobody is holding.

release/1 is therefore a promptness optimisation rather than the guarantee, and it is idempotent: releasing twice, or releasing without holding, is :ok. The one thing the monitor cannot cover is a long-lived caller whose release did not reach a wedged semaphore — see release/1. Acquiring twice from one process is {:error, :already_held} — a slot is a process' budget, and a second one would be an accounting error rather than a deadlock worth waiting on.

The caller-timeout race

A grant can land in the window between acquire/1's timeout firing and the caller doing anything about it. acquire/1 closes it by releasing and then draining: release/1 is synchronous, so by the time it returns the server has processed it and any grant it sent is already in the mailbox — the drain is exact rather than a race of its own.

Configuration is read per operation

Capacity and queue size come from AudioProxy.Config on every call, not from init/1. There is no reconfiguration at runtime, so in production this is just a :persistent_term read; in tests it means put_config/1 takes effect without restarting a supervised process. Lowering capacity below the number of slots currently held grants nothing new until it drains, which is the only sensible reading of it.

Tests that want a semaphore of their own can pass :name, :capacity and :queue_size to start_link/1, which pins them and skips the config read.

Retry-After

Derived from a moving average of recent slot-hold durations, scaled by how deep the queue already is: roughly "how long the renders in front of you have been taking, times how many of them there are, over how many run at once". Coarse by construction — §5 only requires the header to exist and be sane, and a client that retries a little early finds the queue full again and is told again. Before any render has completed there is nothing to average, so a placeholder stands in until the first one does.

Events

EventMeasurementsWhen
[:audio_proxy, :semaphore, :acquired]held, queued, waita slot was taken, immediately or after queueing
[:audio_proxy, :semaphore, :queued]held, queueda caller joined the wait queue
[:audio_proxy, :semaphore, :rejected]held, queued, retry_afterthe queue was full
[:audio_proxy, :semaphore, :released]held, queued, durationa holder gave its slot back
[:audio_proxy, :semaphore, :abandoned]held, queueda waiter left the queue before its turn
[:audio_proxy, :semaphore, :displaced]held, queued, retry_aftera waiter was displaced by a higher class

held and queued are the occupancy and depth after the event; wait and duration are native time units. Metadata is always %{capacity:, queue_size:, class:}, where class is the class of whoever the event is about — the caller granted, queued, rejected or displaced, or the holder that released. :abandoned is one more event than design.md listed, and it is here so that a queue draining by attrition — clients giving up — does not leave queued reading high until the next unrelated event.

Every event also carries queued_interactive, queued_high, queued_normal and queued_low: the depth of each class, not just the one the event is about. Reporting only the event's own class would publish a low depth that goes stale for as long as low is starved — which is precisely the state an operator needs to see, and precisely when no low event fires.

AudioProxy.Metrics counts :rejected from this set and takes its occupancy gauges from stats/2 instead, so the accuracy of what it publishes does not depend on every event being seen.

Summary

Types

An admission class. See Admission classes in the moduledoc.

What request/2 answers. See the moduledoc.

Seconds a rejected caller is told to wait. Always at least 1.

Occupancy, for tests and for anyone asking rather than subscribing.

Functions

Takes a slot for the calling process, waiting for one if the queue has room.

Returns a specification to start this module under a supervisor.

The admission classes, highest first.

Every event this module emits, for a consumer attaching to all of them.

Gives back the calling process' slot, or removes it from the wait queue.

How long release/1 may block before it gives up and leaves the slot to the monitor.

Asks for a slot for the calling process, answering immediately.

The Retry-After this semaphore would put on a rejection right now.

Starts the semaphore.

Current occupancy and the limits it is measured against.

Types

class()

@type class() :: :interactive | :high | :normal | :low

An admission class. See Admission classes in the moduledoc.

outcome()

@type outcome() ::
  :granted
  | :queued
  | {:error, {:queue_full, retry_after()}}
  | {:error, :already_held}

What request/2 answers. See the moduledoc.

retry_after()

@type retry_after() :: pos_integer()

Seconds a rejected caller is told to wait. Always at least 1.

stats()

@type stats() :: %{
  held: non_neg_integer(),
  queued: non_neg_integer(),
  capacity: pos_integer(),
  queue_size: non_neg_integer()
}

Occupancy, for tests and for anyone asking rather than subscribing.

Functions

acquire(opts \\ [])

@spec acquire(keyword()) :: :ok | {:error, term()}

Takes a slot for the calling process, waiting for one if the queue has room.

Returns :ok, {:error, {:queue_full, retry_after}} when there was no room to wait, {:error, {:displaced, retry_after}} when a higher class took the place this caller was waiting in, {:error, :timeout} when :timeout elapsed first, or {:error, :already_held}. Options:

  • :timeout — how long to wait for a queued slot. Defaults to :infinity, because the thing that bounds a render is AP_RENDER_TIMEOUT and a shorter wait here would only turn a queued request into a failed one.
  • :class — the admission class, defaulting to :interactive. A caller that leaves it alone can never be displaced, so it never sees the :displaced error above.
  • :server — which semaphore, for tests running their own.

A timeout releases whatever the race may have granted, so it does not leak a slot; see The caller-timeout race in the moduledoc, and release/1 for the one case that release cannot cover.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

classes()

@spec classes() :: [class()]

The admission classes, highest first.

Public so that a caller choosing a class, or a consumer labelling a metric by one, reads the order off the module that enforces it rather than restating it.

events()

@spec events() :: [:telemetry.event_name()]

Every event this module emits, for a consumer attaching to all of them.

The same service AudioProxy.Telemetry.render_events/0 performs for the render lifecycle, and for the same reason: a consumer that listed the names itself would go on working, silently short of one, the next time this list grows. It has grown once already — :displaced arrived with the admission classes — and the consumers that read it needed no edit.

release(server \\ AudioProxy.Semaphore)

@spec release(GenServer.server()) :: :ok

Gives back the calling process' slot, or removes it from the wait queue.

Idempotent, and :ok even if the semaphore is not running — a caller releasing from its own terminate/2 during shutdown must not crash because the semaphore stopped first.

:ok here means "this caller is done with the slot", not "the semaphore has processed that". A dead semaphore has nothing to release into, and a wedged one that does not answer within release_timeout/0 is caught the same way, which is deliberate: crashing the caller would achieve nothing the monitor is not already going to do. What recovers the slot in that second case is the holder exiting, so a long-lived caller that releases into a wedged semaphore does hold its slot until it dies. Every caller in this codebase releases on its way out.

release_timeout()

@spec release_timeout() :: pos_integer()

How long release/1 may block before it gives up and leaves the slot to the monitor.

Public for the same reason AudioProxy.Ffmpeg.Render.cancel_timeout/0 is: a caller that releases from its own terminate/2 has to size its shutdown budget against this, and a hardcoded number there would drift the moment this one changes.

request(server \\ AudioProxy.Semaphore, opts \\ [])

@spec request(
  GenServer.server(),
  keyword()
) :: outcome()

Asks for a slot for the calling process, answering immediately.

See the moduledoc for the three outcomes. A :queued caller receives {AudioProxy.Semaphore, :granted} when a slot comes free, and must then treat itself as a holder — including releasing it. It may instead receive {AudioProxy.Semaphore, {:displaced, retry_after}}, which means a higher class took its place in a full queue and it is no longer waiting; only a caller that named a class below interactive can receive it.

Options:

  • :class — one of classes/0, defaulting to :interactive. See Admission classes in the moduledoc.

An unrecognised :class raises ArgumentError, in the caller's process and before the call is made. That is a programmer error rather than a request outcome — the class is an argument this codebase writes, not something a client can send — so it is not part of outcome/0 and the semaphore never sees it. The request paths' "errors are data" rule governs what a client can cause; a typo'd atom is a bug, and a bug is louder as an exception.

retry_after(server \\ AudioProxy.Semaphore)

@spec retry_after(GenServer.server()) :: retry_after()

The Retry-After this semaphore would put on a rejection right now.

request/2 already carries one on the rejection it returns. This is for the caller that got as far as queueing and then gave up waiting: the same "come back later", with the same estimate behind it, but no rejection to read it off.

start_link(opts \\ [])

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

Starts the semaphore.

Options, all optional:

  • :name — defaults to this module, which is what the application tree starts and what every other function defaults to.
  • :capacity, :queue_size — pin the limits instead of reading AP_MAX_CONCURRENCY / AP_QUEUE_SIZE per operation. For tests that want a semaphore of their own; production uses the config.

stats(server \\ AudioProxy.Semaphore, timeout \\ 5000)

@spec stats(GenServer.server(), timeout()) :: stats()

Current occupancy and the limits it is measured against.

For tests and for AudioProxy.Readiness; nothing on the render path reads it.

timeout is exposed because the readiness probe has a budget an orchestrator set — a probe that waits the default five seconds for a wedged semaphore has already failed, whatever it eventually answers. Callers with no deadline of their own should leave it alone.