Concurrency Fundamentals

Message Passing

Message passing is a concurrency model in which threads communicate by transferring ownership of discrete messages over a channel rather than mutating shared memory directly. In Rust, std::sync::mpsc (Multi-Producer, Single-Consumer) provides standard thread-safe channels where the type system guarantees that once a non-Copy value is sent into a channel, the sending thread can never access or mutate it again.

Ownership Transfer Across a Channel

A channel consists of two halves returned by mpsc::channel() (unbounded asynchronous queue) or mpsc::sync_channel(bound) (bounded backpressure queue):

  • Sender<T> (or SyncSender<T>): Implements Clone and Send, so multiple producer threads can send values concurrently.
  • Receiver<T>: Implements Send (but not Clone in mpsc), owned by the single consumer thread that pulls values in FIFO order.
Producer Thread 1 (Sender<Job>) ---\
                                    +---> [ FIFO Channel Queue ] ---> Consumer Thread (Receiver<Job>)
Producer Thread 2 (Sender<Job>) ---/
(Ownership of `Job` moves out of producer on `.send(job)` and into consumer on `.recv()`!)

Compile-Time Prevention of Use-After-Send (E0382)

In languages without move semantics, sending a pointer or object reference over a channel still leaves the sender with a reference to the object—allowing the sender and receiver to mutate it simultaneously. In Rust, tx.send(msg) takes msg by value: attempting to use msg after sending it fails at compile time with E0382:

// ❌ does not compile
use std::sync::mpsc;

fn main() {
    let (tx, _rx) = mpsc::channel();
    let payload = String::from("ownership transferred");

    tx.send(payload).unwrap();
    // Error E0382: `payload` was moved into the channel by `tx.send(payload)`!
    println!("sender still reading: {}", payload);
}

error[E0382]: borrow of moved value: `payload`
  --> src/main.rs:10:42
   |
 6 |     let payload = String::from("ownership transferred");
   |         ------- move occurs because `payload` has type `String`, which does not implement the `Copy` trait
 7 |
 8 |     tx.send(payload).unwrap();
   |             ------- value moved here
 9 |     // Error E0382: `payload` was moved into the channel by `tx.send(payload)`!
10 |     println!("sender still reading: {}", payload);
   |                                          ^^^^^^^ value borrowed here after move

Multi-Producer Worker Pipeline with Automatic Channel Shutdown

Iterating over a Receiver<T> (for msg in rx) blocks waiting for the next message and terminates the loop automatically when all Sender<T> clones have been dropped:

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

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

    for worker_id in 0..3 {
        let worker_tx = tx.clone();
        thread::spawn(move || {
            let msg = format!("result from worker {}", worker_id);
            worker_tx.send((worker_id, msg)).unwrap();
        }); // `worker_tx` drops here when the worker finishes.
    }

    // Drop the original `tx` on the main thread so `rx` closes once all workers finish:
    drop(tx);

    let mut results: Vec<(usize, String)> = rx.into_iter().collect();
    results.sort_by_key(|(id, _)| *id);

    for (_, msg) in results {
        println!("received: {}", msg);
    }
}

received: result from worker 0
received: result from worker 1
received: result from worker 2