๐ŸŽ“ A1Learner
๐Ÿ”

Rust Multiple Producers

Multiple producers means many sender threads feeding one receiver on the same channel. The mpsc name means many producers, one consumer.

Clone first, then move one sender into each thread

tx.clone() makes another sender for the same channel. Make each clone first, then move one sender into each thread. Arrival order across threads is not promised, so this program sorts before printing โ€” the same every run:

use std::sync::mpsc;
use std::thread;
fn main() {
    let (tx, rx) = mpsc::channel();
    let tx2 = tx.clone();
    thread::spawn(move || tx.send("from first").unwrap());
    thread::spawn(move || tx2.send("from second").unwrap());
    let mut got: Vec<&str> = rx.iter().collect();
    got.sort();
    println!("arrived: {got:?}");
}

One sender's messages keep their order

Reading with rx.iter() (or a for loop) collects from every producer and ends when the last sender is dropped:

use std::sync::mpsc;
fn main() {
    let (tx, rx) = mpsc::channel();
    for n in [1, 2, 3] { tx.send(n).unwrap(); }
    drop(tx);
    let got: Vec<i32> = rx.iter().collect();
    println!("in order: {got:?}");
}

A sender left alive keeps the receiver waiting

Give every sender away or drop it: while any sender still lives, the receiver keeps waiting for more:

// every worker thread has finished, yet this loop never ends:
for msg in rx { /* ... */ }
// the likely cause: main never dropped its own original tx.
drop(tx); // the loop can end only once EVERY sender is gone

Each send still moves its value

Exactly as with a single sender: the value now belongs to the receiver:

let word = String::from("news");
tx.send(word).unwrap();
// println!("{word}"); // refused: word now belongs to the receiver

Example

use std::sync::mpsc; use std::thread;
fn main() {
    let (tx, rx) = mpsc::channel();
    let tx2 = tx.clone();
    thread::spawn(move || { tx.send("hi").unwrap(); tx.send("from").unwrap(); });
    thread::spawn(move || { tx2.send("more").unwrap(); tx2.send("news").unwrap(); });
    let mut got: Vec<&str> = rx.iter().collect();
    got.sort();
    println!("got {} messages: {:?}", got.len(), got);
}

Key points

  • Many producers, one consumer: every sender clone feeds the same single receiver.
  • Messages from one sender keep their order.
  • Messages from different senders can arrive in any order.
  • Code that needs an order should sort the messages, or number them, or collect results and arrange them at the end.

In this example, tx.clone() makes a second sender, tx2, for the same channel. The first thread moves tx and sends "hi" and "from". The second thread moves tx2 and sends "more" and "news". Main owns no sender now, so rx.iter().collect() gathers messages until both senders are dropped at the end of their threads โ€” all 4 arrive. The order across the two threads is not promised. So the code sorts before printing. The shown output is the same on every run.

Note

Arrival order across producers is not promised.

Try it Yourself

Edit the code, then press Run.

Running is simulated ยท Rust Playground

Exercise

Your for msg in rx loop never ends even though every worker thread finished. What is the likely cause?