Channels, Semaphores, Synchronization, and Backpressure

Async makes it easy to create work faster than a model, GPU, decoder, database, or client can consume it. Backpressure is the mechanism that slows, rejects, sheds, or durably queues producers when capacity is exhausted.

An unbounded channel is not elasticity. It is a memory-backed outage timer.

Learning objectives

choose channels from ownership and delivery semantics;
distinguish queue capacity from concurrency capacity;
use bounded mpsc and explicit overload behavior;
use semaphores to represent in-flight scarce resources;
compare mutexes, watch, broadcast, oneshot, and messages;
avoid holding permits/locks across unrelated work;
design global/tenant/provider capacity hierarchy;
test full-queue behavior deterministically.

Dependency and mental model

Depends on futures and cancellation.

producer rate > consumer rate
        │
        ▼
bounded queue reaches capacity
        ├── await producer (backpressure)
        ├── reject/429/busy
        ├── shed low-priority work
        └── persist durable job

Every edge needs a stated choice.

Bounded MPSC

Tokio mpsc supports multiple producers and one receiver:

let (sender, receiver) = tokio::sync::mpsc::channel(capacity);

send().await waits for capacity. try_send returns immediately:

match sender.try_send(job) {
    Ok(()) => Accepted,
    Err(TrySendError::Full(job)) => RejectBusy(job),
    Err(TrySendError::Closed(job)) => Shutdown(job),
}

Keep the rejected job when the caller needs retry/persistence. Do not collapse full and closed.

The companion rejects zero capacity in bounded_channel; a rendezvous channel has different semantics and Tokio’s bounded MPSC requires positive capacity.

Queue capacity versus concurrency

A channel bounds waiting items. A semaphore bounds concurrent use:

queue capacity: 32 waiting jobs
worker count: 4 tasks
provider semaphore: 2 in-flight calls
GPU semaphore: 1 resident inference

If workers dequeue then wait indefinitely on a downstream semaphore, the effective queue includes blocked workers. Measure all stages.

Semaphore permits should cover exactly the scarce resource:

let permit = model_capacity.acquire_owned().await?;
let response = model.infer(request).await;
drop(permit);

Do not acquire GPU/provider permits before slow unrelated retrieval or hold them while writing results.

Channel choices

oneshot: one result from one producer.
mpsc: owned work messages to one receiver.
watch: latest state/value; slow readers may skip intermediate versions.
broadcast: each active receiver sees messages, but lagging receivers can lose old entries.
stream/SSE plus durable store: resumable important events.

Do not use broadcast as a durable audit log. Lag is an explicit error and history is bounded.

Message passing versus shared state

One-owner actor/worker plus commands makes sequential invariants clear:

enum Command {
    Submit(Job),
    Cancel(RunId),
    Shutdown,
}

Mutex/shared state is useful for short synchronous critical sections. Avoid:

one global application mutex;
guards across await;
lock-order cycles;
synchronous mutex in long async critical section.

Choose based on state ownership. Channels add queueing; locks add contention. Neither creates capacity.

Capacity hierarchy and fairness

Production limits may include:

global model calls;
per-provider quota;
per-model/GPU residency;
per-tenant concurrency/spend;
per-run tool calls;
per-media decoder;
database pool;
outgoing stream buffers.

Acquire multiple permits in one consistent order to avoid deadlock. Prefer a scheduler that makes one atomic admission decision rather than nesting many arbitrary awaits.

Fairness matters: one tenant can fill the global queue. Use per-tenant queues/weights/reservations and reject before large upload/inference allocations.

Overload is a product contract

Possible responses:

HTTP 429 with retry guidance for quota/rate;
503/busy for global temporary capacity;
accepted durable 202 job with position/status;
lower-quality/cheaper route under explicit policy;
drop nonessential progress events while preserving terminal/durable events;
refuse before charging/partial effects.

Do not wait forever. Queue timeout/deadline must be less than the run’s remaining deadline.

Backpressure through streams

provider stream → parser → verifier → durable trace → SSE

If client is slow:

pause upstream if flow control propagates;
buffer within a strict byte/event bound;
detach durable run from client;
cancel;
drop only explicitly nonessential events.

One-item MPSC capacity does not bound item bytes. Limit both.

Runnable companion implementation

cd rust/rust-ai-engineering
cargo run -q -p mosaic-runtime --example bounded_backpressure
cargo test -p mosaic-runtime --examples

Expected:

received: first; second: rejected_full

The example fills a capacity-one queue, proves the second item returns Full, then confirms the first was not lost.

Failure investigation: memory grows while GPU is saturated

inspect every upstream queue (including spawned futures/vectors);
count item bytes, not only items;
measure semaphore waiters and hold time;
locate dequeue-before-downstream-wait amplification;
bound admission before expensive decode;
add overload response and load test.

Changing the final channel to bounded does not help if an earlier vector already collected all requests.

Security and performance implications

Security:

attacker-controlled queues need per-principal quotas;
reject before large allocation/side effects;
never let low-priority progress crowd out terminal/audit events;
cancellation/closed channels must release permits;
prevent deadlock from permit acquisition order.

Performance:

too-small capacity reduces utilization under jitter;
too-large capacity raises tail latency/memory;
semaphore contention and queues require metrics;
batching can improve throughput but adds wait latency;
fairness scheduling costs work but prevents starvation.

Exercises

Add try_run to BoundedExecutor.
Build tenant plus global admission with stable acquisition order.
Add queue wait deadline and typed timeout.
Load-test variable-sized messages under a byte budget.
Compare watch/broadcast/durable event behavior for a lagging client.

Solution guidance

Return the original rejected work when possible. Record queue time separately from service time. For byte bounds, reserve declared/validated byte permits before enqueue and release on drop.

Check your understanding

Does a semaphore bound queued waiters?
Does bounded item count bound memory?
What does a watch receiver miss?
Why is broadcast not an audit log?
Where should a provider permit be acquired/released?

Completion checklist

Every queue has capacity, byte limit, deadline, and overload policy.
Scarce resources have independent concurrency limits.
Tenant fairness is explicit.
Locks/permits are not held across unrelated awaits.
Streams propagate or resolve slow-consumer pressure.
I ran the full-channel test.