Lesson 66 of 94
Channels: Message Passing
The previous lesson shared memory between threads and protected it with locks. The alternative is to not share at all: each piece of data is owned by one thread at a time, and threads communicate by sending messages. Go summarized it as "don't communicate by sharing memory; share memory by communicating", and it fits Rust's ownership model perfectly: sending a value moves it to the receiving thread.
Rust's standard channel is std::sync::mpsc: multi-producer, single-consumer. You get a Sender<T> (which you can clone) and a Receiver<T> (which you can't).
| Need | Rust |
|---|---|
| Unbounded queue | mpsc::channel() |
| Bounded queue of size n | mpsc::sync_channel(n) |
| Rendezvous (hand-off) | mpsc::sync_channel(0) |
| Put / take a message | tx.send(x) / rx.recv() |
| Poll without blocking / with a timeout | rx.try_recv() / rx.recv_timeout(d) |
| Tell consumers to stop | drop every Sender; the receiver's loop ends |
| Thread pool | a pool of threads reading jobs from a channel |
| Get a reply to a request | send a reply Sender along with the request |
The external crate crossbeam-channel adds multi-consumer channels and select!; the async world has its own channels (Async Channels & Locks). The ideas are the same everywhere.
1. A first channel
mpsc::channel() returns a (Sender, Receiver) pair. send puts a value into the channel; recv blocks until a value arrives.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
tx.send(String::from("hello from the worker")).unwrap();
});
let msg = rx.recv().unwrap(); // blocks until the message arrives
println!("main got: {msg}");
}main got: hello from the workerThe channel's element type is inferred from what you send: here Sender<String> and Receiver<String>.
2. Receiving until the channel closes
A Receiver is an iterator. for msg in rx receives until every Sender has been dropped, then the loop ends. No poison pill needed: the channel closing is the "done" signal.
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
for step in ["download", "unzip", "index"] {
tx.send(step).unwrap();
thread::sleep(Duration::from_millis(10));
}
// `tx` is dropped here, which closes the channel.
});
for step in rx {
println!("finished: {step}");
}
println!("channel closed, worker is done");
}finished: download
finished: unzip
finished: index
channel closed, worker is done3. Sending moves the value
send takes ownership of the value. After sending, the sending thread can't use it, so two threads can never touch the same data at once.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let report = vec![1, 2, 3];
tx.send(report).unwrap();
println!("sent {} rows", report.len());
});
println!("{:?}", rx.recv().unwrap());
}error[E0382]: borrow of moved value: `report`
--> src/main.rs:9:34
|
7 | let report = vec![1, 2, 3];
| ------ move occurs because `report` has type `Vec<i32>`, which does not implement the `Copy` trait
8 | tx.send(report).unwrap();
| ------ value moved here
9 | println!("sent {} rows", report.len());
| ^^^^^^ value borrowed here after moveSending a value and then modifying it would be a classic race, but in Rust it doesn't compile. Read what you need before sending, or send a clone.
4. Multiple producers
Clone the Sender to give each producer its own. Messages from different producers interleave in whatever order they're sent, so this example sorts before printing.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
for sensor in ["kitchen", "garage", "attic"] {
let tx = tx.clone();
thread::spawn(move || {
for reading in 0..2 {
tx.send(format!("{sensor}: reading {reading}")).unwrap();
}
});
}
drop(tx); // drop the original, or the loop below never ends
let mut all: Vec<String> = rx.iter().collect();
all.sort();
for line in &all {
println!("{line}");
}
}attic: reading 0
attic: reading 1
garage: reading 0
garage: reading 1
kitchen: reading 0
kitchen: reading 1Messages from a single producer always arrive in the order they were sent. Only the interleaving between producers varies.
5. The forgotten Sender
If any Sender is still alive, the receiver can't know that no more messages are coming, so it waits forever. The most common cause is the original tx in main, kept alive after cloning:
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 || tx.send(id).unwrap());
}
// Missing: drop(tx);
for id in rx {
println!("got {id}");
}
println!("never printed: main's own `tx` keeps the channel open");
}The fix is the drop(tx) from the previous example, or moving the original tx into the last thread instead of cloning it.
6. send fails when the receiver is gone
If the Receiver has been dropped, send returns Err(SendError(value)), giving you the value back. Producers use this to notice that nobody is listening anymore and stop.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel::<u32>();
let producer = thread::spawn(move || {
let mut sent = 0;
for n in 1.. {
if tx.send(n).is_err() {
return sent; // consumer went away: stop producing
}
sent += 1;
}
sent
});
let first_three: Vec<u32> = rx.iter().take(3).collect();
drop(rx); // hang up
let sent = producer.join().unwrap();
println!("consumer took {first_three:?}");
println!("producer stopped after sending at least 3: {}", sent >= 3);
}consumer took [1, 2, 3]
producer stopped after sending at least 3: true7. recv fails when all senders are gone
recv() returns Err(RecvError) once the channel is empty and closed. That's what ends the for loop; you can also handle it directly.
use std::sync::mpsc;
fn main() {
let (tx, rx) = mpsc::channel();
tx.send("last message").unwrap();
drop(tx);
println!("{:?}", rx.recv()); // buffered messages are still delivered
println!("{:?}", rx.recv()); // then: closed
}Ok("last message")
Err(RecvError)8. try_recv: check without blocking
try_recv() returns immediately: Ok(value), Err(TryRecvError::Empty) if nothing is waiting, or Err(TryRecvError::Disconnected) if the channel is closed. It's the equivalent of queue.poll().
use std::sync::mpsc::{self, TryRecvError};
fn main() {
let (tx, rx) = mpsc::channel();
println!("{:?}", rx.try_recv());
tx.send(42).unwrap();
println!("{:?}", rx.try_recv());
drop(tx);
match rx.try_recv() {
Ok(v) => println!("value {v}"),
Err(TryRecvError::Empty) => println!("nothing yet"),
Err(TryRecvError::Disconnected) => println!("channel closed"),
}
}Err(Empty)
Ok(42)
channel closedtry_iter() drains everything currently waiting without blocking, which is handy in a loop that also does other work (like a game loop or UI thread).
9. recv_timeout: wait, but not forever
recv_timeout(duration) waits up to a limit and returns Err(RecvTimeoutError::Timeout) if nothing arrives. A typical use is a heartbeat monitor.
use std::sync::mpsc::{self, RecvTimeoutError};
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
for beat in 1..=3 {
thread::sleep(Duration::from_millis(20));
tx.send(beat).unwrap();
}
thread::sleep(Duration::from_millis(300)); // the service hangs...
drop(tx);
});
loop {
match rx.recv_timeout(Duration::from_millis(150)) {
Ok(beat) => println!("heartbeat {beat}"),
Err(RecvTimeoutError::Timeout) => {
println!("no heartbeat for 150 ms: service looks dead");
break;
}
Err(RecvTimeoutError::Disconnected) => {
println!("service shut down");
break;
}
}
}
}heartbeat 1
heartbeat 2
heartbeat 3
no heartbeat for 150 ms: service looks dead10. Messages as an enum
Real channels carry more than one kind of message. An enum makes the protocol explicit, and match makes sure the receiver handles every kind.
use std::sync::mpsc;
use std::thread;
enum Event {
Login { user: String },
Purchase { user: String, cents: u64 },
Logout { user: String },
}
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
tx.send(Event::Login { user: String::from("ana") }).unwrap();
tx.send(Event::Purchase { user: String::from("ana"), cents: 2599 }).unwrap();
tx.send(Event::Purchase { user: String::from("ana"), cents: 1000 }).unwrap();
tx.send(Event::Logout { user: String::from("ana") }).unwrap();
});
let mut revenue = 0;
for event in rx {
match event {
Event::Login { user } => println!("{user} logged in"),
Event::Purchase { user, cents } => {
revenue += cents;
println!("{user} spent {cents}");
}
Event::Logout { user } => println!("{user} logged out"),
}
}
println!("revenue: {revenue}");
}ana logged in
ana spent 2599
ana spent 1000
ana logged out
revenue: 359911. Bounded channels and backpressure
mpsc::channel() is unbounded: a fast producer can fill memory. mpsc::sync_channel(n) holds at most n messages; send blocks when it's full, slowing the producer down to the consumer's pace. That's backpressure, like ArrayBlockingQueue.put.
use std::sync::mpsc;
use std::thread;
use std::time::{Duration, Instant};
fn main() {
let (tx, rx) = mpsc::sync_channel(2); // room for 2 messages
let start = Instant::now();
let producer = thread::spawn(move || {
for i in 1..=6 {
tx.send(i).unwrap(); // blocks while the channel is full
}
start.elapsed()
});
let mut got = Vec::new();
for item in rx {
thread::sleep(Duration::from_millis(20)); // slow consumer
got.push(item);
}
let producer_time = producer.join().unwrap();
println!("received {got:?}");
println!("producer was slowed down to the consumer's pace: {}", producer_time >= Duration::from_millis(40));
}received [1, 2, 3, 4, 5, 6]
producer was slowed down to the consumer's pace: true12. sync_channel(0): a rendezvous
With capacity zero, send blocks until the receiver actually takes the value. The two threads meet at that point. It's a way to hand off work and know that it was picked up.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::sync_channel::<&str>(0);
let (log_tx, log_rx) = mpsc::channel();
let producer = thread::spawn(move || {
for task in ["task-a", "task-b"] {
tx.send(task).unwrap(); // returns only once the consumer has received it
log_tx.send(format!("handed off {task}")).unwrap();
}
});
for task in rx {
println!("consumer took {task}");
// The producer's log line for this task can only exist after this point.
}
producer.join().unwrap();
println!("{:?}", log_rx.iter().collect::<Vec<_>>());
}consumer took task-a
consumer took task-b
["handed off task-a", "handed off task-b"]13. Request and reply
To get an answer back, send a reply channel inside the request. The requester waits on its own receiver. This is how you build a CompletableFuture-style call on top of channels.
use std::sync::mpsc;
use std::thread;
struct PriceRequest {
sku: String,
reply: mpsc::Sender<Option<u32>>,
}
fn main() {
let (tx, rx) = mpsc::channel::<PriceRequest>();
// A "pricing service" thread.
thread::spawn(move || {
for req in rx {
let price = match req.sku.as_str() {
"BOOK" => Some(1500),
"PEN" => Some(200),
_ => None,
};
let _ = req.reply.send(price); // the requester may have given up; ignore
}
});
for sku in ["PEN", "LAPTOP", "BOOK"] {
let (reply_tx, reply_rx) = mpsc::channel();
tx.send(PriceRequest { sku: sku.to_string(), reply: reply_tx }).unwrap();
println!("{sku}: {:?}", reply_rx.recv().unwrap());
}
}PEN: Some(200)
LAPTOP: None
BOOK: Some(1500)14. Keeping results in order
When workers finish in random order but you need results in input order, send the index along with the result and put each result in its slot.
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let urls = ["/a", "/bb", "/ccc", "/dddd"];
let (tx, rx) = mpsc::channel();
for (i, url) in urls.into_iter().enumerate() {
let tx = tx.clone();
thread::spawn(move || {
// Later URLs finish first, to show the reordering.
thread::sleep(Duration::from_millis(40 - 10 * i as u64));
tx.send((i, format!("{url} -> {} bytes", url.len() * 100))).unwrap();
});
}
drop(tx);
let mut results = vec![String::new(); urls.len()];
for (i, res) in rx {
results[i] = res;
}
println!("{results:#?}");
}[
"/a -> 200 bytes",
"/bb -> 300 bytes",
"/ccc -> 400 bytes",
"/dddd -> 500 bytes",
]15. Channels with scoped threads
Channels work with thread::scope too. Scoped workers can borrow input data and send owned results back.
use std::sync::mpsc;
use std::thread;
fn main() {
let words = vec!["tokio", "serde", "rayon", "clap", "regex", "anyhow"];
let (tx, rx) = mpsc::channel();
thread::scope(|s| {
for chunk in words.chunks(2) {
let tx = tx.clone();
s.spawn(move || {
for w in chunk {
tx.send((w.len(), w.to_uppercase())).unwrap();
}
});
}
});
drop(tx);
let mut out: Vec<(usize, String)> = rx.into_iter().collect();
out.sort();
println!("{out:?}");
}[(4, "CLAP"), (5, "RAYON"), (5, "REGEX"), (5, "SERDE"), (5, "TOKIO"), (6, "ANYHOW")]Note that thread::scope returns only after all workers finish, so every message is already in the channel when we read it.
16. A pipeline of stages
Channels connect stages that run at the same time: while stage 2 processes item 1, stage 1 is already parsing item 2. Each stage owns its receiver and the sender for the next stage; when a stage finishes, dropping its sender closes the next channel, and shutdown ripples down the pipeline.
use std::sync::mpsc;
use std::thread;
fn main() {
let lines = vec!["3,apple", "x,broken", "5,pear", "12,melon"];
let (parsed_tx, parsed_rx) = mpsc::channel();
let (priced_tx, priced_rx) = mpsc::channel();
// Stage 1: parse
let parser = thread::spawn(move || {
for line in lines {
if let Some((qty, fruit)) = line.split_once(',') {
if let Ok(qty) = qty.parse::<u32>() {
parsed_tx.send((qty, fruit.to_string())).unwrap();
continue;
}
}
eprintln!("skipping bad line {line:?}");
}
});
// Stage 2: price
let pricer = thread::spawn(move || {
for (qty, fruit) in parsed_rx {
let unit = if fruit == "melon" { 300 } else { 50 };
priced_tx.send(format!("{qty} x {fruit} = {}", qty * unit)).unwrap();
}
});
// Stage 3 (main): print
for line in priced_rx {
println!("{line}");
}
parser.join().unwrap();
pricer.join().unwrap();
}3 x apple = 150
5 x pear = 250
12 x melon = 360017. Several consumers: share the receiver
mpsc has a single consumer, but a group of workers can share one Receiver behind a Mutex. Each worker locks, takes one job, unlocks, and processes it. This is exactly how a fixed thread pool pulls from its work queue.
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
fn main() {
let (job_tx, job_rx) = mpsc::channel::<u32>();
let job_rx = Arc::new(Mutex::new(job_rx));
let (result_tx, result_rx) = mpsc::channel();
for worker in 0..3 {
let job_rx = Arc::clone(&job_rx);
let result_tx = result_tx.clone();
thread::spawn(move || loop {
let job = job_rx.lock().unwrap().recv(); // lock held only while receiving
match job {
Ok(n) => result_tx.send((n, n * n, worker)).unwrap(),
Err(_) => break, // channel closed: no more jobs
}
});
}
drop(result_tx);
for n in 1..=8 {
job_tx.send(n).unwrap();
}
drop(job_tx);
let mut results: Vec<(u32, u32)> = result_rx.iter().map(|(n, sq, _worker)| (n, sq)).collect();
results.sort();
println!("{results:?}");
}[(1, 1), (2, 4), (3, 9), (4, 16), (5, 25), (6, 36), (7, 49), (8, 64)]18. A thread pool with graceful shutdown
Putting it together: a small ThreadPool. Jobs are boxed closures (Box<dyn FnOnce() + Send>). When the pool is dropped, it closes the job channel and joins every worker, so all submitted jobs finish first.
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
type Job = Box<dyn FnOnce() + Send + 'static>;
struct ThreadPool {
sender: Option<mpsc::Sender<Job>>,
workers: Vec<thread::JoinHandle<()>>,
}
impl ThreadPool {
fn new(size: usize) -> ThreadPool {
let (sender, receiver) = mpsc::channel::<Job>();
let receiver = Arc::new(Mutex::new(receiver));
let workers = (0..size)
.map(|_| {
let receiver = Arc::clone(&receiver);
thread::spawn(move || loop {
let job = receiver.lock().unwrap().recv();
match job {
Ok(job) => job(),
Err(_) => break,
}
})
})
.collect();
ThreadPool { sender: Some(sender), workers }
}
fn execute(&self, f: impl FnOnce() + Send + 'static) {
self.sender.as_ref().unwrap().send(Box::new(f)).unwrap();
}
}
impl Drop for ThreadPool {
fn drop(&mut self) {
drop(self.sender.take()); // close the channel: workers exit after draining it
for w in self.workers.drain(..) {
w.join().unwrap();
}
}
}
fn main() {
let (tx, rx) = mpsc::channel();
{
let pool = ThreadPool::new(4);
for i in 1..=10u64 {
let tx = tx.clone();
pool.execute(move || tx.send((i, (1..=i).product::<u64>())).unwrap());
}
} // pool dropped here: waits for all 10 jobs
drop(tx);
let mut factorials: Vec<(u64, u64)> = rx.iter().collect();
factorials.sort();
println!("{:?}", &factorials[..5]);
println!("10! = {}", factorials[9].1);
}[(1, 1), (2, 2), (3, 6), (4, 24), (5, 120)]
10! = 3628800Drop shuts the pool down and waits for the workers, so it can't be forgotten.
19. Stopping a worker with a message
To stop a long-running worker, send it a message it checks between units of work. try_recv lets it check without blocking.
use std::sync::mpsc::{self, TryRecvError};
use std::thread;
use std::time::Duration;
fn main() {
let (stop_tx, stop_rx) = mpsc::channel::<()>();
let worker = thread::spawn(move || {
let mut batches = 0;
loop {
match stop_rx.try_recv() {
Ok(()) | Err(TryRecvError::Disconnected) => break,
Err(TryRecvError::Empty) => {}
}
batches += 1; // process one batch
thread::sleep(Duration::from_millis(5));
}
batches
});
thread::sleep(Duration::from_millis(50));
stop_tx.send(()).unwrap();
let batches = worker.join().unwrap();
println!("worker stopped cleanly after some batches: {}", batches > 0);
}worker stopped cleanly after some batches: trueTreating Disconnected as "stop" too means that simply dropping stop_tx also shuts the worker down. An AtomicBool flag is the other common way (Atomics).
20. Merging several sources into one channel
std has no select! for waiting on several channels at once. The usual workaround is to send everything into one channel, wrapping each source's messages in an enum variant.
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
enum Input {
Key(char),
Tick(u32),
NetworkMessage(String),
}
fn main() {
let (tx, rx) = mpsc::channel();
let keys = tx.clone();
thread::spawn(move || {
for c in ['h', 'i'] {
keys.send(Input::Key(c)).unwrap();
}
});
let ticks = tx.clone();
thread::spawn(move || {
for t in 1..=3 {
ticks.send(Input::Tick(t)).unwrap();
thread::sleep(Duration::from_millis(5));
}
});
thread::spawn(move || tx.send(Input::NetworkMessage(String::from("new mail"))).unwrap());
let (mut typed, mut ticks_seen, mut messages) = (String::new(), 0, Vec::new());
for input in rx {
match input {
Input::Key(c) => typed.push(c),
Input::Tick(_) => ticks_seen += 1,
Input::NetworkMessage(m) => messages.push(m),
}
}
println!("typed {typed:?}, {ticks_seen} ticks, messages {messages:?}");
}typed "hi", 3 ticks, messages ["new mail"]21. Errors travel as values
A worker that can fail sends Results. The receiver decides what to do with each failure instead of the worker panicking.
use std::sync::mpsc;
use std::thread;
fn parse_record(line: &str) -> Result<(String, u32), String> {
let (name, age) = line.split_once(':').ok_or(format!("no ':' in {line:?}"))?;
let age = age.trim().parse().map_err(|_| format!("bad age in {line:?}"))?;
Ok((name.trim().to_string(), age))
}
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
for line in ["ana: 34", "ben 40", "cy: forty", "dee: 29"] {
tx.send(parse_record(line)).unwrap();
}
});
let (ok, failed): (Vec<_>, Vec<_>) = rx.iter().partition(|r| r.is_ok());
let people: Vec<(String, u32)> = ok.into_iter().map(Result::unwrap).collect();
println!("parsed: {people:?}");
for e in failed {
println!("error: {}", e.unwrap_err());
}
}parsed: [("ana", 34), ("dee", 29)]
error: no ':' in "ben 40"
error: bad age in "cy: forty"22. Batching messages
A consumer that writes to a database or network is more efficient with batches. Collect messages until the batch is full or a short timeout passes, then flush.
use std::sync::mpsc::{self, RecvTimeoutError};
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
for i in 1..=7 {
tx.send(format!("event-{i}")).unwrap();
}
thread::sleep(Duration::from_millis(100)); // quiet period
tx.send(String::from("event-8")).unwrap();
});
let mut batch = Vec::new();
loop {
match rx.recv_timeout(Duration::from_millis(30)) {
Ok(event) => {
batch.push(event);
if batch.len() == 3 {
println!("flush (full): {batch:?}");
batch.clear();
}
}
Err(RecvTimeoutError::Timeout) => {
if !batch.is_empty() {
println!("flush (timeout): {batch:?}");
batch.clear();
}
}
Err(RecvTimeoutError::Disconnected) => {
if !batch.is_empty() {
println!("flush (closing): {batch:?}");
}
break;
}
}
}
}flush (full): ["event-1", "event-2", "event-3"]
flush (full): ["event-4", "event-5", "event-6"]
flush (timeout): ["event-7"]
flush (closing): ["event-8"]23. The actor pattern
An actor is a thread that owns some state and changes it only in response to messages. Nobody else can touch the state, so no locks are needed. A small handle type with ordinary methods hides the channels from callers.
use std::collections::HashMap;
use std::sync::mpsc;
use std::thread;
enum Command {
Deposit { account: String, amount: i64 },
Withdraw { account: String, amount: i64, reply: mpsc::Sender<Result<i64, String>> },
Balance { account: String, reply: mpsc::Sender<i64> },
}
#[derive(Clone)]
struct BankHandle {
tx: mpsc::Sender<Command>,
}
impl BankHandle {
fn start() -> BankHandle {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let mut balances: HashMap<String, i64> = HashMap::new(); // owned by this thread only
for cmd in rx {
match cmd {
Command::Deposit { account, amount } => *balances.entry(account).or_default() += amount,
Command::Withdraw { account, amount, reply } => {
let bal = balances.entry(account).or_default();
let result = if *bal >= amount {
*bal -= amount;
Ok(*bal)
} else {
Err(format!("insufficient funds: balance {bal}"))
};
let _ = reply.send(result);
}
Command::Balance { account, reply } => {
let _ = reply.send(balances.get(&account).copied().unwrap_or(0));
}
}
}
});
BankHandle { tx }
}
fn deposit(&self, account: &str, amount: i64) {
self.tx.send(Command::Deposit { account: account.to_string(), amount }).unwrap();
}
fn withdraw(&self, account: &str, amount: i64) -> Result<i64, String> {
let (reply, rx) = mpsc::channel();
self.tx.send(Command::Withdraw { account: account.to_string(), amount, reply }).unwrap();
rx.recv().unwrap()
}
fn balance(&self, account: &str) -> i64 {
let (reply, rx) = mpsc::channel();
self.tx.send(Command::Balance { account: account.to_string(), reply }).unwrap();
rx.recv().unwrap()
}
}
fn main() {
let bank = BankHandle::start();
let tellers: Vec<_> = (0..4)
.map(|_| {
let bank = bank.clone();
thread::spawn(move || {
for _ in 0..25 {
bank.deposit("ana", 10);
}
})
})
.collect();
for t in tellers {
t.join().unwrap();
}
println!("balance: {}", bank.balance("ana"));
println!("withdraw 300: {:?}", bank.withdraw("ana", 300));
println!("withdraw 900: {:?}", bank.withdraw("ana", 900));
}balance: 1000
withdraw 300: Ok(700)
withdraw 900: Err("insufficient funds: balance 700")Messages from one sender are processed in order, so after all four tellers are joined, every deposit is in the actor's queue before the balance request.
24. A dedicated logging thread
Writing logs from many threads to one output needs coordination. Sending log lines to one logger thread keeps request threads fast and the output unmixed.
use std::sync::mpsc;
use std::thread;
enum Log {
Line { level: &'static str, source: String, text: String },
Flush(mpsc::Sender<usize>),
}
fn main() {
let (log_tx, log_rx) = mpsc::channel();
let logger = thread::spawn(move || {
let mut buffer = Vec::new();
let mut written = 0;
for msg in log_rx {
match msg {
Log::Line { level, source, text } => buffer.push(format!("[{level}] {source}: {text}")),
Log::Flush(done) => {
buffer.sort(); // make the demo's output stable
for line in buffer.drain(..) {
println!("{line}");
written += 1;
}
let _ = done.send(written);
}
}
}
});
let workers: Vec<_> = ["auth", "orders", "payments"]
.into_iter()
.map(|source| {
let log = log_tx.clone();
thread::spawn(move || {
log.send(Log::Line { level: "INFO", source: source.to_string(), text: String::from("started") }).unwrap();
if source == "payments" {
log.send(Log::Line { level: "WARN", source: source.to_string(), text: String::from("gateway slow") }).unwrap();
}
})
})
.collect();
for w in workers {
w.join().unwrap();
}
let (done_tx, done_rx) = mpsc::channel();
log_tx.send(Log::Flush(done_tx)).unwrap();
println!("-- logger wrote {} lines --", done_rx.recv().unwrap());
drop(log_tx);
logger.join().unwrap();
}[INFO] auth: started
[INFO] orders: started
[INFO] payments: started
[WARN] payments: gateway slow
-- logger wrote 4 lines --25. Realistic: a file-checksum pipeline with a worker pool
A three-stage pipeline: one thread "reads files", a pool of hasher threads computes checksums, and main collects the results. The bounded channel between reader and hashers limits how many file contents are held in memory at once.
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
fn fnv1a(data: &[u8]) -> u64 {
data.iter().fold(0xcbf29ce484222325, |h, &b| (h ^ b as u64).wrapping_mul(0x100000001b3))
}
fn main() {
let files: Vec<(String, Vec<u8>)> = (1..=6)
.map(|i| (format!("doc{i}.txt"), format!("contents of document number {i}").into_bytes()))
.collect();
let (content_tx, content_rx) = mpsc::sync_channel::<(String, Vec<u8>)>(2); // at most 2 files buffered
let content_rx = Arc::new(Mutex::new(content_rx));
let (hash_tx, hash_rx) = mpsc::channel();
let reader = thread::spawn(move || {
for file in files {
content_tx.send(file).unwrap();
}
});
let hashers: Vec<_> = (0..3)
.map(|_| {
let rx = Arc::clone(&content_rx);
let tx = hash_tx.clone();
thread::spawn(move || loop {
let next = rx.lock().unwrap().recv();
let Ok((name, bytes)) = next else { break };
tx.send((name, bytes.len(), fnv1a(&bytes))).unwrap();
})
})
.collect();
drop(hash_tx);
let mut results: Vec<(String, usize, u64)> = hash_rx.iter().collect();
results.sort();
reader.join().unwrap();
for h in hashers {
h.join().unwrap();
}
for (name, len, hash) in results {
println!("{name:<9} {len:>3} bytes {hash:016x}");
}
}doc1.txt 29 bytes 5bc7a664749e6651
doc2.txt 29 bytes 5bc7a364749e6138
doc3.txt 29 bytes 5bc7a464749e62eb
doc4.txt 29 bytes 5bc7a964749e6b6a
doc5.txt 29 bytes 5bc7aa64749e6d1d
doc6.txt 29 bytes 5bc7a764749e680426. Realistic: progress reporting from workers
Workers report progress on a channel; a single "UI" thread (main) aggregates it and prints a summary, so worker threads never print directly.
use std::collections::BTreeMap;
use std::sync::mpsc;
use std::thread;
enum Progress {
Started { job: &'static str, total: u32 },
Advanced { job: &'static str, done: u32 },
Finished { job: &'static str },
}
fn main() {
let (tx, rx) = mpsc::channel();
for (job, total) in [("resize-images", 40), ("send-emails", 25), ("backup-db", 10)] {
let tx = tx.clone();
thread::spawn(move || {
tx.send(Progress::Started { job, total }).unwrap();
for done in (10..=total).step_by(10) {
tx.send(Progress::Advanced { job, done }).unwrap();
}
tx.send(Progress::Finished { job }).unwrap();
});
}
drop(tx);
let mut state: BTreeMap<&str, (u32, u32, bool)> = BTreeMap::new(); // job -> (done, total, finished)
let mut updates = 0;
for msg in rx {
updates += 1;
match msg {
Progress::Started { job, total } => {
state.insert(job, (0, total, false));
}
Progress::Advanced { job, done } => state.get_mut(job).unwrap().0 = done,
Progress::Finished { job } => state.get_mut(job).unwrap().2 = true,
}
}
println!("{updates} progress messages");
for (job, (done, total, finished)) in state {
println!("{job:<14} {done:>2}/{total:<2} {}", if finished { "done" } else { "running" });
}
}13 progress messages
backup-db 10/10 done
resize-images 40/40 done
send-emails 20/25 doneEach job's own messages arrive in order (Started before Advanced), which is why state.get_mut(job).unwrap() is safe here.
27. Realistic: a key-value store actor with snapshots
A slightly bigger actor: a key-value store that supports get, set, and snapshot. Many client threads use it concurrently; the actor processes one command at a time, so each command is atomic.
use std::collections::BTreeMap;
use std::sync::mpsc::{self, Sender};
use std::thread;
enum Cmd {
Set(String, i64),
Incr(String, Sender<i64>),
Get(String, Sender<Option<i64>>),
Snapshot(Sender<BTreeMap<String, i64>>),
}
fn start_store() -> (Sender<Cmd>, thread::JoinHandle<usize>) {
let (tx, rx) = mpsc::channel();
let handle = thread::spawn(move || {
let mut data = BTreeMap::new();
let mut handled = 0;
for cmd in rx {
handled += 1;
match cmd {
Cmd::Set(k, v) => {
data.insert(k, v);
}
Cmd::Incr(k, reply) => {
let v = data.entry(k).or_insert(0);
*v += 1;
let _ = reply.send(*v);
}
Cmd::Get(k, reply) => {
let _ = reply.send(data.get(&k).copied());
}
Cmd::Snapshot(reply) => {
let _ = reply.send(data.clone());
}
}
}
handled // returned when every Sender is gone
});
(tx, handle)
}
fn main() {
let (store, actor) = start_store();
store.send(Cmd::Set(String::from("visits"), 100)).unwrap();
let clients: Vec<_> = (0..5)
.map(|_| {
let store = store.clone();
thread::spawn(move || {
for _ in 0..20 {
let (reply, rx) = mpsc::channel();
store.send(Cmd::Incr(String::from("visits"), reply)).unwrap();
rx.recv().unwrap();
}
})
})
.collect();
for c in clients {
c.join().unwrap();
}
let (reply, rx) = mpsc::channel();
store.send(Cmd::Get(String::from("visits"), reply)).unwrap();
println!("visits: {:?}", rx.recv().unwrap());
let (reply, rx) = mpsc::channel();
store.send(Cmd::Snapshot(reply)).unwrap();
println!("snapshot: {:?}", rx.recv().unwrap());
drop(store); // last Sender: the actor's loop ends
println!("actor handled {} commands", actor.join().unwrap());
}visits: Some(200)
snapshot: {"visits": 200}
actor handled 103 commands28. Choosing between channels and locks
Both styles are correct in Rust; the compiler checks both. Some rules of thumb:
| Use channels when | Use shared state when |
|---|---|
| work flows through stages (pipelines, producers/consumers) | many threads read and occasionally update the same data |
| one thread should own a resource (a connection, a file, a UI) | the data is naturally a shared table (cache, registry) |
| you want backpressure between fast and slow parts | you need low overhead for tiny operations (counters: atomics) |
| you want requests handled one at a time (actors) | you need to read a consistent snapshot of several fields |
Many programs mix them: an Arc<RwLock<Config>> read by workers that pull jobs from a channel.
Summary
mpsc::channel()gives a cloneableSender<T>and a singleReceiver<T>.sendmoves the value to the receiving thread.- A
Receiverloop ends when everySenderis dropped. Forgetting to drop the originaltxmakes it hang. sendfails if the receiver is gone;recvfails when the channel is empty and closed.try_recvandrecv_timeoutavoid blocking forever.sync_channel(n)is bounded and gives backpressure;sync_channel(0)is a rendezvous.- Send an enum for several message kinds, a reply
Senderfor request/response, and an index to restore order. - Share a
ReceiverbehindArc<Mutex<_>>for a worker pool; a thread pool'sDropcan close the channel and join workers for graceful shutdown. - Actors own their state and handle one message at a time, with no locks.
Next: Atomics, Send & Sync, which looks under the hood at lock-free primitives and the traits that make all of this compile-time safe.