38.3 Thread messaging

One technique to solve concurrency between threads involves communication. If threads are able to send each other messages, they can synchronize to, for instance, access shared resources.

The std::sync::mpsc module provides functions (see Table 1) to open communication channels between threads. In an MPSC (Multiple Producers, Single Consumer) model, only one thread listen for arriving messages, while one or more threads can emit messages.

In practice, the std::sync::mpsc module provides methods to pass ownership of an object from one thread to another.

38.3.1Send¶

The channels provided by the std::sync::mpsc module are made to transmit objects of the same type from one thread to another. The only condition is that the type must implement the Send trait.

• The Send trait is a marker trait, used by the compiler. We must never implement it manually for types.

• Most of built-in types and types from the standard library implement the Send trait.

• Any type composed of types that implement Send, will also automatically implement Send.

• Among the types that do not implement Send is Rc. Instead, one must use Arc, which is thread-safe.

38.3.2Asynchronous channel¶

In an asynchronous communication, the channel has an unlimited buffer in which it stores the messages:

• A send operation does not block the execution of the emitter thread.

• On the receiver thread, the receive operation blocks until a message is received.

To open an asynchronous channel, we use the channel() function. It returns a tuple containing first a transmitter (tx), then a receiver (rx):

:clear

let (tx, rx) = std::sync::mpsc::channel::<&str>();

Couldn't automatically determine type of variable `tx`.
Please give it an explicit type.

We spawn a thread that sends a message into the channel:

let handle = std::thread::spawn(move || { tx.send("Hello!") });

We wait for a message to arrive, and then print it:

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

Before quiting, we make sure the spawned has terminated:

handle.join().unwrap()?;

38.3.3Synchronous channel¶

In a synchronous communication, the channel has a size limited buffer in which it stores the messages:

• A send operation does not block the execution of the emitter thread, unless the buffer becomes full.

• On the receiver thread, the receive operation blocks until a message is received.

• A buffer size of zero is allowed. In that case all send operations will block until a receive operation is paired with it.

To open a synchronous channel, we use the sync_channel() function. It takes as argument the buffer size (i.e.: the maximum number of messages that can be stored in it). It returns a tuple containing first a transmitter (tx), then a receiver (rx).

In the following example, we create a synchronous channel with a buffer capicity of two messages only:

:clear

let (tx, rx) = std::sync::mpsc::sync_channel::<String>(2);
let start = std::time::Instant::now();

We spawn a thread that sends multiple messages and prints a timestamp after each successful send:

use std::error::Error;
type ThreadErr = Box<dyn Error + Send + Sync>;

let sender = std::thread::spawn(move || -> Result<(), ThreadErr> {
  for i in 1..=10 {
    let msg = format!("Message #{i}");
    tx.send(msg.clone())?;
    println!("[{:>4} ms] Sent \"{msg}\".", start.elapsed().as_millis());
  }
  Ok(())
});

We spawn a receiver thread that waits for messages to arrive, prints them, and pause for 20 ms after each received message:

let t = 20;
let receiver = std::thread::spawn(move || -> Result<(), ThreadErr> {
  for msg in rx {
    println!("[{:>4} ms] Received \"{msg}\", pause {t} ms.",
             start.elapsed().as_millis());
    std::thread::sleep(std::time::Duration::from_millis(t));
  }
  Ok(())
});

We wait for the two threads to terminate:

sender.join().unwrap().expect("Error with sender thread");
receiver.join().unwrap().expect("Error with receiver thread");

We can observer that the sender thread must wait each time for a slot to be freed inside the channel, starting at the third message.

see solution

38.3.4Multiple senders¶

As stated in the introduction, with the std::sync::mpsc module, multiple threads can send messages, while only one can receive them. To allow multiple threads to send messages on the same channel, we just have to clone the transmitter for each new thread.

Let us create an asynchronous channel:

:clear

let (tx1, rx) = std::sync::mpsc::channel::<&str>();

And clone the transmitter:

let tx2 = tx1.clone();

We can then spawn two sender threads using both transmitters:

let sender1 = std::thread::spawn(move || { tx1.send("Hello 1!") });
let sender2 = std::thread::spawn(move || { tx2.send("Hello 2!") });

And listen for arriving messages on the receiver:

for msg in rx {
  println!("Received message: {msg}");
}

We finally wait for the two threads to terminate:

sender1.join().unwrap().expect("Error with sender1 thread");
sender2.join().unwrap().expect("Error with sender2 thread");