Backpressure in AI Pipelines: Bounded Channels and Flow Control in Tokio

An embedding service that looks fine under load and then dies at 3x traffic usually has the same bug: an unbounded queue between the HTTP handler and the GPU worker. Nothing failed, nothing errored - the queue just grew until memory ran out. The failure arrives far away from the decision that caused it.

Backpressure is the fix: make the slowest stage push back on the stages before it, so that overload shows up at the boundary of the system where you can still make a decision - accept, queue with a deadline, or shed. In Tokio this is mostly a matter of choosing the right primitive and then giving every bound a number you can defend.

Tokio 1.53.3 ships the pieces you need: bounded mpsc channels that park senders at capacity, Semaphore for limiting concurrent work on a shared resource, and AtomicUsize for cheap gauges. This post walks through why unbounded channels are the wrong default for inference workloads, how to size the ones you keep, how to propagate pressure all the way back to ingress, and a complete ingest -> embed -> index example you can adapt.

Unbounded queues hide overload

Tokio’s own tutorial is blunt about it: whenever concurrency or queuing is introduced, the queuing must be bounded, because “unbounded queues will eventually fill up all available memory and cause the system to fail in unpredictable ways.” The failure mode is not graceful degradation, it is an OOM kill in a process that reported no errors.

Two properties of async Rust make this worse than it looks:

• Async operations are lazy. loop { async_op(); } queues nothing, because .await never happens. Concurrency and queuing are only introduced when you explicitly reach for tokio::spawn, select!, join! or mpsc::channel. That is good - but it also means every bit of queuing in your pipeline was a deliberate choice somebody made.
• The unbounded variant never blocks. mpsc::unbounded_channel completes send immediately, so producers race ahead of consumers with no signal that they are losing. By contrast, a bounded channel parks the sender: once the capacity is reached, send(...).await waits until the receiver drains a slot.

That single difference decides whether your system tells you it is overloaded while you can still respond, or tells you with an OOM stack trace.

What the bound actually costs

A bounded channel is cheap. The mpsc implementation stores elements in blocks organized as a linked list: a block holds 32 messages on a 64-bit target (16 on 32-bit) regardless of message size, with about one byte of bookkeeping overhead per message. Capacity is not a per-slot allocation, so a capacity of 1024 does not reserve 1024 slots - it is a limit on how far ahead producers may run.

Primitive	Bound	Blocks when full	Use for
mpsc::channel(n)	Queue depth	send().await parks the producer	Between pipeline stages
mpsc::unbounded_channel()	None	Never	Almost nothing in a request path
Semaphore::new(k)	Concurrency	acquire().await parks the worker	GPU / model / DB concurrency
AtomicUsize gauge + try_*	Admission	Caller decides: accept or reject	Ingress admission control
Semaphore token bucket	Rate	acquire().await waits for refill	Requests per second ceilings

The two most common mistakes are using a channel bound to express concurrency (a queue of 256 with 64 workers still allows 256 in-flight documents if workers are spawned per message) and using a semaphore to express queueing depth (a semaphore with no queue behind it admits nothing and just stalls callers with no visibility).

Sizing the numbers

There is no universal right answer - Tokio’s tutorial says bound values are application specific and that picking good bounds is a big part of writing reliable Tokio applications. What you can do is derive them from two measurements: service time per item and acceptable latency.

Queue depth follows directly: capacity ~= arrival_rate * acceptable_queue_delay. If the embedder handles 200 docs/s and you are willing to absorb a 2-second burst before ingress starts rejecting, a capacity of 400 is defensible. If you are only willing to absorb 200 ms, the capacity is 40.

Concurrency follows from the resource: if one embedding call can be in flight per GPU shard and you have four shards, the semaphore takes 4 permits - not 64, and not the queue depth.

Bound	Too small	Too large
Channel capacity	Spurious rejections under normal bursts	Latency grows before rejection; memory follows
Semaphore permits	GPU idle while work queues elsewhere	OOM from concurrent batching, context thrash
Ingress send_timeout	Client retries amplify load	Requests sit long enough to be stale

Instrument before you tune: export queue depth - the channel’s remaining capacity versus its configured maximum (Sender::capacity and Sender::max_capacity in current Tokio releases) - as a gauge, along with semaphore.available_permits() as free capacity and the age of the oldest queued item. Queue depth is your early-warning signal; latency is the one your users feel.

Propagating pressure to ingress: queue, reject, or shed

Once the last stage is bounded, pressure travels backwards automatically: embed stops draining, the embed channel fills, ingest’s send parks, and the HTTP handler stops accepting new work. At that point you must choose a policy for the request in your hands.

• Queue. send().await with a deadline. The caller waits, and if the pipeline recovers, nothing was lost. Give it a timeout so a stuck stage cannot pin connections: tx.send_timeout(doc, Duration::from_millis(250)).
• Reject. try_send and translate Full into a 503 with Retry-After. Rejection is honest: the server tells the client the truth while it still can.
• Shed. Drop the lowest-value work - cache-miss re-embeddings, background reindex jobs - and keep interactive traffic. Shedding only works if the sheddable work is on its own bounded channel with its own priority.

The ingress handler is where all three meet:

use tokio::sync::mpsc;
use tokio::sync::mpsc::error::TrySendError;

async fn ingest(
    tx: mpsc::Sender<Doc>,
    doc: Doc,
) -> Result<u64, (u16, &'static str)> {
    match tx.try_send(doc) {
        Ok(()) => Ok(202),
        Err(TrySendError::Full(_)) => Err((503, "pipeline saturated")),
        Err(TrySendError::Closed(_)) => Err((500, "pipeline down")),
    }
}

Three details that matter in production:

1. Full and Closed are different failures. One is load, the other is an outage. Alerting on the wrong one trains you to ignore real incidents.
2. Backpressure is not a circuit breaker. A 429 from a model provider is often correct backpressure rather than an incident; do not let ingress rejection logic double as health detection, or a legitimate quota ceiling will look like an outage.
3. Sync callers need the sync API. From blocking code use blocking_send / blocking_recv on the same bounded channel instead of calling .await on a runtime thread you do not own.

A worked example: ingest -> embed -> index

The pipeline below bounds every hop, limits GPU concurrency with a semaphore, exposes gauges, and answers ingress with 503 instead of buffering without limit.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use tokio::sync::{mpsc, Semaphore};
use tokio::task::JoinSet;

#[derive(Debug)]
struct Doc {
    id: u64,
    text: String,
}

#[derive(Debug)]
struct Embedded {
    id: u64,
    vector: Vec<f32>,
}

struct Gauges {
    in_flight: AtomicUsize,
}

pub async fn run_pipeline(docs: Vec<Doc>) -> anyhow::Result<()> {
    // Stage bounds: ingress absorbs a short burst, later stages stay tight.
    let (ingest_tx, mut ingest_rx) = mpsc::channel::<Doc>(512);
    let (embed_tx, mut embed_rx) = mpsc::channel::<Embedded>(128);
    let (index_tx, mut index_rx) = mpsc::channel::<Embedded>(64);

    // Four concurrent embedding calls: one per GPU shard.
    let gpu = Arc::new(Semaphore::new(4));
    let gauges = Arc::new(Gauges { in_flight: AtomicUsize::new(0) });

    // Producer: respects the first bound instead of building a backlog.
    let producer = tokio::spawn(async move {
        for doc in docs {
            if ingest_tx.send(doc).await.is_err() {
                break; // downstream closed: stop, do not spin
            }
        }
    });

    // Stage 1 consumer. Backpressure point: the task parks here while
    // the semaphore has no permits, so ingest keeps draining into the
    // bounded channel only as fast as the GPU can take work.
    let embedder = {
        let gpu = Arc::clone(&gpu);
        let gauges = Arc::clone(&gauges);
        let embed_tx = embed_tx.clone();
        tokio::spawn(async move {
            let mut running: JoinSet<()> = JoinSet::new();
            while let Some(doc) = ingest_rx.recv().await {
                let permit = gpu.acquire_owned().await.expect("semaphore open");
                let embed_tx = embed_tx.clone();
                let gauges = Arc::clone(&gauges);
                running.spawn(async move {
                    gauges.in_flight.fetch_add(1, Ordering::Relaxed);
                    let vector = embed(&doc.text).await; // your model call
                    drop(permit); // concurrency released with the permit scope
                    gauges.in_flight.fetch_sub(1, Ordering::Relaxed);
                    let _ = embed_tx.send(Embedded { id: doc.id, vector }).await;
                });
            }
            while let Some(res) = running.join_next().await {
                res?;
            }
            anyhow::Ok(())
        })
    };
    drop(embed_tx);

    // Indexer: single writer, tightest bound.
    let indexer = tokio::spawn(async move {
        while let Some(item) = embed_rx.recv().await {
            write_vector(&item).await?;
        }
        anyhow::Ok(())
    });

    producer.await??;
    embedder.await??;
    // Dropping index_tx closes the channel: recv() now returns None.
    drop(index_tx);
    while index_rx.recv().await.is_some() {}
    indexer.await??;
    Ok(())
}

async fn embed(text: &str) -> Vec<f32> { vec![0.0; 384] }
async fn write_vector(item: &Embedded) -> anyhow::Result<()> { Ok(()) }

The shape is what matters, not the exact numbers: every arrow between stages is a bounded mpsc, and the only thing limiting GPU concurrency is a permit that is held for exactly the duration of the model call. Note that the permit is taken with acquire_owned before spawning the task and moved into it, so a task cannot exist without already holding admission rights - and note the placement of drop(permit), which happens after the model call but before the send: moving it after the send would accidentally widen the concurrency limit by the time the next stage takes.

Graceful shutdown and degradation

Backpressure design is incomplete without shutdown behavior, because the two interact: draining a full queue on SIGTERM is how deployments turn a rollout into data loss.

• Clean close. When the receiver should stop accepting, call Receiver::close() first, then drain what is buffered. Dropping the receiver outright discards every unread message and causes all senders to fail.
• Termination detection. recv() returns None once every Sender has been dropped and the buffer is drained - use that as your stage-completion signal rather than a timer.
• Degradation ladder. Under sustained pressure, shed in a fixed order: background re-embedding first, batch imports second, interactive traffic last. Each class gets its own channel so shedding is a routing decision, not an accident.
• Cancelling is not free. Cancelling Semaphore::acquire loses your place in the queue - the semaphore hands out permits in request order. A caller that repeatedly times out and retries can starve itself; prefer one long wait over a loop of short ones, and use try_acquire when the correct answer is “reject now”.

Checklist

• Replace every unbounded_channel in a request path with a bounded one, or justify it in a comment.
• Derive channel capacity from arrival rate times acceptable queue delay; derive semaphore permits from physical concurrency (GPU shards, connection pools).
• Keep queueing and concurrency separate: capacity for bursts, permits for the resource.
• Answer saturation at ingress with try_send, distinguishing Full from Closed, and map them to different status codes and alerts.
• Never .await a send without a deadline in a request handler; use send_timeout.
• Hold the semaphore permit for exactly the model call, and take it before spawning when tasks are spawned per item.
• Track queue depth, available permits and oldest-item age as metrics; alert on depth before latency.
• Close and drain deliberately on shutdown so backpressure does not become data loss.

Sources: Tokio tutorial: channels, tokio::sync::mpsc, tokio::sync::Semaphore, Tokio