Channels: Passing Messages Between Threads

Why this matters

Threads that never communicate are of limited use. One of the cleanest ways for threads to cooperate is message passing: instead of sharing memory, threads send each other values through a channel, like a pipe. A popular slogan (from Go) puts it: "do not communicate by sharing memory; share memory by communicating". In Rust, sending a value moves its ownership to the receiver, so the type system guarantees that no two threads use the same data at once.

A channel

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

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(format!("message {i}")).unwrap();
        }
    });
    for received in rx {
        println!("got {received}");
    }
    println!("channel closed");
}

got message 1
got message 2
got message 3
channel closed

• mpsc::channel() returns a pair: a sender (tx, transmitter) and a receiver (rx). mpsc stands for "multiple producer, single consumer": many senders, one receiver.
• tx.send(value) puts a value into the channel; it returns Err only if the receiver has been dropped.
• rx.recv() blocks until a value arrives, returning Ok(value), or Err once all senders are gone and the channel is empty. for x in rx (or rx.iter()) loops over values until that happens.
• The channel is unbounded: send never blocks; values queue up in order. Messages from one sender arrive in the order sent.

When the spawned thread finishes, its tx is dropped, the channel closes, and the for loop in main ends. Closing by dropping senders is the normal way to say "no more messages".

Sending moves ownership

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

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let report = String::from("all good");
        tx.send(report).unwrap();
        println!("sent {report}");
    });
    println!("{}", rx.recv().unwrap());
}

error[E0382]: borrow of moved value: `report`
 --> main.rs:9:24
  |
7 |         let report = String::from("all good");
  |             ------ move occurs because `report` has type `String`, which does not implement the `Copy` trait
8 |         tx.send(report).unwrap();
  |                 ------ value moved here
9 |         println!("sent {report}");
  |                        ^^^^^^^^ value borrowed here after move
  |

After send, the String belongs to the channel and then to the receiver. The sending thread cannot read or modify it any more, so the receiver can be certain nobody else is touching it. No locks are needed for the data itself.

Multiple producers

Clone the sender to give each thread its own:

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

fn main() {
    let (tx, rx) = mpsc::channel();
    for id in 0..3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let result = (id, (1..=10 * (id + 1)).sum::<u32>());
            tx.send(result).unwrap();
        });
    }
    drop(tx);
    let mut results: Vec<(u32, u32)> = rx.iter().collect();
    results.sort();
    for (id, sum) in results {
        println!("worker {id}: {sum}");
    }
}

worker 0: 55
worker 1: 210
worker 2: 465

Two details matter:

• drop(tx): the original sender, still owned by main, would keep the channel open forever, and rx.iter() would never end. Drop it once all clones are handed out.
• Sorting: messages from different threads arrive in whatever order the threads finish. Sort, or include an index, if the output order matters.

Pipelines

Channels connect stages, each running in its own thread, like a shell pipeline (cat | grep | sort):

 main --lines--> [parse thread] --numbers--> [sum thread] --total--> main

Each stage owns its receiver, processes values, and sends results onward; when its input closes, it finishes and drops its output sender, closing the next channel in turn. Shutdown propagates down the pipeline automatically. The second exercise builds exactly this.

Bounded channels and backpressure

mpsc::sync_channel(n) creates a channel that holds at most n messages; send blocks when it is full. This keeps a fast producer from piling up unlimited data in memory while a slow consumer catches up ("backpressure"). sync_channel(0) is a rendezvous: each send waits until the receiver takes the value.

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

fn main() {
    let (tx, rx) = mpsc::sync_channel(2);
    let producer = thread::spawn(move || {
        for i in 0..5 {
            tx.send(i).unwrap();
        }
    });
    let got: Vec<i32> = rx.iter().collect();
    producer.join().unwrap();
    println!("{:?}", got);
}

[0, 1, 2, 3, 4]

Common mistakes

• Forgetting to drop the original sender: the receiving loop hangs forever (in the exercises: "timed out").
• Using a value after sending it: E0382, by design.
• Relying on arrival order across threads: sort or tag messages.
• recv().unwrap() after all senders dropped: returns Err(RecvError) and the unwrap panics. Loop with for instead, or handle the Err.
• Two receivers: Receiver cannot be cloned (single consumer). To share work among several consumers, wrap the receiver in Arc<Mutex<...>> (next lesson) or use one channel per worker.

Recap

• let (tx, rx) = mpsc::channel(); creates a channel; tx.send(v) moves v in; rx.recv() or for v in rx takes values out.
• The channel closes when all senders are dropped; receiver loops then end.
• Clone tx for multiple producers; drop the original.
• Arrival order between different senders is unpredictable.
• sync_channel(n) bounds the queue and makes senders wait when it is full.

Going further: read the Rust Book chapter 16.2. Try rx.recv_timeout(Duration::from_millis(100)) and try_recv(), the non-blocking variants.