The firehose problem
You are building an async web scraper in Rust. It fetches a list of URLs, spawns a task for each one, downloads the HTML, and pushes the results into a channel for a downstream parser. In Python or JavaScript, you might reach for asyncio.gather or Promise.all. The code runs fast until it does not. Your memory usage climbs linearly with the number of URLs. The garbage collector panics. The process gets killed by the OOM killer.
Rust async tasks are cheap to spawn, but they are not free. Every task holds a future, a state machine, and a stack frame. Every message you push into an unbounded channel sits in a heap allocation until someone reads it. When the producer outpaces the consumer, the queue grows until your RAM runs out. The system does not politely ask you to slow down. It exhausts your memory and crashes.
Backpressure is the mechanism that stops this from happening. It forces the fast side to wait for the slow side. In async Rust, you do not implement backpressure with custom locks or manual semaphore counters. You get it for free by picking the right channel type.
What backpressure actually is
Backpressure is just flow control applied to data streams. Think of a garden hose connected to a sprinkler. If you open the tap fully, water rushes out. If you step on the hose, the pressure builds upstream. The water stops flowing until you lift your foot. The hose itself does not stretch to hold infinite water. It forces the tap to match the sprinkler's pace.
In async Rust, an unbounded channel is like a hose that stretches infinitely. It accepts every message you throw at it. A bounded channel is the hose with a fixed length. Once it is full, the next send operation cannot proceed. The producer task yields control back to the runtime. It sleeps until the consumer reads a message and frees up a slot. The producer wakes up, sends the next message, and the cycle continues.
This behavior is cooperative. The runtime does not preempt your task. Your code explicitly says await at the send boundary. The scheduler notes that the task is waiting for channel capacity, marks it as parked, and runs something else. When capacity opens, the scheduler unparks the task. No spinning. No wasted CPU cycles. Just a clean pause.
How the channel enforces the limit
The magic happens inside the await keyword. When you call tx.send(msg).await, the future checks the internal queue length. If there is space, it pushes the message and returns Poll::Ready(Ok(())). The task continues immediately.
If the queue is full, the future registers a waker with the channel. A waker is a lightweight callback that tells the runtime how to resume this specific task. The future returns Poll::Pending. Tokio parks the task on a wait queue. The CPU moves on to other work.
When the consumer calls rx.recv().await and pulls a message out, the channel decrements its internal count. It sees that a sender is waiting. It calls the waker. The runtime unparks the sender task and places it back on the ready queue. On the next scheduler tick, the sender resumes, pushes its message, and returns Ready.
This waker registration is what makes async backpressure efficient. You are not polling the channel in a loop. You are not burning CPU cycles checking a flag. The runtime handles the wake-up signal directly. The producer only runs when it can actually make progress.
The bounded channel solution
Tokio provides tokio::sync::mpsc for multi-producer, single-consumer channels. The mpsc name comes from classic concurrency literature. It means multiple tasks can hold a sender, but only one task holds the receiver. The channel capacity is the backpressure valve.
use tokio::sync::mpsc;
/// Demonstrates bounded channel backpressure with a producer and consumer.
async fn run_pipeline() {
// Capacity of 10 means at most 10 messages sit in memory at once.
let (tx, mut rx) = mpsc::channel::<String>(10);
// Spawn the producer task.
let producer = tokio::spawn(async move {
for i in 0..100 {
let msg = format!("msg-{i}");
// Awaits if the channel is full. Yields to the runtime instead of spinning.
if tx.send(msg).await.is_err() {
eprintln!("Channel closed. Stopping producer.");
break;
}
}
});
// Spawn the consumer task.
let consumer = tokio::spawn(async move {
while let Some(msg) = rx.recv().await {
// Simulate slow processing to trigger backpressure.
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
println!("Processed: {msg}");
}
});
// Wait for both tasks to finish.
let _ = tokio::join!(producer, consumer);
}
The capacity parameter 10 is the only thing that enforces the limit. If you change it to 1000, the producer will blast through 1000 messages before the consumer even reads the first one. The backpressure still exists, but the threshold is higher. The key insight is that the channel capacity directly controls how much work is allowed to be in flight.
When tx.send(msg).await hits a full channel, the future returns Pending. Tokio parks the task. The consumer runs, reads a message, and frees a slot. The channel internally wakes the waiting sender. The sender resumes, pushes the message, and returns Ok. If the receiver is dropped while the sender is waiting, the await resolves to Err(SendError). The producer catches it and stops. The pipeline shuts down cleanly.
Trust the channel capacity. It is your memory budget.
Real-world async pipeline
Production code rarely uses a single sender and receiver. You usually have a fan-out pattern: one producer, multiple workers, one aggregator. Backpressure still applies, but it propagates upstream through the chain.
Consider a log ingestion service. A network task receives raw bytes. It deserializes them into structured events. A pool of worker tasks processes each event. A final task writes the results to a database. If the database is slow, the workers pile up. If the workers pile up, the deserialization channel fills. If that channel fills, the network task stops reading from the socket. The TCP buffer fills. The remote server slows down its transmission. The backpressure travels all the way to the source.
use tokio::sync::mpsc;
use std::time::Duration;
/// Represents a structured log event ready for processing.
struct LogEvent {
id: u64,
payload: String,
}
/// Spawns a worker pool that consumes events and applies backpressure upstream.
async fn run_worker_pool() {
// Bounded channel between network receiver and worker pool.
let (tx, mut rx) = mpsc::channel::<LogEvent>(50);
// Spawn three workers sharing the same receiver.
// The runtime distributes messages fairly among waiting tasks.
let workers = (0..3).map(|worker_id| {
tokio::spawn(async move {
while let Some(event) = rx.recv().await {
// Simulate variable processing time.
let delay = Duration::from_millis(10 + (worker_id * 5));
tokio::time::sleep(delay).await;
println!("Worker {worker_id} handled event {id}", id = event.id);
}
})
}).collect::<Vec<_>>();
// Producer simulates incoming network data.
let producer = tokio::spawn(async move {
for i in 0..1000 {
let event = LogEvent {
id: i,
payload: "sample-data".to_string(),
};
// Backpressure kicks in when the 50-slot buffer fills.
// The producer yields until a worker finishes and frees a slot.
if tx.send(event).await.is_err() {
break;
}
}
});
let _ = producer.await;
// Drop the sender to signal workers to stop.
drop(tx);
for worker in workers {
let _ = worker.await;
}
}
Notice the drop(tx) after the producer finishes. When the last sender is dropped, the channel closes. The recv().await calls in the workers return None. The loops exit. The tasks finish. This is the standard graceful shutdown pattern. You do not need to send a special quit message. The channel itself carries the lifecycle signal.
Keep your channel capacity proportional to your consumer's throughput. A capacity of one forces strict serialization. A capacity of one hundred allows burst absorption. Pick a number that matches your latency tolerance, not your peak memory.
When things go wrong
Backpressure fails when you bypass the await boundary or use the wrong channel type. The most common mistake is reaching for try_send when you actually need send. The try_send method returns Result immediately. If the channel is full, it returns Err(TrySendError::Full(msg)). Your code must handle the error, usually by dropping the message or retrying later. If you drop the message, you lost data. If you retry in a tight loop without yielding, you spin the CPU and starve other tasks. You defeated backpressure by writing a busy-wait loop.
Another trap is mixing synchronous blocking code inside an async task. If your consumer calls a blocking database driver or runs a heavy CPU calculation without tokio::task::spawn_blocking, it holds the runtime thread hostage. The channel fills up. The producer awaits. Nothing else runs. The entire application stalls. The borrow checker will not catch this. The runtime will not warn you. You just get a frozen process.
If you accidentally move a sender into multiple tasks without cloning it, the compiler stops you with E0382 (use of moved value). You must call tx.clone() for each task that needs to send. The community convention is to clone the sender before the loop or closure, then drop the original if you do not need it. Cloning a mpsc::Sender is cheap. It just increments an internal reference count.
// Cloning the sender is cheap and safe.
let tx_clone = tx.clone();
tokio::spawn(async move {
tx_clone.send("hello").await.unwrap();
});
If you forget to await the send, you get a compiler error about unused impl Future. If you use an unbounded channel by accident, you get no compiler error. You just get an OOM crash later. The type system does not distinguish between mpsc::UnboundedSender and mpsc::Sender at the API level. They share the same send method signature. The difference is purely in the runtime behavior.
Treat channel capacity as a contract. If you cannot justify the number, lower it.
Choosing your channel
Rust and Tokio offer several channel types. Picking the wrong one breaks backpressure or adds unnecessary overhead. Match the channel to your data flow pattern.
Use tokio::sync::mpsc::channel when you need multiple producers sending to a single consumer and require strict backpressure. Use tokio::sync::mpsc::unbounded_channel when the producer is extremely fast, the consumer is guaranteed to keep up, and you prioritize allocation speed over memory bounds. Use tokio::sync::oneshot::channel when exactly one message will ever be sent, such as waiting for a task to return a result or signaling cancellation. Use tokio::sync::watch::channel when you need to broadcast the latest value to multiple listeners, like updating a UI state or propagating configuration changes. Use tokio::sync::broadcast::channel when every listener must receive every message, even if they fall behind, accepting that slow listeners will eventually be dropped.
The decision matrix is straightforward. Bounded mpsc is the default for pipelines. oneshot is the default for task results. watch is the default for state updates. broadcast is the default for event logs. unbounded is the exception, not the rule.