Channels between threads

A channel is a queue with two ends that may live on different threads: a Sender pushes and a Receiver pops. The standard library provides the multi-producer, single-consumer kind in std::sync::mpsc, and the ownership of the two ends is what makes the handover safe without a lock.

This lesson sends one value, then values from several workers, then watches the channel close when the last sender goes. The last two sections are the bounded form, which slows a fast producer down, and the compiler error for a sender that was moved into a thread and used again.

A channel and one value

mpsc::channel returns the pair, and each end is moved to where it is used: the sender into the worker, the receiver stays with the caller. recv blocks until a value arrives or every sender is gone.

The program starts a worker, sends forty-two and prints what the receiver got. The type of the value is inferred from the send.

use std::sync::mpsc; use std::thread; fn main() { let (sender, receiver) = mpsc::channel(); thread::spawn(move || { sender.send(42).unwrap(); }); println!("{}", receiver.recv().unwrap()); }

rustc --edition 2024 one.rs
./one

42

The channel is not shared, it is split: each end has one owner, and the ownership rules move them across the thread boundary. That is the difference from a shared buffer behind a lock, which the next lesson builds.

recv_timeout is the form for a consumer that must not wait forever: it returns Err(RecvTimeoutError::Timeout) when nothing arrives, and Disconnected when every sender is gone. A protocol with a heartbeat uses it to notice a silent peer.

Several senders

Sender is cloneable and Receiver is not, which is the multi-producer part of the name. Each worker gets its own clone and its own value, and the original sender is dropped so the receiver knows when the last worker is done.

Three workers send a multiple of ten. The program collects everything the receiver yields, sorts the vector and prints it, so the output does not depend on the order the threads ran in.

use std::sync::mpsc; use std::thread; fn main() { let (sender, receiver) = mpsc::channel(); for index in 0..3 { let sender = sender.clone(); thread::spawn(move || { sender.send(index * 10).unwrap(); }); } drop(sender); let mut values: Vec<i32> = receiver.iter().collect(); values.sort(); println!("{values:?}"); }

rustc --edition 2024 clone.rs
./clone

[0, 10, 20]

drop(sender) is the line that lets the loop over the receiver end: while any sender exists, iter waits for another value. Forgetting it is a hang, not an error, which is why the value is dropped explicitly here instead of being left to the end of main.

The channel closes

When the last sender is dropped the channel closes, and recv returns Err(RecvError) instead of blocking. That is the signal a consumer loop uses to stop, and it is the reason the sender of a worker is owned by the worker's closure.

The worker sends three squares and ends, which drops the sender. The program reads four times: three values and the closing error.

use std::sync::mpsc; use std::thread; fn main() { let (sender, receiver) = mpsc::channel(); let worker = thread::spawn(move || { for value in 1..=3 { sender.send(value * value).unwrap(); } }); worker.join().unwrap(); for _ in 0..4 { println!("{:?}", receiver.recv()); } }

rustc --edition 2024 closed.rs
./closed

Ok(1)
Ok(4)
Ok(9)
Err(RecvError)

The join before the reads is what makes the output ordered: without it the receiver would print the values as they arrive, which is still three values and the error but interleaved with the worker's work. A program that must stop on the close writes while let Ok(value) = receiver.recv(), which ends on the same error without printing it.

A bounded channel

sync_channel takes a capacity and blocks the sender when the queue is full, which is back-pressure: a fast producer cannot run away from a slow consumer. A capacity of zero is a rendezvous, where every send waits for a matching receive.

The program uses a capacity of one, reads the first value and then drains the rest. The worker waits while the queue is full, so the last two values arrive only after the receiver has made room.

use std::sync::mpsc; use std::thread; fn main() { let (sender, receiver) = mpsc::sync_channel(1); let worker = thread::spawn(move || { for value in [1, 2, 3] { sender.send(value).unwrap(); } }); let first = receiver.recv().unwrap(); let rest: Vec<i32> = receiver.iter().collect(); worker.join().unwrap(); println!("{first} {rest:?}"); }

rustc --edition 2024 bounded.rs
./bounded

1 [2, 3]

The first value is read before the drain, which is the shape of a consumer that does something with the head of a stream and then processes the tail. The lesson on concurrency patterns builds the same queue with a lock and a condition variable.

A sender used after the move

A closure that takes the sender by move owns it, so the caller cannot send afterwards. The compiler reports E0382 and the suggestion is the clone that the previous section used on purpose.

The program moves the sender into a worker and then tries to send from main. The compiler stops at the second send.

use std::sync::mpsc; use std::thread; fn main() { let (sender, receiver) = mpsc::channel(); thread::spawn(move || sender.send(1).unwrap()); sender.send(2).unwrap(); println!("{:?}", receiver.recv()); }

rustc --edition 2024 moved_err.rs

error[E0382]: borrow of moved value: `sender`
--> moved_err.rs:7:5
|
5 | let (sender, receiver) = mpsc::channel();
| ------ move occurs because `sender` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait
6 | thread::spawn(move || sender.send(1).unwrap());
| ------- ------ variable moved due to use in closure
| |
| value moved into closure here
7 | sender.send(2).unwrap();
| ^^^^^^ value borrowed here after move
|
help: consider cloning the value before moving it into the closure
|
6 ~ let value = sender.clone();
7 ~ thread::spawn(move || value.send(1).unwrap());
|

error: aborting due to 1 previous error

For more information about this error, try `rustc --explain E0382`.

The help spells out the fix in two lines, and it is the same clone the multi-sender section used. A send to a closed channel is not an error at compile time: it returns Err with the message sending on a closed channel, which is what a program prints when the receiver went away first.

Exercises

  1. Send a String through a channel and print its length on the other side.
  2. Start three workers that each send their index, collect the values and print them sorted.
  3. Change the bounded channel to capacity zero and watch the send wait for the receive.

Questions

A new set of questions in a new order every time. Choose one answer for each, then check.

  1. What does mpsc::channel return?

    Answer

    A Sender and a Receiver

    The pair is the two ends of the queue, and each is moved where it is used.

  2. Which end of a channel can be cloned?

    Answer

    The Sender

    Many producers and one consumer are what the mpsc name describes.

  3. What does recv return when every sender is gone?

    Answer

    Err(RecvError)

    The closed channel reports the error instead of waiting for a value that cannot come.

  4. What does a capacity of zero in sync_channel mean?

    Answer

    Every send waits for a matching receive

    A zero capacity is a rendezvous between the two threads.

  5. What does this program print? use std::sync::mpsc; fn main() { let (sender, receiver) = mpsc::channel(); sender.send(5).unwrap(); println!("{}", receiver.recv().unwrap()); }

    Answer

    5

    A channel works inside one thread as well, and the value is read back.

  6. What does this program print? use std::sync::mpsc; fn main() { let (sender, receiver) = mpsc::channel(); sender.send("a").unwrap(); sender.send("b").unwrap(); drop(sender); let values: Vec<&str> = receiver.iter().collect(); println!("{values:?}"); }

    Answer

    ["a", "b"]

    One queue keeps the order of the sends, and the drop closes it for the collect.

  7. What does this program print? use std::sync::mpsc; fn main() { let (sender, receiver) = mpsc::channel::<i32>(); drop(receiver); match sender.send(1) { Ok(()) => println!("sent"), Err(error) => println!("{error}"), } }

    Answer

    sending on a closed channel

    The receiver is gone, so the send returns the error whose message is printed.

  8. What does this program print? use std::sync::mpsc; fn main() { let (sender, receiver) = mpsc::channel(); for value in 1..=2 { sender.send(value).unwrap(); } drop(sender); println!("{:?}", receiver.recv()); println!("{:?}", receiver.recv()); println!("{:?}", receiver.recv()); }

    Answer

    Ok(1) / Ok(2) / Err(RecvError)

    Two values then the closing error, in the order of the sends.

  9. What is the first line of the compiler output for this program? use std::sync::mpsc; use std::thread; fn main() { let (sender, receiver) = mpsc::channel(); thread::spawn(move || sender.send(1).unwrap()); sender.send(2).unwrap(); println!("{:?}", receiver.recv()); }

    Answer

    error[E0382]: borrow of moved value: sender

    The closure took the sender by move, and the later call borrows the moved binding.

  10. What is the first line of the compiler output for this program? use std::sync::mpsc; fn main() { let (sender, receiver) = mpsc::channel(); receiver.clone(); }

    Answer

    error[E0599]: no method named clone found for struct std::sync::mpsc::Receiver<T> in the current scope

    The receiver is deliberately not cloneable, so the method does not exist.

The code on this page was built with rustc 1.96.1 and run on Alpine Linux 3.24.3 (x86_64) in a Docker container.

Article author: Arthur Isaev