10.3 Data Channels and Shared State

Passing Data Between Threads: Message Passing with Channels

While move closures allow transferring ownership of data to a new thread, often you need threads to communicate with each other after they have been spawned. Rust’s standard library provides channels for this purpose, enabling safe and efficient message passing. Please note that this section is important to the Assignment 2 code template.

std::sync::mpsc Channels

The std::sync::mpsc module provides “multiple producer, single consumer” channels. This means you can have multiple sending ends (Sender) but only one receiving end (Receiver).

The channel() function creates a new channel and returns a tuple containing the sending half (Sender<T>) and the receiving half (Receiver<T>). The T is the type of messages that will be sent through the channel.

This Rust code example demonstrates inter-thread communication using a channel, specifically a Multi-Producer, Single-Consumer (MPSC) channel.

use std::thread;
use std::sync::mpsc; // mpsc: multiple producer, single consumer

fn main() {
    // Create a new channel: tx is the transmitting end, rx is the receiving end
    let (tx, rx) = mpsc::channel();

    // Spawn a new thread and move the transmitting end into its closure
    thread::spawn(move ||   {
        // Inside the spawned thread, a String message is created.
        let message = String::from("Hello from the spawned thread!");
        println!("Spawned thread sending: \"{}\"", message);

        // The spawned thread then uses its tx handle to send the message through
        // the channel. The unwrap() handles potential errors (e.g., if the receiving
        // end has been dropped). Once message is sent, its ownership is transferred
        // to the channel. Note tx has been captured by the closure, but not rx.

        tx.send(message).unwrap(); // Send the message. unwrap() for simplicity.
    });

    // The main thread receives the message. This method attempts to receive a message
    // from the channel. It is a blocking call, meaning the main thread will pause
    // its execution and wait indefinitely until a message is available to be received
    // from the channel. Once a message is received, its ownership is transferred to
    // received_message in the main thread.

    let received_message = rx.recv().unwrap();

    // recv() blocks until a message is available.
    println!("Main thread received: \"{}\"", received_message);
}

Note

Note: because the closure uses tx inside its body, the Rust compiler will take ownership of tx from the surrounding scope, and move it into the closure environment. The closure does not reference rx in any way, so the compiler does not move rx into the closure. Instead, rx stays in the main thread and the main thread later calls rx.recv().

The output from this code is:

Spawned thread sending: "Hello from the spawned thread!"
Main thread received: "Hello from the spawned thread!"

In this example, the String message is moved from the spawned thread to the main thread via the channel. After tx.send(message), the message variable is no longer valid in the spawned thread, as its ownership has been transferred.

In summary, this example sets up a communication pipeline where:

• The spawned thread acts as a producer, generating a message.
• The main thread acts as a consumer, waiting for and receiving that message.

Channel Lifecycle and Iterators

In real applications, you often send more than one message. A Receiver can be used as an iterator, which makes it easy to process every message until the channel is closed. A channel is closed when all Sender handles are dropped.

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

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();
        }
        // tx goes out of scope and is dropped here
    });

    // The receiver as an iterator
    for received in rx {
        println!("Got: {}", received);
    }
    // The loop ends when tx is dropped and the channel is empty
}

This is the idiomatic way to handle streams of data in Rust. The for loop implicitly calls recv() and handles the termination of the channel automatically.

This pattern is fundamental for building concurrent applications where different threads need to share or coordinate data safely, avoiding common concurrency pitfalls like data races.

Multiple Producers

Since Sender implements the Clone trait, you can create multiple sending handles from a single channel. This allows multiple threads to send messages to the same Receiver. This Rust code example further illustrates inter-thread communication using MPSC channels, specifically demonstrating the “Multiple Producer” aspect.

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

fn main() {
    // A single channel giving us one transmitting end (tx) and one receiving end (rx).
    let (tx, rx) = mpsc::channel();

    // Clone the sender to create multiple producers
    let tx1 = mpsc::Sender::clone(&tx);

    // Thread 1: Sends "Hello"
    thread::spawn(move || {
        tx1.send(String::from("Hello")).unwrap();
        thread::sleep(Duration::from_millis(10));
    });

    // Thread 2: Sends "World"
    thread::spawn(move || {
        tx.send(String::from("World")).unwrap();
        thread::sleep(Duration::from_millis(10));
    });

    // Main thread receives messages
    // Messages received in the order sent, but thread scheduling is non-deterministic
    let msg1 = rx.recv().unwrap();
    let msg2 = rx.recv().unwrap();

    println!("Received: {} {}", msg1, msg2);
}

The important changes in this code:

• let tx1 = mpsc::Sender::clone(&tx);: While a single Sender can send multiple messages sequentially, it cannot be shared across multiple threads directly. To have multiple threads send messages to the same receiver, you must create a new, independent transmitting handle for each thread. The clone() method on the Sender type allows you to create these handles, all of which pipe data into the same channel.
• Thread 1 (thread::spawn(move || { tx1.send(String::from("Hello")).unwrap(); ... });): A new thread is spawned. It takes ownership of tx1 (the cloned sender). It sends the String::from("Hello") message through its tx1 handle.
• Thread 2 (thread::spawn(move || { tx.send(String::from("World")).unwrap(); ... });): Another new thread is spawned. It takes ownership of the original tx handle. It sends the String::from("World") message through its tx handle.
• Main Thread Receives (let msg1 = rx.recv().unwrap(); let msg2 = rx.recv().unwrap();): The main thread uses the single receiver rx. It calls rx.recv() twice. Each call will block until a message is available from either of the sending threads. The messages are received in the order they arrive in the channel, but the order in which the sending threads actually execute and send their messages is non-deterministic. This means msg1 could be “Hello” and msg2 “World”, or vice-versa, depending on which spawned thread happens to send its message first.

The output from this example in this case is:

Received: Hello World

The order of “Hello” and “World” in the output depends on which thread’s send operation completes first, but both messages will eventually be received.

**Essentially, **this example sets up a scenario where two separate threads (multiple producers) are sending messages to a single main thread (one consumer) via the same MPSC channel. The clone() method on the Sender is vital for enabling this multi-producer capability.

🎬Code Demo: Rust mpsc Channels

mpsc Channels

Rust mpsc Channels

derekmolloy.ie

🛰️

Edge AI Scenario

Two sensor threads — a temperature sensor and a humidity sensor — each send readings into the same channel. main consumes them in FIFO order. That's the mp in mpsc: many producers, one consumer.

The program

PC @ line 5

▶use std::sync::mpsc;
▶use std::thread;
▶ 
▶fn main() {
▶    let (tx, rx) = mpsc::channel();
▶ 
▶    let tx1 = tx.clone();  // temp sensor
▶    thread::spawn(move || {
▶        tx1.send(22.5).unwrap();
▶        // ... more readings ...
▶    });
▶ 
▶    let tx2 = tx.clone();  // humidity sensor
▶    thread::spawn(move || {
▶        tx2.send(45).unwrap();
▶        // ... more readings ...
▶    });
▶ 
▶    drop(tx);  // keep only the clones
▶ 
▶    for value in rx {        // blocks until next; ends when channel closes
▶        println!("got {:?}", value);
▶    }
▶}

▶ Play◀ StepStep ▶↺ Reset

SpeedSlowNormalFast

0 ms320 ms

What's happening

t = 000 ms

main calls mpsc::channel() — returns a (tx, rx) pair. tx is the Sender; rx is the Receiver.

Channel diagram

🌡

Temperature sensor

thread A · tx clone

sent0

tx dropped

💧

Humidity sensor

thread B · tx clone

sent0

tx dropped

Channel (FIFO queue)

depth 0

queue empty

📥

main (consumer)

owns rx · only one allowed

received0

recv log

— nothing yet —

Key takeaways

• tx.send(v) pushes a value into a FIFO queue. rx.recv() pops the head.
• tx can be .clone()'d and shared with many producer threads. rx cannot — only one consumer.
• The channel stays open as long as at least one sender exists. When the last sender drops, recv() returns Err — the consumer's signal to stop.

🧩Knowledge Check

Concept Match

Match the Channel Concepts

derekmolloy.ie

Drag each definition into its matching concept slot, then click Submit. Tap × to return a placed card to the pool.

mpsc::channel()

drag a definition here…

tx.send(val)

drag a definition here…

rx.recv()

drag a definition here…

for msg in rx

drag a definition here…

Sender::clone(&tx)

drag a definition here…

Definition Pool

Creates a new channel and returns a (Sender<T>, Receiver<T>) tuple; the T is inferred from the first value sent through it.

Transfers ownership of val into the channel; returns Err if the Receiver has already been dropped.

Creates an additional sending handle that routes messages into the same channel, enabling multiple producer threads from a single Receiver.

Treats the Receiver as an iterator, yielding each message in turn and ending the loop automatically when the channel closes.

Blocks the calling thread until a message arrives; returns Err when all Senders have been dropped and the channel is empty.

Submit

Shared State Concurrency: Mutexes and Read-Write Locks

While message passing is a powerful concurrency pattern, sometimes threads need to share access to the same mutable data. Rust provides synchronisation primitives like Mutex and RwLock to safely manage shared state.

Mutexes (std::sync::Mutex)

A Mutex (mutual exclusion) allows only one thread to access some data at any given time. To access the data protected by a mutex, a thread must first acquire the mutex’s lock.

• Acquiring the Lock: The lock() method is used to acquire the lock. This call will block the current thread until it’s its turn to have the lock.
• Releasing the Lock: When the MutexGuard (the smart pointer returned by lock()) goes out of scope, the lock is automatically released. This is an example of Rust’s RAII (Resource Acquisition Is Initialisation) in action, preventing common errors like forgetting to unlock.

In simple terms, a MutexGuard must act like a smart pointer because you need a safe way to access the data inside the mutex. Essentially, you lock a mutex to get at the data it protects, and the guard is the thing that gives you temporary, safe access to that data.

Example: Mutex in a Single Thread (Illustrative)

use std::sync::Mutex;

fn main() {
    let m = Mutex::new(5); // Create a mutex protecting an integer
    {
        let mut num = m.lock().unwrap(); // Acquire the lock. num is a MutexGuard.
        *num = 6; // Modify the data through the MutexGuard
        // num (MutexGuard) goes out of scope here, automatically releasing the lock
    } // Lock is released here

    println!("Mutex value: {:?}", m); // Output: Mutex value: Mutex { data: 6 }
}

This gives the output:

Mutex value: Mutex { data: 6, poisoned: false, .. }

This demonstrates the basic interaction: lock() returns a smart pointer that allows access to the inner data, and the lock is released when that smart pointer is dropped.

MutexGuard and RAII
The value returned by m.lock() is a MutexGuard<T> — a smart pointer that implements both Deref and DerefMut, giving you direct access to the protected data through the * operator. More importantly, it implements Drop: when the MutexGuard goes out of scope, the lock is released automatically. This is the RAII (Resource Acquisition Is Initialisation) pattern that Rust uses throughout its standard library. You never call an explicit unlock() function; the compiler ensures the lock is always released when the guard is dropped, even if the function returns early or an error occurs.

This design eliminates an entire class of bugs common in C/C++ where a programmer forgets to unlock a mutex, or where an early return or thrown exception bypasses the unlock call.

Mutex Poisoning
When a thread panics while holding a Mutex lock, Rust poisons the mutex. A poisoned mutex signals that the protected data may be in an inconsistent state — the thread that panicked may have partially modified it before the panic.

Subsequent calls to lock() on a poisoned mutex return Err(PoisonError) rather than Ok(MutexGuard). The code in this chapter uses .unwrap() on the lock result, which will itself panic if the mutex has been poisoned. In production code you would match on the result and decide whether to recover or propagate the error:

match shared_data.lock() {
    Ok(guard) => { /* use guard */ }
    Err(poisoned) => {
        // Recover the guard and attempt to repair the data, or abort
        let guard = poisoned.into_inner();
        eprintln!("Mutex was poisoned — data may be inconsistent");
    }
}

Avoiding Deadlocks
While Rust prevents data races, it does not prevent deadlocks. A deadlock occurs when Thread A waits for a lock held by Thread B, and Thread B waits for a lock held by Thread A. The simplest way to avoid deadlocks is to always acquire locks in the same order across all threads. If every thread always locks Mutex A before Mutex B, a deadlock cannot occur.

Sharing Mutexes Across Multiple Threads with Arc

To share a Mutex (or any data) across multiple threads, you need a way to manage multiple ownership. std::sync::Arc<T> (Atomic Reference Counted) is a thread-safe smart pointer that enables multiple threads to share ownership of data.

Like a MutexGuard, Arc<T> is also a smart pointer, but for a different purpose: it provides shared ownership of data across multiple threads. Instead of giving temporary access, it keeps a reference count, and when the last Arc pointing to the data is dropped, the data is freed. Arc must be a smart pointer because it needs to behave like a normal reference when you use the data (e.g., arc.some_method()), but it also needs built-in logic to safely manage and update the reference count across threads.

Example: Mutex with Arc for Shared Counter

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

fn main() {
    // Create an Arc to share ownership of the Mutex across threads
    // We are protecting a simple i32 with initial value 0
    // You could have used Mutex::new(100), or Mutex::new(String::from("hello")) etc.

    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];   // we need a way to rejoin all 10 threads

    for i in 0..10 {
        // Clone the Arc for each new thread. This increments the reference count.
        let counter_clone = Arc::clone(&counter);

        let handle = thread::spawn(move || {
            // Acquire the lock on the cloned Arc. This blocks until the lock is available.
            let mut num = counter_clone.lock().unwrap();
            *num += 1; // Increment the shared counter
            // num (MutexGuard) goes out of scope here, releasing the lock
            println!("Thread {} incremented counter to: {}", i, *num);
        });

        handles.push(handle);
    }

    // Wait for all threads to complete
    for handle in handles {
        handle.join().unwrap();
    }

    // Access the final value from the main thread
    println!("Final counter value: {}", *counter.lock().unwrap());
   // Output: Final counter value: 10
}

This gives the output (note the order of the output is arbitrary):

Thread 0 incremented counter to: 1
Thread 1 incremented counter to: 2
Thread 4 incremented counter to: 3
Thread 8 incremented counter to: 4
Thread 2 incremented counter to: 5
Thread 5 incremented counter to: 6
Thread 6 incremented counter to: 7
Thread 7 incremented counter to: 8
Thread 3 incremented counter to: 9
Thread 9 incremented counter to: 10
Final counter value: 10

This example demonstrates how Arc allows multiple threads to safely share and mutate a single Mutex-protected counter. Each thread gets its own Arc clone, ensuring that the Mutex remains valid as long as any thread holds a reference to it.

Choosing Between Channels and Shared State

Rust offers two distinct communication models for concurrent code: message passing (channels) and shared state (Mutex/Arc). Both are safe, but they suit different problems.

	Channels (mpsc)	Shared state (Arc<Mutex<T>>)
Data ownership	Transferred with the message	Shared; each thread borrows temporarily
Best for	Pipelines, task queues, producer–consumer	Counters, caches, configuration, sensor state
Coupling	Low — sender and receiver are decoupled	Higher — all threads must agree on lock protocol
Edge use case	Sensor thread → processing thread → transmit thread	Shared device state read by many threads
Deadlock risk	Low	Present if locks are acquired in inconsistent order

Use channels when one thread produces data and another consumes it. The pipeline shape maps cleanly onto channels, ownership is clear, and you avoid lock contention entirely.

Use shared state when multiple threads need random-access read/write to a single value and the overhead of serialising updates through a channel is not justified. The classic example on an edge device is a shared sensor reading that several threads query at arbitrary times.

Prefer channels for pipelines

The Rust community generally favours message passing for communication between logically independent tasks, and shared state for truly shared resources. If you find yourself sending a reference to mutable data through a channel, shared state with a mutex is likely the clearer model.

Performance Considerations

Mutex locking is not free. Each lock() call involves an atomic operation and potentially a system call to put the thread to sleep while waiting. For high-frequency operations — a tight sensor-reading loop or a motor control cycle — the locking overhead may be significant.

Practical guidance for edge devices:

• Batch updates: instead of locking once per sensor sample, accumulate several samples and lock once to write them all.
• Prefer RwLock for read-heavy data: if a configuration struct is written once but read by many threads, RwLock eliminates contention between readers.
• Consider atomic types for simple counters and flags: AtomicU32 and AtomicBool (covered in the next section) perform a single atomic instruction with no locking, making them the right choice when the shared value is a primitive.
• Minimise lock scope: hold the MutexGuard for as short a time as possible. Acquire the lock, copy out the value you need, drop the guard, then process — do not call blocking I/O while holding a lock.

Read-Write Locks (std::sync::RwLock)

An RwLock is like a Mutex with two different modes of locking:

• Read mode: many threads can read the data at the same time
• Write mode: only one thread can write, and all readers must wait

This makes it useful when your data is read a lot but only sometimes written, because it lets multiple readers run in parallel instead of forcing them to wait one at a time (as a Mutex would). So in contrast to a Mutex, which always gives one thread exclusive access, an RwLock lets you choose between:

• shared access (for reading), or
• exclusive access (for writing)

while still working seamlessly with Arc<T> when shared across threads. An RwLock can improve performance and liveness in read-heavy workloads by allowing many readers to run in parallel, but offers no benefit (and can even degrade performance) when writes are frequent or contention is high.

Example: RwLock for Shared Data

use std::thread;
use std::sync::{RwLock, Arc};
use std::time::Duration;

fn main() {
    let data = Arc::new(RwLock::new(String::from("Initial data")));
    let mut handles = vec![];

    // Multiple reader threads
    for i in 0..3 {
        let data_clone = Arc::clone(&data);
        let handle = thread::spawn(move || {
            let read_guard = data_clone.read().unwrap();
            // Acquire read lock
            println!("Reader {}: {}", i, *read_guard);
            thread::sleep(Duration::from_millis(5)); // Simulate work
            // read_guard goes out of scope, releasing read lock
        });
        handles.push(handle);
    }

    // One writer thread
    let data_clone_writer = Arc::clone(&data);
    let writer_handle = thread::spawn(move || {
        thread::sleep(Duration::from_millis(2)); // Give readers a head start
        let mut write_guard = data_clone_writer.write().unwrap();
        // Acquire write lock (blocks until no readers)
        *write_guard = String::from("Modified data");
        println!("Writer: Data modified.");
        // write_guard goes out of scope, releasing write lock
    });
    handles.push(writer_handle);

    // Another reader thread after writer
    let data_clone_reader_after = Arc::clone(&data);
    let reader_after_handle = thread::spawn(move || {
        thread::sleep(Duration::from_millis(10)); // Wait for writer to finish
        let read_guard = data_clone_reader_after.read().unwrap();
        println!("Reader (after write): {}", *read_guard);
    });
    handles.push(reader_after_handle);

    for handle in handles {
        handle.join().unwrap();
    }
}

This code gives the output:

Reader 0: Initial data
Reader 1: Initial data
Reader 2: Initial data
Writer: Data modified.
Reader (after write): Modified data

This example shows how multiple readers can access the data concurrently, but the writer thread will block until all read locks are released.

Condition Variables (std::sync::Condvar)

While Mutexes and RwLocks protect data, Condvar (Condition Variable) allows threads to wait for a specific condition to become true. A thread can “sleep” until another thread signals it that something has changed. This is much more efficient than “spinning” (repeatedly checking a value in a loop), which wastes CPU cycles — a critical consideration on battery-powered edge devices.

A Condvar is always used in conjunction with a Mutex.

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

fn main() {
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    let pair2 = Arc::clone(&pair);

    thread::spawn(move || {
        let (lock, cvar) = &*pair2;
        let mut started = lock.lock().unwrap();
        *started = true;
        // Signal the waiting thread that we have started
        cvar.notify_one();
    });

    let (lock, cvar) = &*pair;
    let mut started = lock.lock().unwrap();
    // Wait until the value in the mutex is true
    while !*started {
        started = cvar.wait(started).unwrap();
    }
    println!("Started!");
}

The wait() method automatically releases the mutex lock and puts the thread to sleep. When the thread is woken up by notify_one() or notify_all(), it automatically re-acquires the lock. Note the while loop: this is necessary because threads can occasionally wake up spuriously.

🧩Knowledge Check

Concept Match

Match the Shared-State Concepts

derekmolloy.ie

Drag each definition into its matching concept slot, then click Submit. Tap × to return a placed card to the pool.

Mutex<T>

drag a definition here…

MutexGuard<T>

drag a definition here…

Arc::clone(&x)

drag a definition here…

RwLock<T>

drag a definition here…

Condvar

drag a definition here…

Definition Pool

Allows any number of threads to hold a read lock concurrently, but grants only one thread an exclusive write lock; ideal for read-heavy shared data.

Lets a thread sleep efficiently until another thread calls notify_one() or notify_all() to signal that a condition protected by a Mutex has changed.

Wraps data so only one thread can access it at a time; calling lock() blocks until no other thread holds the lock.

Smart pointer returned by lock(); provides access to the protected data and releases the lock automatically when it drops (RAII).

Increments the atomic reference count and returns a new handle to the same heap-allocated data, allowing it to be moved into another thread.

Submit

Quiz

Select 0/1

What does Rust do to a Mutex when the thread holding its lock panics before releasing it?

derekmolloy.ie

Quiz

Select 0/1

A shared configuration struct is written once at startup and then read by many threads throughout the program's lifetime. Which primitive gives the best read throughput?

derekmolloy.ie