Rust Channels: Message Passing Between Threads

On the Concurrency page, threads shared data by locking it with Arc<Mutex<T>>. That works, but there’s another way that is often simpler: instead of sharing data, threads send messages to each other.

Think of a channel like a pipe with two ends. One thread drops values into the sender end. Another thread takes them out of the receiver end, in the same order. Nobody shares anything: each value moves from one thread to the other. The Go community has a saying that Rust borrows here: “Do not communicate by sharing memory; instead, share memory by communicating.”

Your first channel

The standard library’s channel lives in std::sync::mpsc. The name means multiple producer, single consumer: many senders, one receiver.

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

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("hello from the worker");
        tx.send(message).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("main got: {received}");
}

Output:

main got: hello from the worker

▶ Try it Yourself

Step by step:

• mpsc::channel() returns a pair: a Sender and a Receiver. By tradition they’re called tx (transmit) and rx (receive).
• move hands tx to the new thread, the same way you moved data into threads on the Concurrency page.
• tx.send(value) puts a value into the channel. It returns a Result, because it fails if the receiver is already gone.
• rx.recv() waits until a value arrives. It returns Err only when every sender is gone and the channel is empty.

Notice we didn’t need join(). recv() already waits for the message, so main can’t finish too early.

Receiving many values

A Receiver works as an iterator. A for loop over it keeps calling recv() for you, and the loop ends by itself when all senders have been dropped:

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

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i * 10).unwrap();
        }
        // tx is dropped here, when the thread ends
    });

    for value in rx {
        println!("got {value}");
    }
    println!("channel closed, all done");
}

Output:

got 10
got 20
got 30
channel closed, all done

▶ Try it Yourself

There’s only one sender here, so the values arrive in exactly the order they were sent. This “the loop stops when the senders are gone” rule is important. Keep it in mind for the next section.

Multiple producers: clone the sender

To let several threads send, give each one its own copy of the sender with tx.clone(). All copies feed the same channel:

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

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(format!("worker {id} reporting")).unwrap();
        });
    }
    drop(tx); // drop the original, or the loop below never ends

    let mut messages: Vec<String> = rx.iter().collect();
    messages.sort();
    for m in messages {
        println!("{m}");
    }
}

Output:

worker 1 reporting
worker 2 reporting
worker 3 reporting

▶ Try it Yourself

Two things to notice:

• drop(tx) matters. Each thread drops its clone when it finishes, but main still holds the original. As long as one sender exists, the receiver keeps waiting, and the program would hang forever. Dropping it says “main won’t send anything”.
• The threads finish in any order. Messages can arrive as worker 2, worker 1, worker 3. We collect them and sort them so the printed output is always the same. Remove the sort() and run it a few times to see the order change.

Not waiting forever: try_recv and recv_timeout

recv() blocks until something arrives. Sometimes you don’t want to wait. Two cousins help:

• try_recv() returns immediately: a value, Empty (nothing yet), or Disconnected (all senders gone).
• recv_timeout(duration) waits, but only up to a time limit. If nothing arrives in time, you get Timeout.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel::<i32>();

    // Nothing sent yet: try_recv returns right away.
    println!("{:?}", rx.try_recv());

    thread::spawn(move || tx.send(7).unwrap()).join().unwrap();

    // A value is waiting, so this returns it without waiting.
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // The sender is gone and the channel is empty.
    println!("{:?}", rx.try_recv());
}

Output:

Err(Empty)
Ok(7)
Err(Disconnected)

▶ Try it Yourself

We join() the thread before receiving, so the value is already waiting and the output is always the same.

try_recv() is handy in a loop that has other work to do, like a game loop that checks for new messages once per frame. recv_timeout() is good for “give up if the worker takes too long”.

Bounded channels with sync_channel

A normal channel() has no size limit. If the sender is much faster than the receiver, messages pile up in memory. mpsc::sync_channel(n) makes a channel that holds at most n messages. When it’s full, send() waits until the receiver takes one out. This is called backpressure: a slow consumer automatically slows down a fast producer.

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

fn main() {
    // Room for only 2 messages at a time.
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // blocks while the buffer is full
        }
    });

    let total: i32 = rx.iter().sum();
    producer.join().unwrap();
    println!("total = {total}");
}

Output:

total = 15

▶ Try it Yourself

The producer can never get more than two numbers ahead of main. With sync_channel(0), there is no buffer at all: each send() waits until the receiver is ready to take the value, like handing over a package in person.

A small worker pool

Now let’s put it together. A worker pool is a few threads that take jobs from a shared queue and send back results. It uses two channels: one for jobs, one for results.

There’s one catch: a Receiver can’t be cloned (it’s single consumer). So the workers share the job receiver behind Arc<Mutex<_>>, and take turns pulling from it:

use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u64>();
    let (result_tx, result_rx) = mpsc::channel::<(u64, u64)>();

    // Receivers can't be cloned, so the workers share one behind Arc<Mutex<_>>.
    let job_rx = Arc::new(Mutex::new(job_rx));

    let mut workers = Vec::new();
    for _ in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        workers.push(thread::spawn(move || loop {
            let job = job_rx.lock().unwrap().recv();
            match job {
                Ok(n) => result_tx.send((n, n * n)).unwrap(),
                Err(_) => break, // no more jobs
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // tells the workers there's no more work

    let mut results: Vec<(u64, u64)> = result_rx.iter().collect();
    for w in workers {
        w.join().unwrap();
    }
    results.sort();
    for (n, square) in results {
        println!("{n} squared is {square}");
    }
}

Output:

1 squared is 1
2 squared is 4
3 squared is 9
4 squared is 16
5 squared is 25
6 squared is 36

▶ Try it Yourself

How it shuts down cleanly:

1. drop(job_tx) closes the job channel. Each worker’s recv() then returns Err, and the worker breaks out of its loop.
2. When a worker ends, its result_tx clone is dropped. We already dropped the original, so once all three workers finish, the result channel closes.
3. result_rx.iter() then stops, and we sort the results because workers finish in any order.

A small detail: job_rx.lock().unwrap().recv() is written in one let statement on purpose. The lock is released at the end of that statement, so the other workers can grab the next job while this one is still squaring its number.

Channels vs Arc<Mutex<T>>

Both are safe. They just fit different shapes of problem:

Use a channel when…	Use Arc<Mutex<T>> when…
Work flows in one direction: producers make things, a consumer handles them	Many threads need to read and update the same value (a counter, a cache)
You want a queue of jobs, events or log lines	The data is big and copying or moving it around would be wasteful
Each value has one owner at a time	Threads need to look at the current state, not a stream of updates

A good rule of thumb: if you can describe your program as “these threads tell that thread something”, use a channel. If it’s “these threads all edit this thing”, use a Mutex. Many real programs use both, like our worker pool.

Common mistakes

Using a value after sending it. send() takes ownership, the same as passing a value to a function. After sending, the value belongs to the receiver:

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

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("hi");
        tx.send(message).unwrap();
        println!("sent {message}");
    });

    println!("{}", rx.recv().unwrap());
}

error[E0382]: borrow of moved value: `message`
  --> src/main.rs:10:25
   |
 8 |         let message = String::from("hi");
   |             ------- move occurs because `message` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(message).unwrap();
   |                 ------- value moved here
10 |         println!("sent {message}");
   |                         ^^^^^^^ value borrowed here after move

This is Rust protecting you: if the sender could still change message, both threads would touch the same data. Print before sending, or send a .clone().

Moving the sender into a loop of threads. The first thread takes tx, so there’s nothing left for the second:

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

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 1..=3 {
        thread::spawn(move || {
            tx.send(id).unwrap();
        });
    }

    for value in rx {
        println!("{value}");
    }
}

error[E0382]: use of moved value: `tx`
  --> src/main.rs:8:23
   |
 5 |     let (tx, rx) = mpsc::channel();
   |          -- move occurs because `tx` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait
...
 8 |         thread::spawn(move || {
   |                       ^^^^^^^ value moved into closure here, in previous iteration of loop

The fix is let tx = tx.clone(); at the top of the loop, as in the multiple-producers example.

A program that never ends. This one has no compiler error, which makes it sneaky. If you loop over a receiver (for x in rx) while some sender is still alive, often the original tx in main, the loop waits forever. When your program “just hangs”, check that every sender gets dropped.

Next steps

std::sync::mpsc covers a lot, but you’ll meet two popular alternatives:

• crossbeam-channel adds multi-consumer channels (receivers you can clone) and a select! macro to wait on several channels at once.
• tokio::sync::mpsc is the channel for async code. If you’re using async/.await (see Async/Await), use it instead of the std one, because its recv().await doesn’t block the whole thread.

Recap

• mpsc::channel() gives you a (Sender, Receiver) pair. send() moves a value in, recv() waits for the next one.
• Clone the sender to have many producers. There is only ever one receiver.
• Looping over a receiver ends when all senders are dropped, so remember to drop() the original.
• try_recv() and recv_timeout() let you check without waiting forever. sync_channel(n) limits the queue to n messages.
• Messages from several threads arrive in any order: sort or sum them if you need stable output.
• Use channels when data flows from thread to thread. Use Arc<Mutex<T>> when threads share and edit one value.

Quick check

Quick check

1. What does "mpsc" stand for?

Many processes, shared cache Message passing, safe concurrency Multiple producer, single consumer Mutex protected shared channel

2. How do you let three threads send into the same channel?

Give each thread a tx.clone() Wrap the receiver in Rc Call mpsc::channel() three times and merge them You can't; a channel has only one sender

3. A for value in rx loop never finishes. What is the most likely cause?

The channel is full A sender (often the original tx in main) was never dropped recv() was called too early The values were not Copy

4. What does send() do when a sync_channel(2) already holds 2 messages?

It drops the oldest message It returns an error right away It panics It waits until the receiver takes a message out

Score: 0 / 4

Read more

• Using Message Passing to Transfer Data Between Threads — The Rust Book: the official introduction to channels, with more examples.
• std::sync::mpsc — standard library docs: every method on Sender, SyncSender and Receiver.
• Channels — Rust by Example: a short, runnable example with several producer threads.
• From Single-Threaded to Multithreaded Server — The Rust Book: builds a full thread pool with channels and Arc<Mutex<Receiver>>, step by step.
• tokio::sync::mpsc — docs.rs: the async channel you’ll use with Tokio, and how it differs from the std one.