🎓 A1Learner
🔍

Rust Backpressure

Backpressure is the pushback a slow consumer gives a fast producer. Backpressure keeps that waiting line from growing forever.

The valve in one line

send().await returns once the item is queued. A full pipe makes it wait right here. That wait is the whole mechanism.

tx.send("job").await; // waits here while the pipe is full

try_send refuses instead

Waiting is not always right. A stale reading can be worth less than a fresh slot. try_send fails at once on a full pipe.

let (tx, _rx) = mpsc::channel::<&str>(1);
tx.send("first").await.unwrap();
match tx.try_send("second") {
    Ok(()) => println!("queued"),
    Err(_) => println!("pipe full — dropped"),
}

throttle paces a stream

throttle lets one item pass, then holds the next for the set time. A stream holding a timer is not Unpin. pin! holds it in place, like the timeout in the Streams lesson.

let mut s = tokio_stream::iter(vec!["a", "b"]).throttle(Duration::from_millis(10));
tokio::pin!(s);
println!("item = {:?}", s.next().await);

A cap inside a stream

The Buffered Streams lesson set this valve. buffer_unordered(4) never runs more than four item futures at once. The stream itself becomes the slow consumer.

let results: Vec<_> = stream::iter(jobs)
    .map(|job| fetch(job))
    .buffer_unordered(4)
    .collect()
    .await;

Example

// Cargo.toml: tokio = { version = "1", features = ["full"] }
use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<i32>(2);

    let producer = tokio::spawn(async move {
        for n in 1..=4 {
            tx.send(n).await.unwrap();
            println!("sent {n}");
        }
    });

    let consumer = tokio::spawn(async move {
        while let Some(n) = rx.recv().await {
            println!("took {n}");
            sleep(Duration::from_millis(15)).await;
        }
        println!("consumer finished");
    });

    producer.await.unwrap();
    consumer.await.unwrap();
}

Key points

  • Backpressure: the pushback a slow consumer gives a fast producer
  • Use a bounded channel as the valve: send().await pauses the producer while the pipe is full
  • Use an unbounded channel only when a pile-up is truly harmless: nothing ever slows the producer there
  • Use try_send when dropping an item beats waiting for a slot
  • Use throttle to pace a stream to a fixed rate
  • Backpressure moves the wait.
  • It never deletes it.
  • The producer waits, or the caller waits, or the user waits.

In this example, the pipe holds two jobs. The producer sends four. The consumer needs 15 milliseconds per job. The first two sends fill the pipe. Their prints land at once. The third send waits. It prints only after the consumer takes job 1. Every later send waits for a taken job. The prints show the valve at work: the producer never runs more than two jobs ahead.

Note

Someone always waits.

Try it Yourself

Edit the code, then press Run.

Running is simulated · Rust Playground

Exercise

A producer fills an unbounded channel far faster than the consumer drains it. Nothing crashes today. What is quietly growing?