Message Passing in Rust With std mpsc Channels Step by Step
std::sync::mpsc moves values between threads through a channel. Learn send, recv, iterating, multiple producers and what a closed channel means.
A worker thread that waits for jobs forever, a main thread that waits for the worker to finish, and a program that never exits: the most common mpsc bug is a receiver that never learns the senders are gone. A channel is a queue with two ends, and the rules about when the receiving end stops waiting are the part worth learning. This page uses std::sync::mpsc from Rust 1.99 with edition 2024. No crates are involved.
One sender, one receiver
mpsc::channel() returns a (Sender<T>, Receiver<T>) pair. "mpsc" means multiple producer, single consumer: you can clone the sender as often as you like, and there is exactly one receiver.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
let producer = thread::spawn(move || {
for n in 1..=3 {
tx.send(n * 10).expect("receiver is gone");
}
// tx is dropped here, which closes the channel
});
for value in rx {
println!("got {value}");
}
producer.join().unwrap();
}got 10
got 20
got 30
The for loop calls recv() repeatedly and ends when it returns an error, which happens when every Sender has been dropped and the queue is empty. That is the closing rule, and it is the one to remember: a channel closes when its last sender goes away, not when the sender says "done". Nothing in the producer announces completion. Dropping tx at the end of the closure is the announcement.
The hang: a sender that is never dropped
Here is the same program with two producers, written the obvious way.
// hangs: the original sender is still alive in main
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
for id in 0..2 {
let tx = tx.clone();
thread::spawn(move || {
tx.send(format!("worker {id} finished")).unwrap();
});
}
for msg in rx {
println!("{msg}");
}
println!("never reached");
}worker 1 finished
worker 0 finished
Then it sits there until you press Ctrl-C. Each worker owns a clone, and both are dropped when the workers finish. But main still holds the original tx, so the channel is not closed, and for msg in rx waits for a third message that no one can send. The fix is to drop the original once the clones are made.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
for id in 0..2 {
let tx = tx.clone();
thread::spawn(move || {
tx.send(format!("worker {id} finished")).unwrap();
});
}
drop(tx);
let mut messages: Vec<String> = rx.into_iter().collect();
messages.sort();
println!("{messages:?}");
}["worker 0 finished", "worker 1 finished"]
Sorting matters: with several producers the arrival order depends on scheduling, so a test or a log that depends on it will be flaky. The channel preserves order per sender and nothing across senders.
Sending moves the value
A value sent through a channel is moved into it. That is the point of the design, since the receiver becomes the only owner, and it also produces a familiar compile error when you try to use the value afterwards.
// does not compile
use std::sync::mpsc;
fn main() {
let (tx, _rx) = mpsc::channel();
let report = String::from("quarterly numbers");
tx.send(report).unwrap();
println!("sent: {report}");
}error[E0382]: borrow of moved value: `report`
--> src/main.rs:8:22
|
6 | let report = String::from("quarterly numbers");
| ------ move occurs because `report` has type `String`, which does not implement the `Copy` trait
7 | tx.send(report).unwrap();
| ------ value moved here
8 | println!("sent: {report}");
| ^^^^^^ value borrowed here after move
|
help: consider cloning the value if the performance cost is acceptable
|
7 | tx.send(report.clone()).unwrap();
| ++++++++Clone before sending if you still need the data, or print before you send. The error is explained in more detail in the E0382 page.
Sending to a receiver that is gone
The mirror image of the closing rule: if the Receiver is dropped, send fails, and the error gives the value back so nothing is lost silently.
use std::sync::mpsc;
fn main() {
let (tx, rx) = mpsc::channel::<String>();
drop(rx);
match tx.send(String::from("anyone there?")) {
Ok(()) => println!("sent"),
Err(e) => println!("send failed, value handed back: {:?} ({e})", e.0),
}
}send failed, value handed back: "anyone there?" (sending on a closed channel)
That error is how a producer learns that its consumer has shut down, and it is why the earlier examples call .unwrap() or .expect() only where a missing receiver would be a bug. A long-lived producer should treat Err as a signal to stop its loop, not as something to retry: nobody will ever read what it sends.
A pool of workers on one job queue
The standard receiver cannot be cloned, so a common pattern gives every worker its own channel and deals jobs out round-robin, then collects results on a second channel that all workers share through cloned senders.
use std::sync::mpsc;
use std::thread;
fn main() {
let (result_tx, result_rx) = mpsc::channel::<(usize, u64)>();
let mut job_senders = Vec::new();
let mut workers = Vec::new();
for id in 0..3 {
let (job_tx, job_rx) = mpsc::channel::<u64>();
job_senders.push(job_tx);
let result_tx = result_tx.clone();
workers.push(thread::spawn(move || {
for n in job_rx {
result_tx.send((id, n * n)).unwrap();
}
}));
}
drop(result_tx);
for (i, n) in (1..=9u64).enumerate() {
job_senders[i % 3].send(n).unwrap();
}
drop(job_senders); // closes every job channel, so the workers' loops end
let mut squares: Vec<u64> = result_rx.iter().map(|(_, sq)| sq).collect();
squares.sort();
println!("{squares:?}");
for w in workers {
w.join().unwrap();
}
}[1, 4, 9, 16, 25, 36, 49, 64, 81]
Both closing rules are at work. Dropping job_senders ends each worker's for n in job_rx loop, which drops that worker's clone of result_tx. Once all three are gone, and the original was dropped earlier, result_rx.iter() ends. Forget either drop and the program hangs at the collect line. The round-robin deal does not balance uneven jobs; a shared queue behind a lock or a work-stealing library does.
Waiting without blocking forever
recv() blocks. Two siblings let you stay in control: try_recv() returns immediately, and recv_timeout(d) waits up to a duration. Both distinguish "nothing yet" from "channel closed".
use std::sync::mpsc::{self, RecvTimeoutError, TryRecvError};
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel::<&str>();
println!("{:?}", rx.try_recv());
let sender = thread::spawn(move || {
thread::sleep(Duration::from_millis(50));
tx.send("late message").unwrap();
});
match rx.recv_timeout(Duration::from_millis(10)) {
Err(RecvTimeoutError::Timeout) => println!("nothing after 10 ms"),
other => println!("unexpected: {other:?}"),
}
println!("{:?}", rx.recv_timeout(Duration::from_millis(500)));
sender.join().unwrap();
match rx.try_recv() {
Err(TryRecvError::Disconnected) => println!("channel closed"),
other => println!("unexpected: {other:?}"),
}
}Err(Empty)
nothing after 10 ms
Ok("late message")
channel closed
The first Err(Empty) means the sender is alive but has sent nothing. The last one, Disconnected, means every sender is gone and the queue is empty. A polling loop that treats both as "try again later" will spin forever after the producers finish, which is the same bug as the hang above wearing a different shirt.
Bounded channels apply backpressure
channel() is unbounded: a fast producer and a slow consumer fill memory. sync_channel(n) holds at most n items, and send blocks when it is full, which slows the producer to the consumer's pace.
use std::sync::mpsc;
use std::thread;
use std::time::{Duration, Instant};
fn main() {
let (tx, rx) = mpsc::sync_channel::<u32>(2);
let start = Instant::now();
let producer = thread::spawn(move || {
for n in 0..6 {
tx.send(n).unwrap();
println!("sent {n} at {:>3} ms", start.elapsed().as_millis());
}
});
for n in rx {
thread::sleep(Duration::from_millis(30));
println!("handled {n}");
}
producer.join().unwrap();
}sent 0 at 0 ms
sent 1 at 0 ms
sent 2 at 0 ms
handled 0
sent 3 at 30 ms
handled 1
sent 4 at 60 ms
handled 2
sent 5 at 90 ms
handled 3
handled 4
handled 5
The producer raced through the first items because the buffer had room (two slots plus the one in flight to the consumer), then slowed to one item per 30 ms, the consumer's rate. Interleaving of the sent and handled lines can vary between runs; the timestamps are what show the throttling. sync_channel(0) is a rendezvous channel where each send waits for a matching receive.
One channel, several kinds of message
A channel carries one type, so when a worker must handle different instructions, make that type an enum. Each variant can carry its own data, and match forces the receiver to deal with every kind, including the one that says "stop".
use std::sync::mpsc;
use std::thread;
enum Command {
Add(i64),
Reset,
Report(mpsc::Sender<i64>),
Shutdown,
}
fn main() {
let (tx, rx) = mpsc::channel();
let worker = thread::spawn(move || {
let mut total = 0;
for cmd in rx {
match cmd {
Command::Add(n) => total += n,
Command::Reset => total = 0,
Command::Report(reply) => {
let _ = reply.send(total);
}
Command::Shutdown => break,
}
}
total
});
for n in [5, 10, 20] {
tx.send(Command::Add(n)).unwrap();
}
let (reply_tx, reply_rx) = mpsc::channel();
tx.send(Command::Report(reply_tx)).unwrap();
println!("running total: {}", reply_rx.recv().unwrap());
tx.send(Command::Reset).unwrap();
tx.send(Command::Add(7)).unwrap();
tx.send(Command::Shutdown).unwrap();
println!("final total: {}", worker.join().unwrap());
}running total: 35
final total: 7
Two things are worth copying. The Report variant carries a Sender, so the caller creates a private reply channel and the worker answers on it, which is how you get a return value out of a thread that owns its state. And Shutdown exits the loop explicitly, so the worker stops even if some sender is still alive elsewhere. That removes the dependence on every clone being dropped, at the price of having to remember to send it. This single-owner thread with a command channel is the shape the actor model grows from (actors in Tokio).
When mpsc is not the right queue
- Several consumers need to share one queue. The standard receiver cannot be cloned; use a crossbeam or flume channel, or put the receiver behind
Arc<Mutex<_>>, which serializes everyrecv. - Async code.
recv()blocks a thread, which stalls an async worker. Use the channels intokio::sync(comparing Tokio channels). - A hot path with millions of tiny messages per second. Every send takes a lock-free queue slot and may wake the receiver, so batching items into a
Vecper message usually beats one message per item, and you should measure before you pick a granularity. - Many threads updating one counter. A channel moves data between threads, but shared state is better served by an atomic or a lock (atomics, Arc and Mutex).
