AudioProxy.Semaphore (audio_proxy v0.4.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/1 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 FIFO 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/1 plus a receive with the caller's own timeout. AudioProxy.RenderCoordinator uses request/1 directly, because a coordinator that blocked waiting for a slot could not answer the joins that are coalescing onto it.

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

held and queued are the occupancy and depth after the event; wait and duration are native time units. Metadata is always %{capacity:, queue_size:}. :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.

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

What request/1 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.

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

outcome()

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

What request/1 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, :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.
  • :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.

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 day a sixth is added.

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)

@spec request(GenServer.server()) :: 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.

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/1 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.