Channels in Rust: message passing with mpsc

How to send data between threads with std::sync::mpsc channels in Rust: send and recv, multiple producers, iterating the receiver, and bounded channels.

A channel is a queue that threads use to send values to each other, instead of sharing memory. "Do not communicate by sharing memory; share memory by communicating." Rust's standard channel lives in std::sync::mpsc, which stands for multiple producer, single consumer: many senders, one receiver.

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

fn main() {
    let (tx, rx) = mpsc::channel();
    let producer = thread::spawn(move || {
        for job in ["resize", "compress", "upload"] {
            tx.send(job.to_string()).unwrap();
        }
    });
    for msg in rx {
        println!("got {msg}");
    }
    producer.join().unwrap();
    println!("all senders gone, loop ended");
}

got resize
got compress
got upload
all senders gone, loop ended

• mpsc::channel() returns a (Sender<T>, Receiver<T>) pair. The type T is inferred from what you send.
• tx.send(value) moves the value into the channel; the sending thread can't use it afterwards. That's how ownership travels between threads without locks.
• rx.recv() blocks until a value arrives. It returns Err once every Sender has been dropped and the queue is empty.
• for msg in rx (or rx.iter()) receives until that happens, so the loop ends by itself when all producers finish.
• rx.try_recv() doesn't block; rx.recv_timeout(d) waits at most d.

Multiple producers

Clone the sender for each producer thread: let tx2 = tx.clone();. Messages from one producer arrive in the order it sent them, but messages from different producers interleave in whatever order the threads happen to run. If output order matters, tag each message (with a worker id or a sequence number) and sort after receiving.

The classic hang: an extra sender

The receiving loop ends only when all senders are dropped. If you clone tx for each worker but keep the original alive in main, the loop waits forever. Either move the original into the last worker, or call drop(tx) after spawning.

Bounded channels and backpressure

mpsc::channel() is unbounded: a fast producer can fill memory. mpsc::sync_channel(n) holds at most n messages, and send blocks when it's full, slowing the producer to the consumer's pace (backpressure). sync_channel(0) is a rendezvous: each send waits for a matching receive.

Channels vs Arc<Mutex>

Use channels when work flows in one direction (producers to a consumer, a pipeline of stages, workers reporting results). Use shared state when many threads need to read and update the same structure. Channels are often easier to reason about because each value has one owner at a time.

Errors

• send returns Err(SendError(value)) if the receiver was dropped, giving the value back.
• recv returns Err(RecvError) when the channel is empty and every sender is gone.

Beyond std

crossbeam-channel offers multi-consumer channels and select! over several channels. In async code, use tokio::sync::mpsc (with .send(x).await and .recv().await) so waiting doesn't block a runtime thread.

Example

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

fn checksum(data: &str) -> u32 {
    data.bytes().map(u32::from).sum()
}

fn main() {
    let files = vec!["config.toml", "main.rs", "README.md", "Cargo.lock", "lib.rs"];
    let (tx, rx) = mpsc::channel();

    let mut handles = Vec::new();
    for (worker, chunk) in files.chunks(2).enumerate() {
        let tx = tx.clone();
        let chunk: Vec<String> = chunk.iter().map(|s| s.to_string()).collect();
        handles.push(thread::spawn(move || {
            for name in chunk {
                let sum = checksum(&name);
                tx.send((name, worker, sum)).expect("receiver is alive");
            }
        }));
    }
    drop(tx); // only the workers' clones remain, so the loop below can end

    let mut results: Vec<(String, usize, u32)> = rx.iter().collect();
    for h in handles {
        h.join().unwrap();
    }
    results.sort(); // arrival order varies between runs; sort for stable output
    for (name, worker, sum) in &results {
        println!("{name:<12} worker {worker} checksum {sum}");
    }

    let (btx, brx) = mpsc::sync_channel::<u32>(1);
    let producer = thread::spawn(move || {
        for n in 1..=3 {
            btx.send(n * n).unwrap(); // blocks while the buffer is full
        }
    });
    let squares: Vec<u32> = brx.iter().collect();
    producer.join().unwrap();
    println!("bounded channel delivered {squares:?}");
}

Output:

Cargo.lock   worker 1 checksum 963
README.md    worker 1 checksum 685
config.toml  worker 0 checksum 1120
lib.rs       worker 2 checksum 586
main.rs      worker 0 checksum 696
bounded channel delivered [1, 4, 9]

Practice it

• Lesson: Sharing data between threads