Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

16.2 Message Passing For Cross-Thread Data Transfer

16.2.1 Message Passing

One very popular technique for safe concurrency is called message passing. In this mechanism, threads (or actors) communicate by sending messages (data) to one another.

There is a famous saying in Go: Do not communicate by sharing memory; instead, share memory by communicating.

Go’s concurrency model reflects this idea. Rust also provides a concurrency approach based on message passing, specifically by using Channel from the standard library. Go has Channel as well, and the idea is similar.

16.2.2 Understanding Channel

You can think of a programming Channel as a directed waterway, such as a stream or a river. If you put a rubber duck into the river, it flows downstream until it reaches the end of the waterway.

A channel has two parts: a sender and a receiver. The sender is like the upstream point where you put the duck into the river, and the receiver is the downstream point where the duck eventually arrives. One part of the code uses the sender’s methods to send data, and another part checks the receiver for arriving messages. If either the sender or the receiver goes away, the channel is said to be closed.

The basic steps are:

  • Call the sender’s method to send data.
  • The receiver checks for and receives arriving data.
  • If either the sender or the receiver is dropped, the Channel is closed.

16.2.3 Creating a channel

Use mpsc::channel to create a Channel. mpsc stands for multiple producer, single consumer, meaning there can be multiple senders but only one receiver.

Calling this function returns a tuple with two elements: the sender and the receiver.

Take a look at this example:

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

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

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

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

  • First, mpsc::channel creates the Channel, and the returned tuple is destructured with pattern matching into tx and rx for the sender and receiver.

  • Next, a thread is created. The move keyword moves ownership of the sender tx into the spawned thread, because a thread must own the sender in order to send messages through the channel. The send method is used to send a message. It returns a Result: if the receiver has been dropped, the return value is Err; otherwise, it is Ok. Here we simply use unwrap for error handling, so if the receiver has been dropped, the program will panic.

  • The receiver has two methods for getting messages. Here we use recv (short for receive). It blocks this thread until a message arrives. The message is wrapped in a Result: if there is a message, it returns Ok; otherwise, it returns Err. We also use unwrap to handle errors simply.

Output:

Got: hi

The Sender’s send Method

The send method takes the data you want to send and returns a Result. If there is a problem—such as the receiver having been dropped—it returns Err.

Receiver Methods

  • recv: blocks the current thread until a value arrives in the Channel. Once a value is received, it returns a Result. If the sender has been closed, it returns Err.

  • try_recv: does not block the current thread. It returns a Result immediately. If data arrives, the Ok variant contains the received value; otherwise, it returns an error. It is often used inside a loop to check the result of try_recv. Once a message arrives, processing begins; if no message has arrived yet, other instructions can run in the meantime.

16.2.4 Ownership Transfer With channel

Ownership is very important in message passing because it helps you write safe concurrent code.

Take a look at this example:

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

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

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

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

This code adds println!("val is {val}"); and then tries to keep using the value in the thread after it has been passed to send.

Output:

$ cargo run
   Compiling message-passing v0.1.0 (/tmp/projects/message-passing)
error[E0382]: borrow of moved value: `val`
  --> src/main.rs:10:27
   |
 8 |         let val = String::from("hi");
   |             --- move occurs because `val` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(val).unwrap();
   |                 --- value moved here
10 |         println!("val is {val}");
   |                           ^^^ value borrowed here after move

For more information about this error, try `rustc --explain E0382`.
error: could not compile `message-passing` (bin "message-passing") due to 1 previous error

The error occurs because val was already moved when it was passed to send, so borrowing it again is not allowed.

The next example uses multiple sent values to observe how the receiver waits:

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

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

    thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for received in rx {
        println!("Got: {received}");
    }
}
  • The spawned thread sends each element in the Vector in a loop, pausing for one second after each send.
  • The main thread uses the receiver as an iterator (because it implements the Iterator trait), so there is no need to call recv explicitly. Each time a value arrives, it is printed. When the sender finishes and is dropped, the Channel closes and the loop ends. The program exits.

Output:

Got: hi
Got: from
Got: the
Got: thread

16.2.5 Creating Multiple Senders With Cloning

Let’s make a small change to the previous example:

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

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

    let tx1 = tx.clone();
    thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx1.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    thread::spawn(move || {
        let vals = vec![
            String::from("more"),
            String::from("messages"),
            String::from("for"),
            String::from("you"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for received in rx {
        println!("Got: {received}");
    }
}

There is now an extra spawned thread, and two spawned threads want to send messages to the main thread. In that case, you need two senders. To do that, simply call clone on the sender variable tx, which is the original line let tx1 = tx.clone();.

Output (receive order is nondeterministic; this is a representative local run):

Got: hi
Got: more
Got: from
Got: messages
Got: the
Got: for
Got: thread
Got: you

The data received by the receiver appears interleaved from the two senders.