Step 2 of 7

Multiple producers: cloning the Sender

The mp in mpsc: any number of threads can send into one channel. Each thread gets its own Sender by cloning it:

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

fn main() {
    let (tx, rx) = mpsc::channel();
    for worker in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let checked = worker * 100;
            tx.send((worker, checked)).unwrap();
        });
    }
    drop(tx);

    let mut results: Vec<(u32, u32)> = rx.iter().collect();
    results.sort();
    for (worker, checked) in results {
        println!("worker {worker} checked {checked} files");
    }
}

worker 1 checked 100 files
worker 2 checked 200 files
worker 3 checked 300 files

• let tx = tx.clone(); makes a new Sender for the same channel and shadows the name inside the loop, so the clone is what moves into each closure.
• Messages from different threads arrive in whatever order the threads happen to run. So each message carries who sent it, and main sorts before printing. Never rely on arrival order between producers.
• drop(tx) throws away main's own Sender. That line matters more than it looks.

The forgotten sender

The receiving loop ends only when every Sender is gone. The workers' clones are dropped when the workers finish, but the original tx belongs to main. Without drop(tx), main waits for a message from itself that never comes, and the program hangs forever:

let (tx, rx) = mpsc::channel();
for worker in 1..=3 {
    let tx = tx.clone();
    thread::spawn(move || tx.send(worker).unwrap());
}
// no drop(tx) here...
for n in rx {
    println!("{n}"); // prints 3 numbers, then waits forever
}

This is the most common channel bug. On this site it shows up as the time limit running out.

Bounded channels and backpressure

mpsc::channel() is unbounded: send never waits, so a fast producer can pile up millions of unread messages in memory. mpsc::sync_channel(n) holds at most n messages; when it's full, send waits until the receiver catches up. That's backpressure: the slow side automatically slows the fast side down. Production pipelines usually use bounded channels.

At work you'll also meet the crossbeam-channel crate (channels with several receivers, and waiting on several channels at once) and Tokio's async channels (module 33).

Common mistakes

• Forgetting drop(tx): the program hangs.
• Moving tx instead of cloning it in a loop: error[E0382]: use of moved value: `tx` with the note "value moved into closure here, in previous iteration of loop".
• Printing in arrival order from several producers: the output changes between runs. Collect, then sort (or use a BTreeMap).

Your task

Three store threads report their sales over one channel. Spawn one thread per store; each sends every sale as a (store, amount) message. Don't forget the original sender. The main thread adds the sales up per store in the BTreeMap and prints the totals (a BTreeMap iterates in key order):

east: 190
north: 245
south: 320

Challenge 2

This program prints nothing and then hits the time limit: it waits forever. Find out why and fix it by adding one line, so it prints sum of squares: 55.

Challenge 3

Write fn search_parallel(docs: Vec<String>, needle: &str) -> Vec<usize>. Spawn one thread per document (with thread::spawn); each thread sends its document's index over a channel if the document contains needle. Return the indices sorted ascending. A spawned thread can't borrow needle, so give each thread its own String copy of it.

Previous: Channels: sending messages between threads Next: Arc: shared ownership across threads