Articles / Rust

866 words

  • rust
  • tokio
  • mpsc
  • concurrency
  • async
  • backpressure

Backpressure in Rust Channels: tokio and std Under Load

The standard library and Tokio both ship a channel called mpsc, and they are not interchangeable. Both have a Sender and a Receiver, and both let multiple producers feed one consumer. It is tempting to think they differ only in whether you write .send(x) or .send(x).await.

The difference shows up the moment a consumer falls behind a producer. The two implementations then do completely different things to your program, and picking the wrong one on a busy Tokio runtime is a good way to stall every other task sharing that worker thread.

std::sync::mpsc::channel() has no limit at all

std::sync::mpsc::channel() is unbounded. There is no capacity argument because there is no capacity. A producer thread calling send never blocks, ever, regardless of how far behind the consumer is.

Under sustained load with a slow consumer this is not backpressure. It is an unbounded queue growing in memory until something else gives out.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel::<Vec<u8>>();

    thread::spawn(move || {
        loop {
            // Never blocks, no matter how far behind rx falls.
            tx.send(vec![0u8; 4096]).unwrap();
        }
    });

    // A slow consumer just lets the queue grow unbounded.
    loop {
        let _chunk = rx.recv().unwrap();
        thread::sleep(Duration::from_millis(10));
    }
}

The fix std actually provides is mpsc::sync_channel(bound).

It is the bounded sibling nobody reaches for by default. Once the channel is full, send blocks the calling OS thread until the receiver makes room. That is real backpressure, but it is backpressure implemented with a mutex and a condvar: the sending thread is parked by the OS scheduler and does nothing else until space appears.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u64>(16);

    thread::spawn(move || {
        for i in 0..1_000_000 {
            // Blocks the OS thread once the buffer of 16 fills up.
            tx.send(i).unwrap();
        }
    });

    for _ in 0..1_000_000 {
        let _ = rx.recv().unwrap();
    }
}

That is fine, arguably correct, for a dedicated producer thread. It is a disaster if that same blocking send is called from inside an async task running on a Tokio worker thread.

Tokio's bounded channel suspends the task, not the thread

tokio::sync::mpsc::channel(bound) looks similar on the surface, capacity argument and all. The mechanism under load is different, though, in a way that matters for anything running on an async runtime.

When the buffer is full, send(x).await does not park an OS thread. It registers the task as waiting on a semaphore permit and returns control to the executor, which is then free to run other tasks on that same worker thread. The backpressure is real, the producer genuinely stops making progress, but the thread underneath it keeps doing useful work for everyone else.

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<u64>(16);

    tokio::spawn(async move {
        for i in 0..1_000_000 {
            // Suspends this task at capacity; the worker thread
            // is freed to poll other tasks in the meantime.
            if tx.send(i).await.is_err() {
                break;
            }
        }
    });

    while let Some(_v) = rx.recv().await {
        sleep(Duration::from_millis(1)).await;
    }
}

Quick detour: why does this compose so well?

Under the hood, the bounded channel is built on an internal semaphore of permits equal to the capacity. Sending acquires a permit (or awaits one); receiving releases it back. This is the same primitive Tokio uses for its own rate limiting elsewhere.

That is why the channel plays cleanly with select!, timeouts and cancellation, none of which play nicely with a thread genuinely blocked inside the OS scheduler.

unbounded_channel() brings the std problem with it

Tokio also ships unbounded_channel(), and it inherits exactly the problem std's unbounded channel has: no capacity means no backpressure, full stop.

It exists for cases where you can prove the producer rate is naturally self-limiting (forwarding UI events, say), not as a default choice. If you find yourself reaching for it because the bounded version's backpressure is "annoying", that annoyance is the channel doing its job.

The real danger is a blocking send on a shared runtime

The genuinely dangerous case is not choosing the wrong channel in isolation, it is mixing them. Calling a std::sync::mpsc::SyncSender::send that blocks, directly inside an async fn running on a multi-threaded Tokio runtime, blocks the OS thread that runtime is using to poll every other task scheduled onto it.

Tokio has no visibility into that block. It looks identical to a task doing legitimate CPU work, except it never yields.

use std::sync::mpsc::SyncSender;

async fn forward(tx: SyncSender<u64>, value: u64) {
    // Blocks the worker thread outright: no await point, no yield,
    // nothing Tokio's scheduler can do about it.
    tx.send(value).unwrap();
}

With a handful of worker threads and enough tasks doing this concurrently, the runtime can genuinely wedge.

Every thread is stuck in a std blocking call, and no thread is free to poll the receiver task that would drain the channel and unblock them. The standard fix is tokio::task::spawn_blocking, which moves the blocking call onto a separate thread pool reserved for exactly this, keeping the async worker threads free.

use std::sync::mpsc::SyncSender;

async fn forward(tx: SyncSender<u64>, value: u64) {
    tokio::task::spawn_blocking(move || tx.send(value))
        .await
        .expect("blocking task panicked")
        .expect("channel closed");
}

Choose by which scheduler owns the sending thread

The rule of thumb is less about which channel is "better" and more about which scheduler owns the thread doing the sending.

  • For a pipeline of OS threads (CPU-bound work, blocking I/O, a thread pool reading files), std::sync::mpsc::sync_channel gives correct, simple, blocking backpressure, and there is no async runtime around to stall.
  • For a pipeline of async tasks on Tokio, tokio::sync::mpsc::channel is the one that composes: it suspends cleanly, works inside select!, and respects cancellation when a task is dropped mid-send.

Crossbeam's crossbeam-channel is worth a mention too, as a common third option for the pure-OS-thread case. It has the same blocking bounded semantics as sync_channel, but is faster under contention and has a select! macro of its own for threads rather than tasks.

It does not change the underlying tokio-vs-std distinction. It just gives the thread-based side of that choice a better implementation.

What does not work, in either direction, is treating "bounded channel" as a single concept and assuming the capacity number means the same thing regardless of which crate it came from. The number is the same. What happens to the caller when that number is reached is not.

← All articles