lib/audio_proxy/semaphore.ex
Rama main · réplica de lectura del repositorio original.
defmodule AudioProxy.Semaphore do
@moduledoc """
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
| Event | Measurements | When |
|---|---|---|
| `[:audio_proxy, :semaphore, :acquired]` | `held`, `queued`, `wait` | a slot was taken, immediately or after queueing |
| `[:audio_proxy, :semaphore, :queued]` | `held`, `queued` | a caller joined the wait queue |
| `[:audio_proxy, :semaphore, :rejected]` | `held`, `queued`, `retry_after` | the queue was full |
| `[:audio_proxy, :semaphore, :released]` | `held`, `queued`, `duration` | a holder gave its slot back |
| `[:audio_proxy, :semaphore, :abandoned]` | `held`, `queued` | a waiter left the queue before its turn |
| `[:audio_proxy, :semaphore, :displaced]` | `held`, `queued`, `retry_after` | a 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.
"""
use GenServer
alias AudioProxy.Config
@name __MODULE__
@telemetry [:audio_proxy, :semaphore]
# Highest first. The list *is* the order — grant scans it front to back,
# eviction scans it back to front — so there is no separate rank table to
# keep in step with it.
@classes [:interactive, :high, :normal, :low]
# `interactive` is not evictable: a listener's patience budget is not a
# batch's to spend.
#
# Redundant *for this one guarantee*, and only for it. `victim/2` also
# considers nothing above the arrival's own class, and nothing is above the
# top class — so with `interactive` first in `@classes`, removing this
# subtraction alone changes no outcome and leaves the suite green (measured).
# The rank filter is not redundant in the same way: removing *it* alone lets
# a class displace its own peers, and two tests say so. This subtraction
# earns its keep against the day a class is added above `interactive`, which
# would make a listener displaceable through the ordering alone.
@evictable @classes -- [:interactive]
@empty_order Map.new(@classes, &{&1, :queue.new()})
# Slot-hold durations kept for the `Retry-After` average. Enough to smooth a
# single outlier, short enough to track a change in what is being rendered.
@samples 20
# How long `release/1` waits for the server before leaving the slot to the
# monitor. Every callback here is O(1) or O(queue), so reaching this at all
# means something is badly wrong; it is a bound, not a budget.
@release_timeout 5_000
# Stands in for the average until a render has actually completed. Nothing
# measured is available at that point, and a wrong-but-small hint is better
# than either no header or a fabricated large one; it self-corrects after the
# first release.
@default_hold_ms 2_000
@typedoc "Seconds a rejected caller is told to wait. Always at least 1."
@type retry_after :: pos_integer()
@typedoc "An admission class. See *Admission classes* in the moduledoc."
@type class :: :interactive | :high | :normal | :low
@typedoc "What `request/2` answers. See the moduledoc."
@type outcome ::
:granted
| :queued
| {:error, {:queue_full, retry_after()}}
| {:error, :already_held}
@typedoc "Occupancy, for tests and for anyone asking rather than subscribing."
@type stats :: %{
held: non_neg_integer(),
queued: non_neg_integer(),
capacity: pos_integer(),
queue_size: non_neg_integer()
}
defstruct [
:capacity,
:queue_size,
held: %{},
waiters: %{},
order: @empty_order,
samples: []
]
@doc """
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.
"""
@spec start_link(keyword()) :: GenServer.on_start()
def start_link(opts \\ []) do
{name, opts} = Keyword.pop(opts, :name, @name)
GenServer.start_link(__MODULE__, opts, name: name)
end
@doc """
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.
"""
@spec events() :: [:telemetry.event_name()]
def events do
Enum.map(
[:acquired, :queued, :rejected, :released, :abandoned, :displaced],
&(@telemetry ++ [&1])
)
end
@doc """
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.
"""
@spec classes() :: [class()]
def classes, do: @classes
@doc """
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 `t: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.
"""
@spec request(GenServer.server(), keyword()) :: outcome()
def request(server \\ @name, opts \\ []) do
GenServer.call(server, {:request, self(), class!(opts)})
end
defp class!(opts) do
case Keyword.get(opts, :class, :interactive) do
class when class in @classes ->
class
other ->
raise ArgumentError,
"unknown admission class #{inspect(other)}, expected one of #{inspect(@classes)}"
end
end
@doc """
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.
"""
@spec acquire(keyword()) :: :ok | {:error, term()}
def acquire(opts \\ []) do
server = Keyword.get(opts, :server, @name)
timeout = Keyword.get(opts, :timeout, :infinity)
case request(server, Keyword.take(opts, [:class])) do
:granted -> :ok
:queued -> await_grant(server, timeout)
{:error, _reason} = error -> error
end
end
defp await_grant(server, timeout) do
receive do
{__MODULE__, :granted} -> :ok
{__MODULE__, {:displaced, retry_after}} -> {:error, {:displaced, retry_after}}
after
timeout ->
release(server)
# `release/1` is a call, so the server has already handled it and
# anything it sent is in this mailbox. Nothing can arrive later. A
# displacement is drained here too: it is the same stray message with
# the same claim on the mailbox, and it is now moot either way.
receive do
{__MODULE__, :granted} -> :ok
{__MODULE__, {:displaced, _retry_after}} -> :ok
after
0 -> :ok
end
{:error, :timeout}
end
end
@doc """
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.
"""
@spec release_timeout() :: pos_integer()
def release_timeout, do: @release_timeout
@doc """
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.
"""
@spec release(GenServer.server()) :: :ok
def release(server \\ @name) do
GenServer.call(server, {:release, self()}, @release_timeout)
catch
:exit, _reason -> :ok
end
@doc """
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.
"""
@spec retry_after(GenServer.server()) :: retry_after()
def retry_after(server \\ @name) do
GenServer.call(server, :retry_after)
catch
# A request answering a client must not fail because the semaphore is
# momentarily unreachable. The header is a hint, and one second is the
# floor the estimate is clamped to anyway.
:exit, _reason -> 1
end
@doc """
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.
"""
@spec stats(GenServer.server(), timeout()) :: stats()
def stats(server \\ @name, timeout \\ 5_000) do
GenServer.call(server, :stats, timeout)
end
## Server
@impl true
def init(opts) do
{:ok,
%__MODULE__{
capacity: Keyword.get(opts, :capacity),
queue_size: Keyword.get(opts, :queue_size)
}}
end
@impl true
def handle_call({:request, pid, class}, _from, state) do
cond do
Map.has_key?(state.held, pid) ->
{:reply, {:error, :already_held}, state}
# A second request from a process already waiting would put it in the
# queue twice and let it be granted twice.
Map.has_key?(state.waiters, pid) ->
{:reply, {:error, :already_held}, state}
map_size(state.held) < capacity(state) ->
{:reply, :granted, hold(state, pid, Process.monitor(pid), 0, class)}
map_size(state.waiters) < queue_size(state) ->
{:reply, :queued, enqueue(state, pid, class)}
# Full. Either this arrival outranks something displaceable, or it is
# refused exactly as it was before classes existed.
victim = victim(state, class) ->
{:reply, :queued, state |> displace(victim) |> enqueue(pid, class)}
true ->
{retry_after, state} = reject(state, class)
{:reply, {:error, {:queue_full, retry_after}}, state}
end
end
def handle_call({:release, pid}, _from, state) do
{:reply, :ok, forget(state, pid)}
end
def handle_call(:retry_after, _from, state) do
{:reply, estimate(state), state}
end
def handle_call(:stats, _from, state) do
{:reply,
%{
held: map_size(state.held),
queued: map_size(state.waiters),
capacity: capacity(state),
queue_size: queue_size(state)
}, state}
end
@impl true
def handle_info({:DOWN, _monitor, :process, pid, _reason}, state) do
{:noreply, forget(state, pid)}
end
def handle_info(_message, state), do: {:noreply, state}
## Slots
defp hold(state, pid, monitor, waited, class) do
slot = %{monitor: monitor, started_at: System.monotonic_time(), class: class}
state = %{state | held: Map.put(state.held, pid, slot)}
emit(state, :acquired, %{wait: waited}, class)
end
defp enqueue(state, pid, class) do
waiter = %{monitor: Process.monitor(pid), queued_at: System.monotonic_time(), class: class}
state = %{
state
| waiters: Map.put(state.waiters, pid, waiter),
order: Map.update!(state.order, class, &:queue.in(pid, &1))
}
emit(state, :queued, %{}, class)
end
defp reject(state, class) do
retry_after = estimate(state)
{retry_after, emit(state, :rejected, %{retry_after: retry_after}, class)}
end
# The lowest-class waiter this arrival outranks, newest first, or `nil` if it
# outranks nothing queued. `@evictable` is scanned back to front — lowest
# class first — and `interactive` is not in it, so no amount of rank
# arithmetic can reach a listener.
defp victim(state, class) do
@evictable
|> Enum.reverse()
|> Enum.take_while(&(&1 != class))
|> Enum.find_value(fn candidate ->
case :queue.peek_r(state.order[candidate]) do
{:value, pid} -> {candidate, pid}
:empty -> nil
end
end)
end
# Drops the displaced waiter and tells it so. The reply is distinct from
# queue-full because the two are distinguishable to the caller — it *was*
# waiting, and something outranked it — even though both mean "come back".
#
# `victim/2` only *peeks*, and the pop happens here against the state this
# function was handed. The two ran back to back against one state either way,
# but carrying a pre-computed queue between them would mean any future edit
# landing in that gap was silently discarded. The pin on `pid` asserts the
# two agree.
defp displace(state, {class, pid}) do
{{:value, ^pid}, rest} = :queue.out_r(state.order[class])
{waiter, waiters} = Map.pop!(state.waiters, pid)
Process.demonitor(waiter.monitor, [:flush])
state = %{state | waiters: waiters, order: Map.put(state.order, class, rest)}
retry_after = estimate(state)
send(pid, {__MODULE__, {:displaced, retry_after}})
emit(state, :displaced, %{retry_after: retry_after}, class)
end
# One function for every way a process stops mattering to this module —
# `release/1`, a `DOWN`, or a caller that timed out — so that the holder and
# waiter cases cannot drift apart. Unknown pids are `:ok` by omission, which
# is what makes releasing twice harmless.
defp forget(state, pid) do
cond do
slot = Map.get(state.held, pid) ->
Process.demonitor(slot.monitor, [:flush])
held = Map.delete(state.held, pid)
duration = System.monotonic_time() - slot.started_at
%{state | held: held, samples: sample(state.samples, duration)}
|> emit(:released, %{duration: duration}, slot.class)
|> grant_next()
waiter = Map.get(state.waiters, pid) ->
Process.demonitor(waiter.monitor, [:flush])
# Deleted from `order` too, not left as a tombstone to be skipped on the
# next grant. Tombstones are only consumed by a grant, so a saturated
# semaphore whose waiters all give up accumulates them without bound —
# measured at 200 dead entries for 200 abandoned waiters — and
# `AP_QUEUE_SIZE` stops bounding what the queue costs. `:queue.delete/2`
# is O(length), but length is now exactly `map_size(waiters)` and so at
# most `AP_QUEUE_SIZE`, on a path that runs once per abandonment.
%{
state
| waiters: Map.delete(state.waiters, pid),
order: Map.update!(state.order, waiter.class, &:queue.delete(pid, &1))
}
|> emit(:abandoned, %{}, waiter.class)
true ->
state
end
end
# Grants as many waiters as the freed capacity allows. Normally one — a single
# release frees a single slot — but capacity is read from config, so a raised
# `AP_MAX_CONCURRENCY` must not leave waiters parked behind a limit that no
# longer applies.
defp grant_next(state) do
if map_size(state.held) < capacity(state) do
case next_waiter(state) do
{nil, state} ->
state
{{pid, waiter}, state} ->
send(pid, {__MODULE__, :granted})
state
|> hold(pid, waiter.monitor, System.monotonic_time() - waiter.queued_at, waiter.class)
|> grant_next()
end
else
state
end
end
# Pops the oldest waiter of the highest non-empty class. `order` and
# `waiters` hold exactly the same pids — `enqueue/3` adds to both, `forget/2`
# and `displace/2` remove from both — so a pid in a queue is always still
# waiting, and `Map.pop!/2` asserts that rather than papering over a drift
# between the two.
defp next_waiter(state), do: next_waiter(state, @classes)
defp next_waiter(state, []), do: {nil, state}
defp next_waiter(state, [class | lower]) do
case :queue.out(state.order[class]) do
{:empty, _rest} ->
next_waiter(state, lower)
{{:value, pid}, rest} ->
{waiter, waiters} = Map.pop!(state.waiters, pid)
{{pid, waiter}, %{state | order: Map.put(state.order, class, rest), waiters: waiters}}
end
end
## Retry-After
defp sample(samples, duration), do: Enum.take([duration | samples], @samples)
defp estimate(state) do
depth = map_size(state.waiters)
capacity = capacity(state)
# `depth + 1` counts the caller being rejected: it is being told when to come
# back, and everyone already queued is in front of it.
estimate = average_hold_ms(state) * (depth + 1) / capacity / 1_000
estimate |> ceil() |> max(1) |> min(Config.get(:render_timeout))
end
defp average_hold_ms(%__MODULE__{samples: []}), do: @default_hold_ms
defp average_hold_ms(%__MODULE__{samples: samples}) do
native = Enum.sum(samples) / length(samples)
System.convert_time_unit(trunc(native), :native, :millisecond)
end
## Telemetry
defp emit(state, event, measurements, class) do
occupancy =
Map.merge(depths(state), %{
held: map_size(state.held),
queued: map_size(state.waiters)
})
:telemetry.execute(
@telemetry ++ [event],
Map.merge(measurements, occupancy),
%{capacity: capacity(state), queue_size: queue_size(state), class: class}
)
state
end
# `:queue.len/1` is O(length) and this runs on every event, but every queue is
# bounded by `AP_QUEUE_SIZE` and their lengths sum to `map_size(waiters)` —
# so this is the same walk the module already pays for an abandonment, once
# per event rather than once per waiter.
defp depths(state) do
Map.new(@classes, fn class -> {:"queued_#{class}", :queue.len(state.order[class])} end)
end
## Limits
defp capacity(%__MODULE__{capacity: nil}), do: Config.get(:max_concurrency)
defp capacity(%__MODULE__{capacity: capacity}), do: capacity
defp queue_size(%__MODULE__{queue_size: nil}), do: Config.get(:queue_size)
defp queue_size(%__MODULE__{queue_size: queue_size}), do: queue_size
end