10. Concurrency and Parallelism

Modern machines have many cores, and programs that use only one leave most of the hardware idle. Concurrency is notoriously hard because threads that share memory can corrupt it in ways that are timing-dependent and nearly impossible to reproduce. Rust's ownership system, extended by the Send and Sync traits, turns the most dangerous of these bugs, data races, into compile errors. This chapter covers threads, message passing, shared state, the safety guarantees behind them, and data parallelism with Rayon.

<a id="101-threads"></a>

10.1 Threads

A thread is an independent sequence of execution inside a process, sharing the process's memory with every other thread. This section covers creating threads, giving them data, borrowing safely with scoped threads, per-thread storage, and what happens when a thread panics.

Spawning and Joining Threads

A program starts with one thread, the main thread, which runs main. Everything it does happens one step after another. When work can be split into independent pieces (resizing a hundred images, handling several network clients, computing while waiting for disk), running those pieces on separate threads lets the operating system schedule them on different CPU cores at the same time, or interleave them on one core so that a blocked task does not stall the others.

Two terms are often confused:

• Concurrency is about structure: several tasks are in progress at overlapping times, possibly interleaved on a single core.
• Parallelism is about execution: several tasks literally run at the same instant on different cores.

A restaurant kitchen is a good picture. One chef juggling three dishes is concurrent. Three chefs each cooking a dish are parallel. Threads give you both.

Rust's std::thread::spawn creates an OS thread (a native thread scheduled by the kernel, with its own stack, by default 2 MiB) and runs a closure on it. It returns a JoinHandle<T>, where T is the closure's return value. Calling join() blocks the current thread until the spawned one finishes and returns Result<T, _>: Ok with the closure's value, or Err if the thread panicked.

   main thread  ──spawn──┬──────── work ─────── join() waits ──┬── continues
                         │                                     │
   worker thread         └── closure runs ── returns value ────┘

use std::thread;
use std::time::Duration;

fn main() {
    // spawn returns immediately; the closure runs on a new OS thread.
    let handle = thread::spawn(|| {
        for i in 1..=3 {
            println!("worker: step {i}");
            thread::sleep(Duration::from_millis(10));
        }
        42                                  // the thread's result
    });

    for i in 1..=2 {
        println!("main:   step {i}");
        thread::sleep(Duration::from_millis(10));
    }

    // join blocks until the worker ends and hands back its return value.
    let answer = handle.join().expect("worker panicked");
    println!("worker returned {answer}");   // worker returned 42

    // Several workers, each computing part of a result.
    let handles: Vec<thread::JoinHandle<u64>> = (0..4)
        .map(|id| thread::spawn(move || (id * 1000..(id + 1) * 1000).sum()))
        .collect();
    let total: u64 = handles.into_iter().map(|h| h.join().unwrap()).sum();
    println!("total = {total}");            // total = 7998000 (sum of 0..4000)

    // Threads can be named and sized with a Builder.
    let h = thread::Builder::new()
        .name("indexer".into())
        .stack_size(4 * 1024 * 1024)        // 4 MiB stack
        .spawn(|| thread::current().name().unwrap().to_string())
        .unwrap();                          // spawn can fail if the OS refuses
    println!("{}", h.join().unwrap());      // indexer
}

The interleaving of main: and worker: lines varies from run to run; the operating system decides who runs when. Two more facts are essential:

• Returning from main ends the process, killing any threads that are still running. Forgetting to join is a classic reason why a worker's output "sometimes disappears".
• Dropping a JoinHandle detaches the thread: it keeps running, but you can no longer wait for it or get its result.

OS threads are not free. Each one costs a kernel object, a stack reservation, and a few microseconds to create, and switching between them costs more. Tens or hundreds of threads are fine. Tens of thousands of mostly idle tasks are better served by async tasks (Chapter 11), and CPU-heavy work by a fixed pool (section 10.5). std::thread::available_parallelism() reports how many threads can usefully run at once.

Key Takeaways

• thread::spawn runs a closure on a new OS thread and returns a JoinHandle<T>; join() waits for it and returns Ok(value) or Err on panic.
• Concurrency is overlapping progress; parallelism is simultaneous execution.
• When main returns, the whole process exits, so join threads whose work must finish; dropping a handle detaches the thread.
• Use thread::Builder for names, stack sizes, and fallible spawning.

🧪 Practice

1. Spawn five threads that each print their index and the value of thread::current().id(). Run the program several times and observe the order.
2. Split the computation of the sum of squares of 1..=10_000_000 across available_parallelism() threads and compare the result with a single-threaded version.
3. Interview: Why does Rust map thread::spawn directly to OS threads instead of providing lightweight "green threads" like Go? (Hint: think about runtime size, FFI with C code, and the zero-cost abstraction principle.)

Moving Data into Threads

A spawned thread may outlive the function that created it, and even the stack frame that created its data. That is why thread::spawn has this signature:

// Simplified from std::thread
use std::thread::JoinHandle;

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
where
    F: FnOnce() -> T + Send + 'static,      // the closure itself
    T: Send + 'static,                      // its return value
{
    unimplemented!()
}

The two bounds encode exactly what can go wrong:

• 'static: the closure must not borrow anything from the spawning function's stack, because that frame may be gone while the thread is still running. It must own everything it uses (or borrow only 'static data).
• Send: everything the closure carries must be safe to hand to another thread (section 10.4).

The tool that satisfies 'static is the move closure. By default a closure borrows the variables it uses. move makes it take ownership of them, transferring the data into the thread. Think of it as moving house: the new thread gets the furniture itself, not the address of the old flat, which may be demolished tomorrow.

use std::sync::Arc;
use std::thread;

fn main() {
    let data = vec![1, 2, 3];

    // Without `move`, the closure would borrow `data`:
    // let h = thread::spawn(|| println!("{data:?}"));
    // error[E0373]: closure may outlive the current function, but it borrows
    //               `data`, which is owned by the current function

    let h = thread::spawn(move || {
        let sum: i32 = data.iter().sum();   // `data` is now owned by this thread
        sum
    });
    // println!("{data:?}");                // error[E0382]: borrow of moved value
    println!("{}", h.join().unwrap());      // 6

    // Need the data in several threads? Clone it, or share it with Arc.
    let config = Arc::new(String::from("verbose=true"));
    let mut handles = Vec::new();
    for id in 0..3 {
        let config = Arc::clone(&config);   // each thread owns one Arc handle
        handles.push(thread::spawn(move || format!("{id}: {config}")));
    }
    for h in handles {
        println!("{}", h.join().unwrap());  // 0: verbose=true  (then 1:, 2:)
    }
    println!("still usable here: {config}");

    // Moving out results: the return value travels back through join().
    let h = thread::spawn(|| (0..5).map(|x| x * x).collect::<Vec<u32>>());
    let squares = h.join().unwrap();        // ownership returns to main
    println!("{squares:?}");                // [0, 1, 4, 9, 16]
}

A move closure moves every captured variable. To move some and share others, create the values to move first (clones, Arc handles) and shadow the originals inside a block, as the loop above does with let config = Arc::clone(&config);. That idiom is so common it is worth memorizing.

Need	Technique
The thread is the only user	move the value in
Several threads read the same data	Arc<T>, cloned per thread
Several threads mutate the same data	Arc<Mutex<T>> or atomics (section 10.3)
The data outlives all the threads	Borrow it with scoped threads (next topic)
Results flow back to the spawner	Return them from the closure, or send them on a channel

Key Takeaways

• thread::spawn requires a Send + 'static closure, because the thread may outlive the current stack frame.
• move closures transfer ownership of captured values into the thread.
• Clone or Arc::clone before the move to give each thread its own handle; shadowing the name in the loop is the standard idiom.
• Results come back by value through join().

🧪 Practice

1. Write a program that spawns a thread to uppercase a String and returns it through join(), without cloning.
2. Given let words = vec!["a", "bb", "ccc"];, spawn one thread per word that prints its length. Make it compile once using move and to_string, and once using Arc<Vec<&str>>.
3. Interview: The 'static bound on thread::spawn does not mean the data lives forever. What does it actually require? (Hint: an owned String is 'static; a &String borrowed from a local is not.)

Scoped Threads

The 'static requirement of thread::spawn is often more than you need. If a function spawns threads to process a local vector and waits for all of them before returning, the vector clearly outlives the threads. Yet spawn still forces you to clone the data or wrap it in an Arc, because the compiler cannot see that the threads are joined in time.

Scoped threads (std::thread::scope, stable since Rust 1.63) make that guarantee visible to the compiler. thread::scope(|s| { ... }) creates a scope object s. Threads spawned with s.spawn may borrow any data that lives outside the scope, because scope does not return until every thread spawned inside it has finished. It is like a meeting room booked for an hour: guests may use the room's whiteboard, because the booking guarantees all of them leave before the room is handed back.

   let data = ...;                    data lives here ─────────────────┐
   thread::scope(|s| {                                                 │
       s.spawn(|| read &data);   ── thread A ──┐                       │
       s.spawn(|| read &data);   ── thread B ──┤                       │
   });   <── implicit join of A and B ─────────┘  (scope returns)      │
   use(data);                                                          │
                                      data dropped here ───────────────┘

The borrow rules still apply across threads: several threads may share &data, or one thread may hold &mut to a part, and the checker verifies this at compile time. Splitting a slice with chunks_mut gives each thread its own disjoint &mut region, with no locks at all.

use std::thread;

fn main() {
    let words = vec!["apple", "banana", "cherry", "date"];
    let mut counts = vec![0usize; words.len()];
    let mut longest_len = 0;

    thread::scope(|s| {
        // Shared borrows: both threads read `words` at the same time.
        s.spawn(|| println!("first: {}", words[0]));
        let longest = s.spawn(|| words.iter().map(|w| w.len()).max().unwrap());

        // Disjoint mutable borrows: each thread gets its own slot of `counts`.
        for (slot, word) in counts.iter_mut().zip(&words) {
            s.spawn(move || *slot = word.chars().filter(|c| "aeiou".contains(*c)).count());
        }

        // Scoped handles can be joined early to use a result inside the scope.
        longest_len = longest.join().unwrap();
    });                                     // all remaining threads joined here

    println!("{counts:?}");                 // [2, 3, 1, 2] (vowels per word)
    println!("longest word: {longest_len}");   // 6
    println!("{words:?}");                  // still owned and usable by main

    // Parallel in-place transform with chunks_mut: no Arc, no Mutex.
    let mut pixels = vec![10u8; 1_000_000];
    thread::scope(|s| {
        for chunk in pixels.chunks_mut(250_000) {
            s.spawn(move || chunk.iter_mut().for_each(|p| *p = p.saturating_mul(2)));
        }
    });
    println!("{} {}", pixels[0], pixels[999_999]);   // 20 20
}

The move in s.spawn(move || ...) moves the references (slot, word, chunk) into each thread, not the data they point at. That is a frequent point of confusion: inside a scope, move is about which thread owns the borrow, and the borrowed data stays where it is.

	thread::spawn	thread::scope + s.spawn
May borrow local data	No ('static required)	Yes
Joining	Manual, optional	Automatic at the end of the scope
Typical data sharing	Arc, cloning, channels	Plain &T, &mut to disjoint parts
Thread outlives caller	Possible (detached)	Never

Key Takeaways

• thread::scope guarantees that all threads spawned in it finish before it returns, so they may borrow non-'static data.
• The normal borrow rules apply across scoped threads: shared reads, or exclusive access to disjoint parts (for example via chunks_mut).
• Inside a scope, move moves references into threads, not the data.
• Prefer scoped threads over Arc when the threads' lifetime is naturally bounded by a function call.

🧪 Practice

1. Count the lines containing "error" in a Vec<String> of log lines by splitting it into four chunks processed by scoped threads, each returning its count via join().
2. Implement a parallel prefix: split an array into two halves with split_at_mut, square each half on a separate scoped thread, and print the result.
3. Interview: How can thread::scope guarantee that borrowed data outlives the threads, when an older API (thread::scoped, removed before Rust 1.0) turned out to be unsound? (Hint: the old API relied on a guard's destructor, and mem::forget is safe.)

Thread Local Storage

Some state is needed everywhere in a thread but should not be shared between threads: a per-thread random number generator, a scratch buffer reused across calls, a request id for logging, or a counter that would otherwise need expensive synchronization. Passing such values through every function signature is tedious, while a global static would be shared and need a lock.

Thread local storage (TLS) gives each thread its own independent copy of a variable. The thread_local! macro declares it, and each thread lazily initializes its copy on first access and destroys it when the thread exits. It is like each employee having their own desk drawer labeled "stapler": the label is the same for everyone, but no two people ever reach into the same drawer, so no coordination is needed.

Because a thread local is only ever touched by one thread, it can hold non-Sync types such as Cell and RefCell, which is how it is mutated. A thread local is a LocalKey<T>. Access goes through with, which passes a reference to a closure, or through the convenience methods get, set, take, replace (for Cell) and with_borrow/with_borrow_mut (for RefCell).

use std::cell::{Cell, RefCell};
use std::thread;

thread_local! {
    // `const` initializers skip lazy-init checks when the value is a constant.
    static CALLS: Cell<u32> = const { Cell::new(0) };
    static SCRATCH: RefCell<String> = RefCell::new(String::with_capacity(1024));
}

fn format_record(id: u32) -> usize {
    CALLS.set(CALLS.get() + 1);             // this thread's counter only
    SCRATCH.with_borrow_mut(|buf| {         // reuse one buffer per thread
        buf.clear();
        buf.push_str("record-");
        buf.push_str(&id.to_string());
        buf.len()
    })
}

fn main() {
    format_record(1);
    format_record(2);

    let h = thread::spawn(|| {
        format_record(99);
        CALLS.get()                         // a fresh copy, starting at 0
    });
    println!("worker calls: {}", h.join().unwrap());   // worker calls: 1
    println!("main calls:   {}", CALLS.get());         // main calls:   2

    // `with` gives a reference that cannot escape the closure.
    let len = SCRATCH.with(|s| s.borrow().len());
    println!("{len}");                      // 8 ("record-2")
}

Thread locals have some sharp edges worth knowing:

• Lifetime: the reference handed to with cannot outlive the closure, because the value is destroyed when its thread exits.
• Destructors: they run when a thread exits normally, but are not guaranteed to run for the main thread or in every platform situation, so do not rely on them for essential cleanup such as flushing files.
• Thread pools: a thread local in a pool worker persists across unrelated tasks that happen to run on that worker. That is good for caches and bad for "per-request" state.
• Async: tasks move between threads at .await points, so thread locals are the wrong tool for per-task state (Chapter 11).

Key Takeaways

• thread_local! gives each thread its own lazily initialized copy of a static, with no synchronization.
• Mutate thread locals through Cell/RefCell, using get/set or with_borrow_mut.
• Use const { .. } initializers for constant values.
• Thread-local state persists for the thread's whole life; beware of pools and async tasks that hop between threads.

🧪 Practice

1. Write a next_id() function backed by a thread-local Cell<u64> and show that two threads each get the sequence 0, 1, 2 independently.
2. Cache the results of an expensive fn classify(n: u64) -> bool in a thread-local RefCell<HashMap<u64, bool>> and measure the speedup when the same inputs repeat.
3. Interview: When would a thread-local counter merged at the end be faster than a shared AtomicU64, and what extra work does merging require? (Hint: think about cache lines bouncing between cores.)

Panics Across Threads

A panic in Rust unwinds the stack of the thread where it happens, running destructors along the way. It does not automatically take down other threads. That isolation is useful: one failing worker in a server should not kill the whole process. It also means failures can be silently lost if nobody checks for them.

The panic travels to whoever joins the thread. JoinHandle::join returns Err(payload), where the payload is a Box<dyn Any + Send> holding whatever was passed to panic!: usually a &'static str (for literal messages) or a String (for formatted ones). You can downcast it to recover the message. It is like a sealed envelope from a field team: the dispatcher learns that something went wrong only when they open the reply.

The rules by situation:

Situation	Effect of a panic in a spawned thread
thread::spawn, then join()	join() returns Err(payload)
thread::spawn, never joined	The message is printed to stderr; nothing else happens
thread::scope	The scope re-panics in the parent after all threads finish
Thread held a Mutex guard	The mutex becomes poisoned (section 10.3)
panic = "abort" in the profile	The entire process aborts immediately
Main thread panics	The process exits, ending all other threads

use std::any::Any;
use std::thread;

fn message(payload: &Box<dyn Any + Send>) -> String {
    if let Some(s) = payload.downcast_ref::<&str>() {
        s.to_string()                       // panic!("literal")
    } else if let Some(s) = payload.downcast_ref::<String>() {
        s.clone()                           // panic!("formatted {x}")
    } else {
        "unknown panic payload".to_string()
    }
}

fn main() {
    let jobs = [4, 0, 2];
    let handles: Vec<_> = jobs
        .into_iter()
        .map(|d| thread::spawn(move || {
            if d == 0 {
                panic!("cannot divide by {d}");   // unwinds this thread only
            }
            100 / d
        }))
        .collect();

    for (d, h) in jobs.iter().zip(handles) {
        match h.join() {
            Ok(v) => println!("100 / {d} = {v}"),
            Err(p) => println!("job {d} failed: {}", message(&p)),
        }
    }
    // 100 / 4 = 25
    // job 0 failed: cannot divide by 0
    // 100 / 2 = 50
    // (the default panic hook also prints the panic message to stderr)

    // In a scope, a panicking child makes the scope itself panic.
    let r = std::panic::catch_unwind(|| {
        thread::scope(|s| {
            s.spawn(|| panic!("child failed"));
        })
    });
    println!("scope panicked: {}", r.is_err());   // scope panicked: true
}

Panics are for bugs, not for expected failures. Worker threads that can fail in normal operation should return Result from the closure (so join yields Ok(Err(e))), and reserve the Err from join for genuine bugs. Applications that want a crashing worker to stop everything can set panic = "abort" in Cargo.toml, or install a hook with std::panic::set_hook to log panics centrally.

Key Takeaways

• A panic unwinds only its own thread; the parent learns about it through join(), which returns Err(Box<dyn Any + Send>).
• Downcast the payload to &str or String to recover the message.
• Scoped threads propagate a child's panic to the parent; panicking while holding a Mutex poisons it.
• Return Result from workers for expected errors, and keep panics for bugs.

🧪 Practice

1. Spawn ten workers where every third one panics, and report how many succeeded and the messages of those that failed.
2. Change the workers to return Result<u32, String> instead of panicking and handle the nested Result<Result<u32, String>, _> from join().
3. Interview: Why is the panic payload typed as Box<dyn Any + Send> rather than String? (Hint: look at what std::panic::panic_any accepts, and which thread ends up owning the payload.)

<a id="102-message-passing"></a>

10.2 Message Passing

"Do not communicate by sharing memory; instead, share memory by communicating." Message passing gives each piece of data exactly one owner at a time and moves it between threads through channels. This section covers the standard mpsc channel, its ownership rules, capacity limits, the more capable crossbeam channels, and the producer-consumer designs built from them.

Channels with mpsc

Sharing memory between threads requires locks, and locks require discipline: every access must be guarded, and forgetting one is a bug. An alternative is to never share at all. Each thread owns its data, and when another thread needs it, the data is sent as a message. Ownership moves with the message, so at every moment exactly one thread can touch it. That is the same single-owner rule as Chapter 3, now stretched across threads.

A channel is the pipe that carries messages. It has two halves: a sender (transmitter) that puts values in, and a receiver that takes them out in the order they were sent (FIFO). Rust's standard channel lives in std::sync::mpsc, which stands for multiple producer, single consumer: the sender can be cloned so many threads produce, but only one receiver consumes.

Picture a mailbox at a post office. Anyone with a key to the slot (a sender) can drop letters in, and keys can be copied. There is exactly one person with the key to the back door who collects the letters, in the order they arrived.

   producer A ── tx  ──┐
   producer B ── tx2 ──┼──>  [ msg3 | msg2 | msg1 ]  ──>  rx  ── consumer
   producer C ── tx3 ──┘          channel (FIFO)

mpsc::channel() returns (Sender<T>, Receiver<T>). send(value) moves the value into the channel and never blocks for this unbounded kind of channel. recv() blocks until a message arrives. Both return Result, because the other side may have been dropped: send fails when the receiver is gone, and recv fails when all senders are gone and the queue is empty. That second rule gives you a natural "end of stream" signal.

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

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

    let producer = thread::spawn(move || {
        for word in ["alpha", "beta", "gamma"] {
            let msg = word.to_string();
            tx.send(msg).unwrap();          // ownership of `msg` moves into the channel
            // println!("{msg}");           // error[E0382]: borrow of moved value
            thread::sleep(Duration::from_millis(5));
        }
    });                                     // tx dropped here: the channel closes

    // The receiver is an iterator that ends when every sender is dropped.
    for msg in &rx {
        println!("got {msg}");              // got alpha / got beta / got gamma
    }
    producer.join().unwrap();

    // Non-blocking and time-limited receives.
    let (tx, rx) = mpsc::channel::<u32>();
    println!("{:?}", rx.try_recv());        // Err(Empty): nothing yet, does not block
    tx.send(7).unwrap();
    println!("{:?}", rx.recv_timeout(Duration::from_millis(50)));   // Ok(7)
    drop(tx);
    println!("{:?}", rx.recv());            // Err(RecvError): channel closed
}

Receive method	Behavior when empty	Behavior when closed and empty
recv()	Blocks	Err(RecvError)
try_recv()	Err(TryRecvError::Empty)	Err(TryRecvError::Disconnected)
recv_timeout(d)	Blocks up to d, then Err(Timeout)	Err(Disconnected)
for msg in &rx / rx.iter()	Blocks	Loop ends
rx.try_iter()	Ends immediately	Ends

Since Rust 1.67 the standard mpsc is implemented on top of the crossbeam channel algorithms, so it is fast. Its limitation is the single consumer: the Receiver cannot be cloned or shared (see the crossbeam topic below for multi-consumer channels).

Key Takeaways

• A channel moves values between threads; ownership travels with the message, so no locking of the data is needed.
• mpsc::channel() returns a cloneable Sender and a single Receiver; messages arrive in FIFO order.
• recv() returns Err once all senders are dropped and the queue is empty, which ends for msg in &rx loops.
• Use try_recv for polling and recv_timeout to wait with a deadline.

🧪 Practice

1. Spawn a thread that sends the numbers 1 to 10 over a channel and have the main thread print their running total as they arrive.
2. Send a Vec<u8> of one megabyte through a channel and explain, in terms of memory, what is copied and what is not.
3. Interview: How does a consumer loop such as for msg in rx know when to stop, and what common mistake makes it hang forever? (Hint: count the Sender values that are still alive, including the original one in the spawning thread.)

Senders, Receivers, and Ownership

The channel's two ends are ordinary Rust values, and their ownership determines the lifetime of the channel itself. Understanding who owns which end is the key to channels that shut down cleanly.

• Sender<T> is Clone. Each clone is an independent handle to the same channel. The channel stays open for receiving as long as at least one sender exists anywhere. Sender<T> is Send and (since Rust 1.72) Sync, so it can be moved to threads or shared by reference.
• Receiver<T> is not Clone. It can be moved to another thread (Send) but not shared between threads (!Sync). When it is dropped, every later send returns Err(SendError(value)), giving the value back so it is not lost.
• Messages are owned values. send takes T by value. To send a borrowed thing you must send something owned instead: a clone, an Arc, or a String instead of a &str. T must be Send, so an Rc cannot be sent.

The most common bug is the forgotten original sender. When producers each get a clone, the original tx in the spawning thread still exists, so the channel never closes and the consumer's loop waits forever. Drop it explicitly, or move it into the last producer.

use std::sync::mpsc::{self, Sender};
use std::thread;

#[derive(Debug)]
enum Event {
    Progress { worker: usize, done: u32 },
    Finished(usize),
}

fn worker(id: usize, tx: Sender<Event>) {
    for done in [50, 100] {
        tx.send(Event::Progress { worker: id, done }).unwrap();
    }
    tx.send(Event::Finished(id)).unwrap();
}                                           // this clone of the sender drops here

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

    for id in 0..3 {
        let tx = tx.clone();                // one sender per producer
        thread::spawn(move || worker(id, tx));
    }
    drop(tx);                               // drop the original, or the loop never ends

    let mut finished = Vec::new();
    for event in rx {                       // ends when all three clones are dropped
        match event {
            Event::Progress { worker, done } if done == 100 => {
                println!("worker {worker} reached 100%")
            }
            Event::Progress { .. } => {}
            Event::Finished(id) => finished.push(id),
        }
    }
    println!("{} workers finished", finished.len());   // 3 workers finished

    // A dropped receiver hands the unsent value back.
    let (tx, rx) = mpsc::channel::<String>();
    drop(rx);
    let err = tx.send("lost?".to_string()).unwrap_err();
    println!("returned: {}", err.0);        // returned: lost?
}

Using an enum as the message type is idiomatic: a single channel can carry several kinds of events, and match makes sure the consumer handles all of them. A related pattern is request-response: include a reply Sender in the request so the worker can answer the specific caller.

use std::sync::mpsc::{self, Sender};
use std::thread;

enum Request {
    Square(u64, Sender<u64>),               // carries its own reply channel
    Shutdown,
}

fn main() {
    let (tx, rx) = mpsc::channel::<Request>();
    let server = thread::spawn(move || {
        for req in rx {
            match req {
                Request::Square(n, reply) => reply.send(n * n).unwrap(),
                Request::Shutdown => break,
            }
        }
    });

    let (reply_tx, reply_rx) = mpsc::channel();
    tx.send(Request::Square(12, reply_tx)).unwrap();
    println!("{}", reply_rx.recv().unwrap());   // 144
    tx.send(Request::Shutdown).unwrap();
    server.join().unwrap();
}

Key Takeaways

• The channel stays open while any Sender clone is alive; the receiver sees the end of the stream only after all of them are dropped.
• Drop the original sender after handing clones to producers.
• Receiver cannot be cloned or shared; a failed send returns the value.
• Use an enum for message types, and embed a reply Sender for request-response.

🧪 Practice

1. Remove the drop(tx) line from the event example and observe what happens. Then fix it without drop, by restructuring the loop that spawns workers.
2. Build a "logger thread" that receives LogLine { level, text } messages from four worker threads and prints them with a sequence number.
3. Interview: Why does mpsc::SendError<T> contain the unsent T? (Hint: send took the value by ownership; what would happen to it otherwise?)

Bounded vs Unbounded Channels

An unbounded channel never blocks its senders: if the consumer is slower than the producers, messages simply pile up in memory. For short bursts that is fine. For a sustained mismatch (a fast network reader feeding a slow database writer, say) the queue grows without limit until the process runs out of memory.

A bounded channel has a fixed capacity. When it is full, send blocks until the consumer makes room. That blocking is called backpressure: a slow consumer automatically slows its producers down to its own pace, and memory use is capped. Think of a conveyor belt with a fixed number of trays: when every tray is full, the person loading the belt must wait.

In std::sync::mpsc, mpsc::sync_channel(n) creates a bounded channel with capacity n and returns a SyncSender<T>. A capacity of 0 is a special case called a rendezvous channel: there is no buffer at all, so each send blocks until a receiver is actually taking the value. That makes it a handshake between the two threads.

   unbounded:  producer ──> [ ][ ][ ][ ][ ][ ][ ][ ] ... grows forever ──> consumer
   bounded(3): producer ──> [x][x][x]  full: send() blocks  ──> consumer
   bounded(0): producer ──> (hand-off only when both are ready) ──> consumer

use std::sync::mpsc;
use std::thread;
use std::time::{Duration, Instant};

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u32>(2);   // capacity 2
    let start = Instant::now();

    let producer = thread::spawn(move || {
        for i in 0..6 {
            tx.send(i).unwrap();            // blocks while 2 items are queued
            println!("{:>4} ms  sent {i}", start.elapsed().as_millis());
        }
    });

    for v in rx {
        thread::sleep(Duration::from_millis(50));   // a slow consumer
        println!("{:>4} ms  handled {v}", start.elapsed().as_millis());
    }
    producer.join().unwrap();
    // The first sends return immediately; after that each `sent` line appears
    // only after a `handled` line: the producer is held to the consumer's pace.

    // try_send never blocks; it reports a full channel instead.
    let (tx, _rx) = mpsc::sync_channel::<u8>(1);
    println!("{:?}", tx.try_send(1));       // Ok(())
    println!("{:?}", tx.try_send(2));       // Err(TrySendError::Full(..))
}

	Unbounded (channel)	Bounded (sync_channel(n))	Rendezvous (sync_channel(0))
send blocks	Never	When n items are queued	Until a receiver takes the value
Memory	Unlimited	At most n messages	No buffer
Backpressure	None	Yes	Maximal
Typical use	Low-volume events, logging	Pipelines, work queues	Hand-offs and synchronization points

Choosing a capacity is a trade-off. A small buffer gives tight memory bounds but makes producers stall on every hiccup of the consumer. A larger buffer absorbs bursts. A common starting point is a small multiple of the number of producers, tuned by measurement. Default to bounded channels in long-running systems: an unbounded queue only postpones an overload until memory runs out.

Key Takeaways

• Unbounded channels never block senders and can grow without limit when the consumer falls behind.
• Bounded channels block senders when full, providing backpressure and capped memory use.
• sync_channel(0) is a rendezvous: every send waits for a matching receive.
• Prefer bounded channels in long-running pipelines; use try_send when a producer must not block.

🧪 Practice

1. Rerun the example with capacities 0, 1, and 10 and describe how the interleaving of sent and handled lines changes.
2. Write a producer that uses try_send and drops (and counts) messages when the channel is full, simulating a metrics pipeline that prefers losing data to blocking.
3. Interview: A service reading requests from a socket into an unbounded channel runs out of memory under load. Explain why and propose two fixes. (Hint: backpressure, and what to do with excess requests at the edge.)

Crossbeam Channels

The standard channel covers the common case, but some designs need more: several consumers pulling from the same queue, a thread that waits on any of several channels at once, or timers that behave like channels. The crossbeam-channel crate (part of the crossbeam family of concurrency tools) provides all of these, and it is the de facto standard for advanced channel use in synchronous Rust.

Its main differences from std::sync::mpsc:

Feature	std::sync::mpsc	crossbeam_channel
Consumers	One (Receiver is not Clone)	Many (Receiver: Clone): MPMC
Constructors	channel(), sync_channel(n)	unbounded(), bounded(n)
Wait on several channels	Not supported	select! macro and Select
Timer channels	No	after(d), tick(d), never()
Inspect state	No	len(), is_empty(), is_full()

Multiple consumers turn a channel into a work queue: each message is delivered to exactly one of the receivers, whichever asks first, so idle workers automatically pick up the next job. The select! macro blocks until one of several operations is ready and runs the matching branch, which is how a worker can listen for jobs and a shutdown signal and a timeout at the same time. Think of select! as a receptionist watching several phone lines and answering whichever rings first.

// Cargo.toml: crossbeam-channel = "0.5"
use crossbeam_channel::{bounded, select, tick, unbounded};
use std::thread;
use std::time::Duration;

fn main() {
    // MPMC work queue: three workers share one Receiver.
    let (jobs_tx, jobs_rx) = bounded::<u64>(16);
    let (results_tx, results_rx) = unbounded::<(usize, u64)>();

    let workers: Vec<_> = (0..3)
        .map(|id| {
            let jobs = jobs_rx.clone();      // cloning the Receiver is allowed here
            let results = results_tx.clone();
            thread::spawn(move || {
                for n in jobs {              // each job goes to exactly one worker
                    results.send((id, n * n)).unwrap();
                }
            })
        })
        .collect();
    drop(results_tx);                       // keep only the workers' clones

    for n in 1..=9 {
        jobs_tx.send(n).unwrap();
    }
    drop(jobs_tx);                          // closes the queue: workers exit their loops

    let mut sum = 0;
    for (_worker, sq) in &results_rx {
        sum += sq;
    }
    workers.into_iter().for_each(|w| w.join().unwrap());
    println!("sum of squares = {sum}");     // sum of squares = 285

    // select!: wait on data, a periodic tick, and a shutdown signal at once.
    let (data_tx, data_rx) = unbounded::<&str>();
    let (stop_tx, stop_rx) = bounded::<()>(0);
    let ticker = tick(Duration::from_millis(20));

    thread::spawn(move || {
        data_tx.send("hello").unwrap();
        thread::sleep(Duration::from_millis(50));
        stop_tx.send(()).unwrap();
    });

    let mut ticks = 0;
    loop {
        select! {
            recv(data_rx) -> msg => if let Ok(m) = msg { println!("data: {m}") },
            recv(ticker) -> _ => ticks += 1,
            recv(stop_rx) -> _ => break,     // shutdown wins the loop
        }
    }
    println!("saw ticks: {}", ticks > 0);
    // data: hello
    // saw ticks: true      (about two 20 ms ticks fit before the 50 ms stop)
}

When several branches of select! are ready at once, one is chosen at random, so no channel can starve the others. A default branch (or default(timeout)) makes the select non-blocking or time-limited.

For new code, a good rule is: start with std::sync::mpsc, and switch to crossbeam as soon as you need multiple consumers or select!. In async code, use the async runtime's own channels instead (Chapter 11), because both of these block the OS thread.

Key Takeaways

• crossbeam-channel provides MPMC channels: receivers can be cloned, and each message is delivered to exactly one of them.
• select! waits on several channel operations and runs whichever becomes ready first, choosing randomly among ready ones.
• tick, after, and never turn timers into channels that compose with select!.
• Use std::sync::mpsc for simple cases and crossbeam for work queues and multiplexing; neither is suitable inside async tasks.

🧪 Practice

1. Port the producer-consumer example from the bounded channel topic to crossbeam and add a second consumer. Count how many items each one handled.
2. Write a loop with select! that receives messages but gives up with a message after 200 ms of silence, using default(Duration::from_millis(200)) or after.
3. Interview: Why does select! pick randomly among ready branches rather than always taking the first one listed? (Hint: imagine one channel that always has data.)

Producer-Consumer Patterns

Channels are building blocks. Most concurrent programs combine them into a few recurring shapes, and recognizing those shapes makes designs easier to reason about. In each of them, producers create work or data, consumers process it, and channels decouple their speeds and lifetimes.

   Fan-out / fan-in (worker pool)          Pipeline (stages)

              ┌─> worker 1 ─┐              reader ──> parser ──> writer
   jobs ──────┼─> worker 2 ─┼──> results        ch1        ch2
              └─> worker 3 ─┘              each stage is a thread; data flows
   one MPMC job queue, one result channel  through bounded channels

Pattern	Shape	Good for
Fan-in	Many producers, one consumer (mpsc)	Logging, metrics, collecting results
Fan-out	One producer, many consumers (MPMC)	Distributing independent jobs
Worker pool	Fan-out + fan-in	CPU-bound batches, request handling
Pipeline	Chain of stages linked by channels	Streaming transforms (read, parse, write)
Actor	One thread owns state; others send it commands	Shared state without locks

Three concerns appear in every design:

• Shutdown. Closing the input channel (dropping all senders) should cascade through the system: each stage finishes its loop, drops its own senders, and so closes the next channel. If a stage keeps an extra sender alive, the shutdown stalls.
• Backpressure. Use bounded channels between stages so that the slowest stage sets the pace instead of letting queues grow.
• Errors. Decide whether a failing item is sent downstream as a Result, logged and skipped, or causes the whole pipeline to stop.

Here is a three-stage pipeline with bounded channels and cascading shutdown:

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

fn main() {
    let (lines_tx, lines_rx) = sync_channel::<String>(4);
    let (nums_tx, nums_rx) = sync_channel::<Result<i64, String>>(4);

    // Stage 1: produce raw input.
    let reader = thread::spawn(move || {
        for line in ["10", "20", "x", "30"] {
            lines_tx.send(line.to_string()).unwrap();
        }
    });                                     // lines_tx dropped -> stage 2 ends

    // Stage 2: parse, sending errors downstream as values.
    let parser = thread::spawn(move || {
        for line in lines_rx {
            let parsed = line.parse::<i64>().map_err(|e| format!("{line:?}: {e}"));
            nums_tx.send(parsed).unwrap();
        }
    });                                     // nums_tx dropped -> stage 3 ends

    // Stage 3 (main thread): aggregate.
    let (mut sum, mut errors) = (0, Vec::new());
    for item in nums_rx {
        match item {
            Ok(n) => sum += n,
            Err(e) => errors.push(e),
        }
    }
    reader.join().unwrap();
    parser.join().unwrap();
    println!("sum = {sum}");                // sum = 60
    println!("errors = {errors:?}");
    // errors = ["\"x\": invalid digit found in string"]
}

And the actor pattern: one thread owns the state outright and serves commands, so no Mutex is needed because no data is ever shared:

use std::collections::HashMap;
use std::sync::mpsc::{channel, Sender};
use std::thread;

enum Cmd {
    Incr(String),
    Get(String, Sender<u32>),
}

fn spawn_counter() -> Sender<Cmd> {
    let (tx, rx) = channel();
    thread::spawn(move || {
        let mut counts: HashMap<String, u32> = HashMap::new();   // owned by the actor
        for cmd in rx {
            match cmd {
                Cmd::Incr(k) => *counts.entry(k).or_default() += 1,
                Cmd::Get(k, reply) => {
                    let _ = reply.send(counts.get(&k).copied().unwrap_or(0));
                }
            }
        }
    });
    tx
}

fn main() {
    let counter = spawn_counter();
    let clients: Vec<_> = (0..4)
        .map(|_| {
            let c = counter.clone();
            thread::spawn(move || {
                for _ in 0..250 {
                    c.send(Cmd::Incr("hits".into())).unwrap();
                }
            })
        })
        .collect();
    clients.into_iter().for_each(|h| h.join().unwrap());

    let (reply_tx, reply_rx) = channel();
    counter.send(Cmd::Get("hits".into(), reply_tx)).unwrap();
    println!("hits = {}", reply_rx.recv().unwrap());   // hits = 1000
}

The actor version never loses an increment, even though four threads update the same map, because only the actor thread ever touches it. Commands queue up in the channel and are applied one by one.

Key Takeaways

• Fan-in, fan-out, worker pools, pipelines, and actors cover most channel designs.
• Shutdown should cascade: dropping a stage's input senders ends its loop, which drops its output senders.
• Bounded channels between stages provide backpressure; send errors downstream as Result values when they should not stop the pipeline.
• An actor owns state on one thread and serves commands, replacing a lock with a queue.

🧪 Practice

1. Add a fourth stage to the pipeline that formats each number as a string, and make stage 3 stop the pipeline at the first error instead of collecting errors. Make sure every thread still exits.
2. Build a worker pool of four threads that compute the length of the Collatz sequence for each of 1..=100_000 and report the number with the longest one.
3. Interview: When would you choose an actor with a channel over an Arc<Mutex<HashMap>>, and when would the mutex be better? (Hint: consider long operations under the lock, latency of a round trip, and how state is read.)

<a id="103-shared-state-concurrency"></a>

10.3 Shared-State Concurrency

Sometimes several threads genuinely need the same data at the same time: a cache, a connection pool, a set of counters. This section covers the synchronization primitives that make shared mutable state safe (locks, atomics, condition variables, and barriers) and the deadlocks that careless locking can cause.

Mutex and Lock Poisoning

When two threads update the same value without coordination, their operations interleave in unpredictable ways. count += 1 is really three steps (load, add, store), and if two threads both load 5, both store 6, one increment is lost. The fix is mutual exclusion: make sure only one thread at a time runs the code that touches the shared value. That code is called a critical section, and the tool that guards it is a mutex (mutual exclusion lock).

In most languages a mutex is a separate object you must remember to lock before touching the data it "protects". Nothing enforces the association. Rust's Mutex<T> is different: the mutex owns the data. The only way to reach the T inside is to call lock(), which waits until the lock is free and returns a MutexGuard<T>. The guard dereferences to &mut T, and when it is dropped the lock is released automatically (RAII, Chapter 3). Accessing the data without holding the lock is not a mistake you can make, because there is no path to it.

The analogy is a single-occupancy restroom with a key attached to a large wooden block. You cannot enter without the key, only one person can hold it, and returning it is part of leaving.

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

fn main() {
    let counter = Arc::new(Mutex::new(0u64));   // Arc to share, Mutex to mutate

    let handles: Vec<_> = (0..8)
        .map(|_| {
            let counter = Arc::clone(&counter);
            thread::spawn(move || {
                for _ in 0..10_000 {
                    let mut n = counter.lock().unwrap();   // blocks until acquired
                    *n += 1;                                // guard derefs to &mut u64
                }                                           // guard dropped: unlocked
            })
        })
        .collect();
    handles.into_iter().for_each(|h| h.join().unwrap());
    println!("{}", *counter.lock().unwrap());   // 80000: no lost updates

    // Keep critical sections short: copy what you need, then release.
    let names = Mutex::new(vec!["ana".to_string(), "ben".to_string()]);
    let snapshot = names.lock().unwrap().clone();   // guard is a temporary: dropped at `;`
    for n in &snapshot {
        println!("{n}");                    // slow work done without holding the lock
    }

    // With exclusive access there is no need to lock at all.
    let mut m = Mutex::new(5);
    *m.get_mut().unwrap() += 1;             // &mut Mutex proves no one else has it
    println!("{}", m.into_inner().unwrap());   // 6
}

Why does lock() return a Result? Because of poisoning. If a thread panics while holding the guard, it may have left the data half-updated, breaking an invariant (a balance debited but not yet credited). Rust marks the mutex as poisoned, and every later lock() returns Err(PoisonError) to warn you. The error still contains the guard, so you can decide to recover:

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

fn main() {
    let accounts = Arc::new(Mutex::new(vec![100, 0]));

    let a = Arc::clone(&accounts);
    let _ = thread::spawn(move || {
        let mut acc = a.lock().unwrap();
        acc[0] -= 50;                       // debit done...
        panic!("crash before the credit");  // ...credit never happens
    })
    .join();

    match accounts.lock() {
        Ok(acc) => println!("consistent: {acc:?}"),
        Err(poisoned) => {
            let mut acc = poisoned.into_inner();   // take the guard anyway
            println!("poisoned, data: {acc:?}");   // poisoned, data: [50, 0]
            acc[0] = 100;                          // repair the invariant
        }
    }
    accounts.clear_poison();                // mark it healthy again (Rust 1.77+)
    println!("{:?}", accounts.lock().unwrap());   // [100, 0]
}

In practice most code calls .lock().unwrap(), choosing to propagate the panic: if another thread crashed mid-update, continuing with possibly corrupt data is usually worse than stopping. The parking_lot crate provides a popular alternative Mutex without poisoning (and with a smaller footprint), for designs where a panic cannot leave the data inconsistent.

Key Takeaways

• Mutex<T> owns its data; the only access is through the MutexGuard returned by lock(), and dropping the guard unlocks.
• Share a mutex across threads with Arc<Mutex<T>>, or by reference in scoped threads.
• Keep critical sections short; do slow work on a copy outside the lock.
• A panic while holding the lock poisons the mutex; lock() then returns Err, from which into_inner() recovers the guard.

🧪 Practice

1. Implement a Bank with Mutex<HashMap<String, i64>> and a transfer(&self, from, to, amount) method, and run 1000 random transfers from four threads. Check that the total balance is unchanged.
2. Write a function that holds a guard in a let binding and then calls a function that locks the same mutex. Explain what happens, then fix it by narrowing the guard's scope.
3. Interview: Why does Rust's Mutex wrap the data (Mutex<T>) instead of being a standalone lock object, as in C++ or Java? (Hint: what mistakes become impossible to express?)

RwLock

A mutex serializes every access, even when all of them are reads. For data that is read constantly and written rarely (configuration, routing tables, a cache of compiled templates) this is wasteful, because concurrent reads cannot interfere with each other. It is the same rule as the borrow checker: many readers or one writer.

RwLock<T> (reader-writer lock) implements exactly that rule at run time. read() returns a RwLockReadGuard (derefs to &T), and any number of read guards may coexist. write() returns a RwLockWriteGuard (derefs to &mut T) and waits until there are no readers or writers. Picture a museum gallery: many visitors can look at the paintings at once, but when the curator rehangs them, the room is closed until the work is done.

   time ──────────────────────────────────────────────────────>
   reader 1  [=== read ===]      [== read ==]
   reader 2     [=== read ===]
   writer                  (waits)[write]          only one, no readers
   reader 3                 (waits)      [= read =]

use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use std::thread;

fn main() {
    let routes: Arc<RwLock<HashMap<&str, &str>>> =
        Arc::new(RwLock::new(HashMap::from([("/", "index"), ("/about", "about")])));

    let readers: Vec<_> = (0..4)
        .map(|i| {
            let routes = Arc::clone(&routes);
            thread::spawn(move || {
                let table = routes.read().unwrap();   // shared: readers run together
                let page = table.get("/").copied().unwrap_or("404");
                format!("reader {i}: {page}")
            })
        })
        .collect();

    {
        let mut table = routes.write().unwrap();      // exclusive
        table.insert("/blog", "blog");
    }                                                 // write guard dropped

    for r in readers {
        println!("{}", r.join().unwrap());            // reader N: index
    }
    println!("{} routes", routes.read().unwrap().len());   // 3 routes
}

RwLock is not automatically faster than Mutex. Its bookkeeping is more expensive, so for short critical sections a plain mutex often wins even under a read-heavy load. It pays off when reads are both frequent and long. Several other details matter in practice:

	Mutex<T>	RwLock<T>
Concurrent readers	No	Yes
Per-operation overhead	Lower	Higher
T bound for Sync	T: Send	T: Send + Sync (readers share &T)
Writer starvation	Not applicable	Possible, depending on the OS policy
Best for	Short or write-heavy critical sections	Long, read-mostly access

The standard RwLock delegates fairness to the operating system, so a constant stream of readers can delay a writer on some platforms. Never try to upgrade a read guard to a write guard by calling write() while holding read() on the same thread: the writer waits for all readers to leave, including you, and the thread deadlocks. For read-mostly data that changes by wholesale replacement, an alternative is to store an Arc<Config> and swap it (the arc-swap crate does this without locking readers at all).

Key Takeaways

• RwLock<T> allows many simultaneous readers or one writer, mirroring the borrow rules at run time.
• It has more overhead than Mutex; measure before assuming it is faster.
• Writer starvation is possible, and acquiring write() while holding read() on the same thread deadlocks.
• RwLock<T> requires T: Send + Sync to be shared, because readers obtain &T on several threads at once.

🧪 Practice

1. Build a Cache with RwLock<HashMap<u64, String>> whose get_or_compute method takes a read lock first, and only on a miss releases it and takes a write lock. Why must it check again after acquiring the write lock?
2. Benchmark Mutex against RwLock for eight threads doing a million reads of a small Vec, with and without a 1% write ratio.
3. Interview: Why does RwLock<T>: Sync require T: Sync while Mutex<T>: Sync only requires T: Send? (Hint: under a mutex, how many threads can see &T at once?)

Atomic Types and Ordering

A mutex is a heavyweight tool for a single integer. Locking means potentially asking the operating system to put threads to sleep and wake them. For simple shared values (counters, flags, ids) modern CPUs offer atomic instructions: operations such as "add 1 to this memory location" or "set it to X if it still equals Y" that complete as one indivisible step, even when many cores access the same location.

std::sync::atomic exposes them as types: AtomicBool, AtomicUsize, AtomicU64, AtomicI32, AtomicPtr<T>, and so on. They are interior mutable (updated through &self) and Sync, so they can be shared freely, usually in an Arc or a static.

Operation	Meaning
load(ord) / store(v, ord)	Read or write the value
fetch_add(v, ord), fetch_sub, fetch_or, ...	Update in place, return the previous value
swap(v, ord)	Store v, return the previous value
compare_exchange(cur, new, ok, fail)	Store new only if the value is still cur
fetch_update(ok, fail, f)	Retry loop applying f with compare-exchange

Every operation takes an Ordering argument, and this is where atomics get subtle. CPUs and compilers reorder memory operations for speed, as long as a single thread cannot tell the difference. Another thread can tell. Memory ordering says which reorderings are forbidden around an atomic operation:

Ordering	Guarantee	Typical use
Relaxed	The operation itself is atomic; no ordering of other memory	Counters, statistics, id generators
Release (on a store)	Writes before it cannot be moved after it	Publishing data: "it's ready"
Acquire (on a load)	Reads after it cannot be moved before it	Consuming published data
AcqRel	Both, for read-modify-write operations	Locks, reference counts on drop
SeqCst	Acquire/Release plus a single global order of all SeqCst operations	When in doubt; rarely needed

The key pattern is release/acquire pairing. If thread A writes some data and then does a Release store to a flag, and thread B does an Acquire load that sees that flag value, then B is guaranteed to see all of A's writes made before the store. It is like sealing a parcel before raising the "ready" flag: whoever sees the flag raised and then opens the parcel finds everything inside.

   Thread A (producer)                    Thread B (consumer)
   DATA.store(42, Relaxed)                while !READY.load(Acquire) {}
   READY.store(true, Release)  ───sync──>  DATA.load(Relaxed) == 42  (guaranteed)
        "everything above stays above"     "everything below stays below"

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::thread;

static NEXT_ID: AtomicUsize = AtomicUsize::new(1);   // statics need no Arc

fn new_id() -> usize {
    NEXT_ID.fetch_add(1, Ordering::Relaxed)  // only uniqueness matters: Relaxed
}

fn main() {
    // 1. A lock-free counter shared by 8 threads.
    let hits = Arc::new(AtomicU64::new(0));
    thread::scope(|s| {
        for _ in 0..8 {
            s.spawn(|| {
                for _ in 0..10_000 {
                    hits.fetch_add(1, Ordering::Relaxed);   // one indivisible step
                }
            });
        }
    });
    println!("hits = {}", hits.load(Ordering::Relaxed));   // hits = 80000

    // 2. Publishing data with Release/Acquire.
    let data = AtomicU64::new(0);
    let ready = AtomicBool::new(false);
    thread::scope(|s| {
        s.spawn(|| {
            data.store(42, Ordering::Relaxed);
            ready.store(true, Ordering::Release);   // publish: data write happens-before
        });
        s.spawn(|| {
            while !ready.load(Ordering::Acquire) {   // wait for the flag
                std::hint::spin_loop();
            }
            println!("data = {}", data.load(Ordering::Relaxed));   // data = 42, always
        });
    });

    // 3. compare_exchange: update only if nobody changed the value meanwhile.
    let max_seen = AtomicU64::new(10);
    let candidate = 25;
    let mut cur = max_seen.load(Ordering::Relaxed);
    while candidate > cur {
        let (ok, fail) = (Ordering::Relaxed, Ordering::Relaxed);   // success, failure
        match max_seen.compare_exchange(cur, candidate, ok, fail) {
            Ok(_) => break,                  // we won the race
            Err(actual) => cur = actual,     // someone else changed it: retry
        }
    }
    println!("max = {}", max_seen.load(Ordering::Relaxed));   // max = 25
    // (AtomicU64::fetch_max does exactly this in one call.)

    println!("{} {}", new_id(), new_id());  // 1 2
}

Practical rules: use Relaxed for independent counters and ids, use Release/Acquire pairs when an atomic guards other data, and use SeqCst if you cannot explain why a weaker ordering is correct. Atomics are not a replacement for a mutex when an update involves several values: two separately atomic fields can still be observed in an inconsistent combination. Note also that x.load() followed by x.store(x + 1) is two atomic operations, not one, and loses updates exactly like the unsynchronized version.

Key Takeaways

• Atomic types provide indivisible operations on single values, updated through &self without locks.
• Read-modify-write methods (fetch_add, compare_exchange, fetch_update) are atomic as a whole; a separate load plus store is not.
• Ordering controls how surrounding memory operations may be reordered: Relaxed for counters, Release/Acquire for publishing data, SeqCst when unsure.
• Use a mutex when an invariant spans more than one value.

🧪 Practice

1. Implement a SpinLock with an AtomicBool, using compare_exchange with Acquire to lock and store(false, Release) to unlock, and use it to guard a counter (you will need UnsafeCell from Chapter 9).
2. Count words across eight scoped threads using one AtomicUsize, then with a local count per thread added once at the end. Compare the timings.
3. Interview: Explain why a Relaxed store of ready = true would not be enough to guarantee that a reader sees data = 42, even on a CPU like x86 that rarely reorders. (Hint: the compiler is allowed to reorder too.)

Condvar and Barriers

Locks protect data, but they do not help a thread wait for a condition: "wait until the queue is non-empty", "wait until all workers finished phase one". Polling in a loop (loop { if lock().ready { break } }) wastes a CPU core, and adding sleep makes the reaction slow. Two primitives let threads wait efficiently.

Condvar (condition variable) lets a thread sleep until another thread signals that something it cares about may have changed. It always works together with a Mutex that protects the condition:

1. Lock the mutex and check the condition.
2. If it is not yet true, call condvar.wait(guard). This atomically releases the lock and puts the thread to sleep, and re-acquires the lock before returning.
3. The other thread changes the state under the lock and calls notify_one() or notify_all().

Because wake-ups can be spurious (a thread may wake without a notification), the condition must always be re-checked in a loop. wait_while(guard, |s| condition) does that loop for you. Think of a waiting room with a buzzer: you sit down until the buzzer sounds, but when it does, you still check the board to see whether your number was called.

Barrier is a meeting point for a fixed number of threads. Each thread calls barrier.wait() and blocks until all n threads have arrived, then all continue together. It is used for phased algorithms where every thread must finish step k before anyone starts step k + 1. Exactly one thread receives is_leader() == true from each round, which is handy for one-off work such as printing a progress line.

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

fn main() {
    // --- Condvar: a blocking job queue ---
    let shared = Arc::new((Mutex::new(VecDeque::<u32>::new()), Condvar::new()));

    let consumer = {
        let shared = Arc::clone(&shared);
        thread::spawn(move || {
            let (queue, ready) = &*shared;
            let mut handled = Vec::new();
            while handled.len() < 3 {
                // Sleep while the queue is empty; re-checks after every wake-up.
                let guard = queue.lock().unwrap();
                let mut q = ready.wait_while(guard, |q| q.is_empty()).unwrap();
                handled.push(q.pop_front().unwrap());
            }
            handled
        })
    };

    for job in [10, 20, 30] {
        let (queue, ready) = &*shared;
        queue.lock().unwrap().push_back(job);   // change state under the lock
        ready.notify_one();                     // wake one waiting thread
    }
    println!("{:?}", consumer.join().unwrap());   // [10, 20, 30]

    // --- Barrier: phased computation ---
    let n = 3;
    let barrier = Barrier::new(n);
    let partial = Mutex::new(Vec::new());
    thread::scope(|s| {
        for id in 0..n {
            let (barrier, partial) = (&barrier, &partial);
            s.spawn(move || {
                partial.lock().unwrap().push(id * 10);   // phase 1
                if barrier.wait().is_leader() {          // everyone waits here
                    println!("phase 1 complete");        // printed exactly once
                }
                let all = partial.lock().unwrap().len(); // phase 2 sees all results
                assert_eq!(all, n);
            });
        }
    });
    let mut results = partial.into_inner().unwrap();
    results.sort();                         // push order depends on scheduling
    println!("{results:?}");                // [0, 10, 20]
}

Primitive	Waits for	Typical use
Condvar	A condition on shared state to become true	Blocking queues, "work available", state machines
Barrier	A fixed number of threads to arrive	Phased simulations, parallel iterations in lockstep
JoinHandle::join	One thread to finish	Collecting results
Channels	A message	Most producer-consumer designs

Condition variables are a low-level tool. If what you are waiting for is "a value arrives", a channel is usually simpler and harder to misuse, and it is built on these same mechanisms. Reach for Condvar when the condition is a predicate over shared state that channels do not model well.

Key Takeaways

• Condvar lets a thread sleep until notified, releasing its mutex while it waits and re-acquiring it before returning.
• Always re-check the condition after waking (spurious wake-ups); use wait_while to get the loop right.
• Change the state under the lock, then call notify_one or notify_all.
• Barrier blocks a fixed number of threads until all arrive; one of them is the leader of each round.

🧪 Practice

1. Use a Mutex<bool> and a Condvar to make the main thread wait until a worker thread signals "initialized", with a timeout of one second using wait_timeout_while.
2. Simulate three rounds of a game with four player threads that must all submit a move (into a shared Vec) before the next round starts, using a Barrier.
3. Interview: Why must Condvar::wait take the MutexGuard and release the lock atomically, instead of letting the caller unlock and then wait? (Hint: what happens if the notification arrives between the two steps?)

Deadlock Avoidance

Locks remove data races but introduce a new failure: deadlock, where two or more threads each hold a lock the other needs, and all wait forever. The program does not crash. It simply stops making progress, often only under load and often only in production.

The classic case is two locks acquired in opposite orders:

   Thread 1                        Thread 2
   lock(A)  (held)                 lock(B)  (held)
   lock(B)  ... waits for T2       lock(A)  ... waits for T1
            └──────────── circular wait: nobody proceeds ────────────┘

A deadlock requires four conditions at once (the Coffman conditions): mutual exclusion, hold-and-wait (holding one lock while waiting for another), no preemption (locks cannot be taken away), and a circular wait. Breaking any one prevents deadlock, and in practice the easiest to break are hold-and-wait and circular wait. Picture four cars at a four-way stop, each waiting for the car on its right: one agreed rule ("lowest street number goes first") dissolves the jam.

Rust prevents data races but not deadlocks: they are a logic error, not a memory-safety violation. Practical defenses:

Technique	How it helps
Global lock ordering	Always acquire locks in one agreed order (for example by id or address), so no cycle can form
One lock at a time	Copy what you need, release, then take the next lock
Short critical sections	Never call unknown code (callbacks, trait objects) or block on I/O while holding a lock
Coarser locks	Put related data under one mutex instead of several
try_lock with back-off	Give up and retry instead of waiting forever
Message passing	No locks to hold at all (section 10.2)

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

struct Account {
    id: u32,
    balance: Mutex<i64>,
}

// Always lock the lower id first: both directions use the same order.
fn transfer(from: &Account, to: &Account, amount: i64) {
    if from.id == to.id {
        return;                             // locking the same mutex twice would deadlock
    }
    let (first, second) = if from.id < to.id { (from, to) } else { (to, from) };
    let mut a = first.balance.lock().unwrap();
    let mut b = second.balance.lock().unwrap();
    // Map the guards back to "from" and "to".
    let (from_bal, to_bal) = if from.id < to.id {
        (&mut *a, &mut *b)
    } else {
        (&mut *b, &mut *a)
    };
    *from_bal -= amount;
    *to_bal += amount;
}                                           // guards dropped in reverse order

fn main() {
    let x = Arc::new(Account { id: 1, balance: Mutex::new(1_000) });
    let y = Arc::new(Account { id: 2, balance: Mutex::new(1_000) });

    // Opposite directions concurrently: deadlock-prone with naive ordering.
    let t1 = {
        let (x, y) = (Arc::clone(&x), Arc::clone(&y));
        thread::spawn(move || for _ in 0..10_000 { transfer(&x, &y, 1) })
    };
    let t2 = {
        let (x, y) = (Arc::clone(&x), Arc::clone(&y));
        thread::spawn(move || for _ in 0..10_000 { transfer(&y, &x, 1) })
    };
    t1.join().unwrap();
    t2.join().unwrap();
    let total = *x.balance.lock().unwrap() + *y.balance.lock().unwrap();
    println!("total = {total}");            // total = 2000, and the program finishes

    // try_lock: fail fast instead of blocking.
    let m = Mutex::new(0);
    let _held = m.lock().unwrap();
    println!("{}", m.try_lock().is_err());  // true: WouldBlock instead of deadlock
}

Rust has a few deadlock traps of its own. A MutexGuard lives until the end of its scope, so a guard bound with let at the top of a long function holds the lock for the whole function. A temporary guard in a match scrutinee lives until the end of the whole match, so match m.lock().unwrap().get(..) { .. } keeps the lock during every arm. In an if let it lives through the body (the 2024 edition only releases it before the else branch), and in a while let it lives through each iteration of the loop body. Finally, std::sync::Mutex is not re-entrant: locking it twice on the same thread deadlocks or panics. When a program hangs, a debugger's "all threads backtrace" (thread apply all bt in gdb) shows which locks each thread is waiting for.

Key Takeaways

• A deadlock is a circular wait between threads holding locks; Rust prevents data races but not deadlocks.
• Acquire multiple locks in one global order, or avoid holding more than one.
• Never hold a lock while calling unknown code, doing I/O, or waiting on a channel; drop guards as early as possible.
• std::sync::Mutex is not re-entrant; locking it twice on one thread hangs or panics.

🧪 Practice

1. Remove the ordering from transfer (lock from, then to) and run the program repeatedly until it hangs. Attach a debugger and identify the two waiting threads.
2. Rewrite transfer to use try_lock on the second account with a short back-off and retry. Discuss whether it can livelock.
3. Interview: The dining philosophers problem has five philosophers and five forks. Describe two different strategies that prevent deadlock and which Coffman condition each one breaks. (Hint: ordering the forks, or limiting how many philosophers sit down.)

<a id="104-concurrency-safety"></a>

10.4 Concurrency Safety

The previous sections relied on the compiler rejecting unsafe uses, such as an Rc sent to another thread or a local borrowed by thread::spawn. This section explains the machinery behind those errors: the Send and Sync marker traits introduced in Chapter 5, how they are derived automatically, and exactly what Rust does and does not guarantee about concurrent code.

The Send Trait

Some values are tied to the thread that created them. An Rc updates its reference count without synchronization, so two threads holding clones of the same Rc could corrupt the count. A MutexGuard from some operating-system lock implementations must be released by the same thread that acquired it. A raw handle from a C library may only be valid on the thread that opened it. Moving such a value to another thread is a bug, but nothing about the value's type in most languages says so.

Send is the marker trait that says it: T: Send means ownership of a T can be transferred to another thread safely. It has no methods. It is a compile-time label that APIs which move values across threads require: thread::spawn requires its closure (and therefore everything captured) to be Send, and mpsc::Sender<T>::send requires T: Send. Think of it as a customs declaration on a parcel: the border (thread boundary) only lets through items that are declared safe to export.

Almost every type is Send. The exceptions are types that would misbehave if used from a different thread than the one that created them, or that share unsynchronized state with values that stay behind:

Type	Send?	Reason
i32, String, Vec<T: Send>, Box<T: Send>	Yes	Owned data, no hidden sharing
Arc<T>	If T: Send + Sync	Clones on other threads share T
Mutex<T>	If T: Send	Owns T; the lock is thread-safe
Rc<T>	No	Non-atomic count shared with clones left behind
MutexGuard<'_, T>	No	Must unlock on the thread that locked
*const T, *mut T	No	The compiler cannot know what they point to
Cell<T>, RefCell<T>	If T: Send	Moving a cell is fine; sharing it is not (Sync)

use std::cell::RefCell;
use std::rc::Rc;
use std::sync::{Arc, Mutex};
use std::thread;

// A zero-cost compile-time check: this compiles only if T is Send.
fn assert_send<T: Send>() {}

fn main() {
    assert_send::<String>();
    assert_send::<Arc<Mutex<Vec<i32>>>>();
    assert_send::<RefCell<i32>>();          // a cell can be moved, just not shared
    // assert_send::<Rc<i32>>();            // error[E0277]: `Rc<i32>` cannot be sent
                                            // between threads safely

    // Moving an owned, Send value: fine.
    let v = vec![1, 2, 3];
    let h = thread::spawn(move || v.len());
    println!("{}", h.join().unwrap());      // 3

    // The fix for Rc is Arc: same API, atomic counts.
    let shared = Arc::new(5);
    let s2 = Arc::clone(&shared);
    thread::spawn(move || println!("{s2}")).join().unwrap();   // 5

    // A non-Send value can still be *created* inside the thread.
    let h = thread::spawn(|| {
        let local = Rc::new(String::from("built here"));   // never crosses a boundary
        local.len()                                         // usize is Send
    });
    println!("{}", h.join().unwrap());      // 10
}

The last example shows a useful trick: Send only restricts values that cross the boundary. A thread can freely create and use Rc, RefCell, or other non-Send types internally, as long as only Send values go in or come out.

Key Takeaways

• T: Send means a T can be moved to another thread; APIs like thread::spawn and Sender::send require it.
• Most types are Send; Rc, raw pointers, and MutexGuard are the classic exceptions.
• Non-Send types can be used freely inside one thread; only values crossing a thread boundary must be Send.
• fn assert_send<T: Send>() {} is a zero-cost way to test a type in code.

🧪 Practice

1. Use assert_send to check Box<dyn Fn()>, Box<dyn Fn() + Send>, Vec<Rc<i32>>, and Option<Arc<str>>. Predict each result before compiling.
2. Write a struct containing an Rc<String>, try to send it through a channel, read the error, and fix it in two different ways.
3. Interview: Why is MutexGuard not Send even though Mutex itself can be shared between threads? (Hint: on some platforms a lock must be released by the thread that acquired it.)

The Sync Trait

Send is about moving a value to another thread. Sync is about sharing: T: Sync means it is safe for several threads to hold &T at the same time. The formal definition ties the two together: T is Sync exactly when &T is Send. If you can hand a shared reference to another thread, the type is Sync.

For most types the question is easy, because shared references only allow reading, and concurrent reads are harmless. The interesting cases are types with interior mutability (Chapter 9), which allow writes through &T:

• Cell<T> and RefCell<T> mutate through &self without synchronization, so two threads with &Cell<T> could write at the same time. They are !Sync.
• Mutex<T>, RwLock<T>, and the atomic types also mutate through &self, but they synchronize, so they are Sync. That is the whole point of them.
• Rc<T> is !Sync, because cloning through &Rc<T> updates the count without synchronization.

A library book is the analogy: a reference book chained to a reading-room table (Sync) can be read by many people at once, while a notebook that anyone may scribble in (Cell) must not be passed around a crowded room.

   T: Sync  <=>  &T: Send

                Send    Sync    Why
   i32           yes     yes    immutable through &T
   Cell<i32>     yes     no     writes through &T with no synchronization
   Mutex<i32>    yes     yes    writes through &T are locked
   AtomicU64     yes     yes    writes through &T are atomic
   Rc<i32>       no      no     non-atomic count, mutated via clone(&self)
   Arc<i32>      yes     yes    atomic count
   MutexGuard    no      yes*   (* if T: Sync) must unlock on its own thread

Sync explains the bounds on the shared-state types you have already used:

• Arc<T> is Send and Sync only if T: Send + Sync, because every clone on every thread can produce &T (needs Sync), and whichever thread drops the last clone drops the T (needs Send).
• Mutex<T> is Sync if T: Send. The lock ensures only one thread reaches the T at a time, so T itself never needs to be shared, only passed along.
• RwLock<T> is Sync only if T: Send + Sync, because readers on several threads hold &T simultaneously.

use std::cell::Cell;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU32, Ordering};
use std::thread;

fn assert_sync<T: Sync>() {}

fn main() {
    assert_sync::<Vec<String>>();
    assert_sync::<Mutex<Cell<u32>>>();      // Mutex makes a !Sync Cell shareable
    // assert_sync::<Cell<u32>>();          // error[E0277]: `Cell<u32>` cannot be
                                            // shared between threads safely

    let plain = Cell::new(0);
    let atomic = AtomicU32::new(0);
    thread::scope(|s| {
        // Scoped threads borrow: the captured `&AtomicU32` must be Send,
        // which requires AtomicU32: Sync.
        s.spawn(|| atomic.fetch_add(1, Ordering::Relaxed));
        s.spawn(|| atomic.fetch_add(1, Ordering::Relaxed));
        // s.spawn(|| plain.set(1));        // error[E0277]: `Cell<i32>` cannot be
                                            // shared between threads safely
    });
    plain.set(plain.get() + 1);             // fine on a single thread
    println!("{} {}", atomic.into_inner(), plain.get());   // 2 1
}

Key Takeaways

• T: Sync means &T can be shared across threads; formally, T: Sync if and only if &T: Send.
• Interior mutability without synchronization (Cell, RefCell, Rc) is !Sync; synchronized interior mutability (Mutex, RwLock, atomics) is Sync.
• Arc<T> needs T: Send + Sync; Mutex<T> needs only T: Send; RwLock<T> needs T: Send + Sync.
• Wrapping a !Sync type in a Mutex is the standard way to share it.

🧪 Practice

1. For each of Arc<Cell<i32>>, Arc<Mutex<Rc<i32>>>, and Arc<RwLock<Cell<i32>>>, decide whether it can be sent to another thread, then check with assert_send. Explain each answer from the rules above.
2. Share a &Vec<u64> (read-only) among four scoped threads that each sum a quarter of it. Which trait made the borrow legal?
3. Interview: Can a type be Sync but not Send? Give an example and explain what it means. (Hint: look at MutexGuard<T> when T: Sync.)

Auto Traits and Negative Impls

You almost never write impl Send for MyType. Send and Sync are auto traits: the compiler implements them for your type automatically if, and only if, every field implements them. A struct of a String and a Vec<u64> is Send + Sync without a line of code. Add one Rc field and the whole struct silently becomes !Send + !Sync. The properties are inferred structurally, the same way that an assembled product is "food safe" only if every component is.

The standard auto traits are Send, Sync, Unpin (Chapter 9), UnwindSafe, and RefUnwindSafe. You cannot define new auto traits on stable Rust.

Sometimes the structural answer is wrong, in one of two directions:

• Too permissive: a type contains only Send fields (say, a usize handle) but represents something thread-bound, such as an index into a thread-local table. You need to opt out.
• Too strict: a type contains a raw pointer (which is !Send), but you know the pointed-to data is owned and safely movable. You need to opt in.

Opting out on stable Rust is done by adding a zero-sized marker field of a type that lacks the trait. Negative impls (impl !Send for T {}) express the same intent directly but are still a nightly-only feature (#![feature(negative_impls)]), used inside the standard library itself.

Opting in requires unsafe impl Send for T {}. It is unsafe because you are making a promise the compiler cannot check: if it is false, safe code can cause data races.

use std::marker::PhantomData;
use std::thread;

// Opt OUT of Send and Sync: *const () is neither, and PhantomData costs 0 bytes.
struct ThreadBound {
    slot: usize,                            // index into a thread-local table
    _not_send: PhantomData<*const ()>,
}

// Opt out of Sync only: Cell<()> is Send but not Sync.
struct SendNotSync {
    _marker: PhantomData<std::cell::Cell<()>>,
}

// Opt IN: a buffer owned through a raw pointer from a C-style allocator.
struct RawBuffer {
    ptr: *mut u8,                           // makes the struct !Send by default
    len: usize,
}

// SAFETY: RawBuffer uniquely owns the allocation behind `ptr` (no aliases are
// kept elsewhere), so moving it to another thread moves sole ownership.
unsafe impl Send for RawBuffer {}

impl RawBuffer {
    fn new(len: usize) -> Self {
        let boxed = vec![0u8; len].into_boxed_slice();
        RawBuffer { ptr: Box::into_raw(boxed) as *mut u8, len }
    }
}

impl Drop for RawBuffer {
    fn drop(&mut self) {
        // SAFETY: ptr/len come from Box<[u8]> in `new` and are freed once.
        let slice = std::ptr::slice_from_raw_parts_mut(self.ptr, self.len);
        unsafe { drop(Box::from_raw(slice)) };
    }
}

fn assert_send<T: Send>() {}

fn main() {
    let tb = ThreadBound { slot: 3, _not_send: PhantomData };
    // thread::spawn(move || drop(tb));     // error[E0277]: `*const ()` cannot be
                                            // sent between threads safely
    // Note: `move || tb.slot` WOULD compile, because since the 2021 edition a
    // closure captures only the fields it uses, here a plain usize.
    println!("slot {}", tb.slot);
    let _ = SendNotSync { _marker: PhantomData };
    assert_send::<SendNotSync>();

    let buf = RawBuffer::new(16);
    let h = thread::spawn(move || buf.len); // allowed by the unsafe impl
    println!("len {}", h.join().unwrap());  // len 16
}

On nightly, the opt-out above can be written directly:

#![feature(negative_impls)]

struct ThreadBound {
    slot: usize,
}

impl !Send for ThreadBound {}               // explicit negative impl
impl !Sync for ThreadBound {}

Auto traits are also part of your public API. If a library type is Send today because of its private fields, adding an Rc field later silently makes it !Send and breaks downstream code that sends it across threads, which is a semver-breaking change (Chapter 8). Library authors often add compile-time assertions such as const _: () = { fn f<T: Send + Sync>() {} let _ = f::<MyType>; }; to catch this.

Key Takeaways

• Send and Sync are auto traits: a type has them exactly when all of its fields do.
• Opt out on stable with a PhantomData marker field of a !Send/!Sync type; negative impls (impl !Send) are nightly-only.
• Opt in with unsafe impl Send/Sync, documenting the safety argument; a wrong promise permits data races in safe code.
• Auto-trait implementations are part of a type's public API, so changing fields can be a breaking change.

🧪 Practice

1. Write a struct Handle { id: u32 } and make it !Send but still Sync using a marker field. Verify both properties with assertion functions.
2. Add a compile-time Send + Sync assertion for a struct, then add an Rc field and read the resulting error.
3. Interview: Why must unsafe impl Sync be written for a custom spin lock wrapping UnsafeCell<T>, and what condition on T should the impl carry? (Hint: compare with unsafe impl<T: Send> Sync for Mutex<T> in the standard library.)

Fearless Concurrency Guarantees

"Fearless concurrency" is the Rust project's name for a specific claim: the compiler rejects programs that would contain data races, so concurrent code can be written and refactored with the same confidence as sequential code. The claim is precise, and it helps to know exactly where it ends, because Rust does not promise that concurrent programs are correct.

The guarantee comes from three pieces working together:

1. Ownership and borrowing: at any time, a value has either one &mut or any number of & references, never both.
2. Send and Sync: those references, and owned values, can only cross thread boundaries when the type says it is safe.
3. Synchronized interior mutability: the only way to mutate shared data is through types such as Mutex and atomics whose APIs enforce synchronization.

Bugs that other languages find with testing, code review, or sanitizers become compile errors:

use std::rc::Rc;
use std::thread;

fn main() {
    let mut list = vec![1, 2, 3];

    // 1. Mutating shared data from two threads without synchronization
    // thread::scope(|s| {
    //     s.spawn(|| list.push(4));
    //     s.spawn(|| list.push(5));
    // });
    // error[E0499]: cannot borrow `list` as mutable more than once at a time

    // 2. Reading while another thread writes
    // thread::scope(|s| {
    //     s.spawn(|| list.push(4));
    //     println!("{}", list.len());
    // });
    // error[E0502]: cannot borrow `list` as immutable because it is also
    //               borrowed as mutable

    // 3. Using thread-unsafe reference counting across threads
    // let rc = Rc::new(1);
    // thread::spawn(move || println!("{rc}"));
    // error[E0277]: `Rc<i32>` cannot be sent between threads safely

    // 4. A thread outliving the data it borrows
    // thread::spawn(|| list.push(6));
    // error[E0373]: closure may outlive the current function, but it borrows `list`

    list.push(4);
    let _ = Rc::new(0);
    thread::spawn(|| ()).join().unwrap();
    println!("{list:?}");                   // [1, 2, 3, 4]
}

What is and is not covered:

Class of bug	Prevented at compile time?
Data races (unsynchronized concurrent access with a write)	Yes
Use-after-free across threads (dangling borrows)	Yes
Sending thread-bound values to other threads	Yes
Forgetting to lock before accessing protected data	Yes (Mutex<T> owns T)
Deadlocks and livelocks	No
Race conditions in program logic (check-then-act)	No
Starvation and unfair scheduling	No
Leaks (for example Arc cycles, forgotten threads)	No
Lost wake-ups, wrong memory Ordering on atomics	Partially (safe, but may be wrong)

All of these guarantees apply to safe code. Inside an unsafe block, or behind an incorrect unsafe impl Send, the programmer takes over the responsibility. The guarantee also holds across library boundaries: a crate cannot accidentally expose an API that lets your safe code cause a data race, unless its own unsafe code is wrong.

The practical effect is a change in where effort goes. Instead of hunting for memory corruption that shows up once a week in production, you spend it on the remaining logical problems: lock ordering, protocol design, and testing interleavings (tools such as loom explore thread schedules systematically, and Chapter 12 covers testing).

Key Takeaways

• Rust's "fearless concurrency" guarantee is the absence of data races in safe code, enforced by ownership, borrowing, Send, and Sync.
• Unsynchronized shared mutation, dangling cross-thread borrows, and sending thread-bound types are all compile errors.
• Deadlocks, logic race conditions, starvation, and leaks are not prevented.
• The guarantee relies on unsafe code (including unsafe impl Send/Sync) being correct.

🧪 Practice

1. Uncomment each of the four error cases in turn, read the compiler message, and fix each one with the most appropriate tool (scoped threads, Mutex, Arc, or move).
2. Write a program that compiles without warnings but deadlocks, to convince yourself that the guarantee has limits.
3. Interview: A colleague says "Rust code can't have concurrency bugs". How would you correct the statement precisely? (Hint: distinguish memory safety from logical correctness, and list one example of each kind of bug Rust allows.)

Data Race Freedom

A data race has a precise definition. It occurs when:

1. two or more threads access the same memory location concurrently,
2. at least one of the accesses is a write, and
3. the accesses are not synchronized (no lock, no atomic operation, no happens-before relationship between them).

In C and C++ a data race is undefined behavior: the compiler assumes it never happens and optimizes accordingly, so the result is not merely "a wrong value" but potentially torn reads (half of an old 64-bit value and half of a new one), values that never existed, infinite loops, or corrupted heap metadata. That is why Rust treats preventing data races as part of memory safety.

Notice how the definition lines up with Rust's rules: condition 2 (a write while others access) is exactly what "one &mut or many &" forbids, and condition 3 is what Send/Sync enforce for interior mutability. A data race needs aliasing plus mutation without synchronization, and safe Rust never gives you all three at once.

A race condition is a different and broader thing: the program's result depends on the timing of threads, even though every individual access is properly synchronized. Rust does not prevent race conditions. An example with atomics, which cannot data race:

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;
use std::thread;

fn main() {
    // RACE CONDITION (but no data race): load and store are each atomic, yet
    // another thread can update the value in between them.
    let racy = AtomicU64::new(0);
    thread::scope(|s| {
        for _ in 0..8 {
            s.spawn(|| {
                for _ in 0..100_000 {
                    let v = racy.load(Ordering::Relaxed);    // read
                    racy.store(v + 1, Ordering::Relaxed);    // write (may overwrite others)
                }
            });
        }
    });
    println!("racy:   {}", racy.load(Ordering::Relaxed));   // usually far below 800000

    // Fix 1: make the read-modify-write a single atomic operation.
    let atomic = AtomicU64::new(0);
    thread::scope(|s| {
        for _ in 0..8 {
            s.spawn(|| {
                for _ in 0..100_000 {
                    atomic.fetch_add(1, Ordering::Relaxed);
                }
            });
        }
    });
    println!("atomic: {}", atomic.load(Ordering::Relaxed));  // 800000

    // Fix 2: check-then-act under one lock, so no thread can interleave.
    let seats = Mutex::new(3u32);
    let booked = AtomicU64::new(0);
    thread::scope(|s| {
        for _ in 0..10 {
            s.spawn(|| {
                let mut free = seats.lock().unwrap();   // check AND act while locked
                if *free > 0 {
                    *free -= 1;
                    booked.fetch_add(1, Ordering::Relaxed);
                }
            });
        }
    });
    println!("booked: {}", booked.load(Ordering::Relaxed));  // 3, never more
}

	Data race	Race condition
Definition	Unsynchronized concurrent access with a write	Result depends on thread timing
Consequence	Undefined behavior (C/C++); impossible in safe Rust	Wrong but well-defined results
Example	Two threads doing *p += 1 through raw pointers	Atomic load then store, or check-then-act
Prevented by Rust	Yes, in safe code	No
Fix	Synchronization (locks, atomics)	Make the whole logical operation atomic

When unsafe code is involved, tools can find data races at run time: Miri (Chapter 14) detects them in interpreted test runs, and the ThreadSanitizer (RUSTFLAGS="-Zsanitizer=thread" on nightly) instruments compiled code to catch them.

Key Takeaways

• A data race is two unsynchronized concurrent accesses to the same location, at least one a write; in C/C++ it is undefined behavior.
• Safe Rust is data-race-free: aliasing XOR mutability plus Send/Sync rule out every case.
• Race conditions (timing-dependent logic, such as check-then-act) remain possible even with atomics and locks.
• Fix race conditions by making the whole logical operation atomic: a single read-modify-write instruction, or one critical section.

🧪 Practice

1. Run the racy counter several times and record the results. Then explain why every result is still a value some thread actually wrote, and not garbage.
2. Write a "withdraw if balance is sufficient" function using two separate lock() calls (one to check, one to subtract) and show with many threads that the balance can go negative. Fix it.
3. Interview: Is a program that uses only atomics with Ordering::Relaxed free of data races? Is it free of race conditions? (Hint: apply the three-part definition, then think about invariants that span two atomics.)

<a id="105-data-parallelism"></a>

10.5 Data Parallelism

Data parallelism applies the same operation to many pieces of data at once: transforming every pixel, hashing every file, summing every row. This section covers Rayon, the de facto standard library for it, the work-stealing scheduler behind it, parallel sorting and reduction, thread pools in general, and how to measure whether parallelism actually helped.

Rayon Parallel Iterators

Manually splitting work across threads is repetitive and easy to get wrong: you must choose chunk sizes, spawn threads, collect results in the right order, and balance the load when some items are slower than others. Most of the time the work has a simple shape: "do this to every element". Iterators (Chapter 6) already describe that shape for sequential code.

Rayon (rayon crate) turns iterator pipelines into parallel ones with a one-word change: iter() becomes par_iter(). Rayon splits the input, schedules the pieces on a global pool of worker threads (one per CPU core by default), and combines the results. Because closures passed to it must be Send + Sync where they are shared, the same compile-time checks from section 10.4 apply: if a parallel pipeline compiles, it has no data races.

Think of a stack of exam papers to grade. Sequentially, one teacher grades them all. With Rayon, the stack is split among all available teachers, who pick up more papers whenever they finish, and the grades are collected in the original order at the end.

Sequential	Parallel (Rayon)	Item type
v.iter()	v.par_iter()	&T
v.iter_mut()	v.par_iter_mut()	&mut T
v.into_iter()	v.into_par_iter()	T
(0..n)	(0..n).into_par_iter()	integers
v.chunks(k)	v.par_chunks(k)	&[T]
any Iterator	iter.par_bridge()	unordered, from a sequential source

Most adapters you know have parallel counterparts: map, filter, filter_map, flat_map, for_each, sum, min_by_key, any, all, find_any, count, collect. Two behaviors differ from sequential iterators:

• Order of side effects is unspecified. for_each(|x| println!(..)) prints in any order. Results from collect on indexed sources (Vec, slices, ranges) are in the original order.
• find_any replaces find when any match will do; find_first exists but has to wait for earlier items.

// Cargo.toml: rayon = "1"
use rayon::prelude::*;                      // brings par_iter() and friends into scope

fn is_prime(n: u64) -> bool {
    n >= 2 && (2..).take_while(|d| d * d <= n).all(|d| n % d != 0)
}

fn main() {
    let numbers: Vec<u64> = (1..=2_000_000).collect();

    // Same pipeline as the sequential version, with par_iter().
    let primes = numbers.par_iter().filter(|&&n| is_prime(n)).count();
    println!("{primes} primes");            // 148933 primes

    // collect keeps the original order for indexed sources.
    let squares: Vec<u64> = (1..=5u64).into_par_iter().map(|x| x * x).collect();
    println!("{squares:?}");                // [1, 4, 9, 16, 25]

    // In-place parallel mutation: each element is touched by exactly one thread.
    let mut pixels = vec![100u8; 1_000_000];
    pixels.par_iter_mut().for_each(|p| *p = p.saturating_add(50));
    println!("{}", pixels[123_456]);        // 150

    // Chunks: per-chunk work with less per-item overhead.
    let checksums: Vec<u32> = numbers
        .par_chunks(100_000)
        .map(|c| c.iter().fold(0u32, |acc, &x| acc.wrapping_add(x as u32)))
        .collect();
    println!("{} chunks", checksums.len()); // 20 chunks

    // Short-circuiting: stops other workers once a match is found.
    let big = numbers.par_iter().find_any(|&&n| n > 1_999_990 && is_prime(n));
    println!("{big:?}");                    // Some(1999993) (the only one in range)

    // Captured state must be thread-safe; this is checked at compile time.
    // let mut total = 0;
    // numbers.par_iter().for_each(|n| total += n);
    // error[E0596]: cannot borrow `total` as mutable, as it is a captured
    //               variable in a `Fn` closure
}

Rayon is for CPU-bound work. Blocking I/O inside a parallel iterator ties up pool threads that could be computing, and async code (Chapter 11) has its own executors. Also, parallelism has overhead: for a few thousand cheap operations, the sequential version is often faster (see "Measuring Parallel Speedup").

Key Takeaways

• Rayon turns iterator pipelines parallel by swapping iter() for par_iter() (or into_par_iter, par_iter_mut, par_chunks).
• The Send/Sync bounds on its closures mean a compiling Rayon pipeline is free of data races.
• collect preserves order for indexed sources; side effects such as printing happen in any order.
• Use it for CPU-bound work with enough items to amortize the overhead.

🧪 Practice

1. Compute the number of words in each of 10,000 generated strings in parallel and collect the counts into a Vec<usize> in input order.
2. Parallelize a function that converts an RGB image (Vec<[u8; 3]>) to grayscale in place with par_iter_mut.
3. Interview: Why does numbers.par_iter().for_each(|n| total += n) fail to compile while the sequential version compiles? (Hint: which closure trait does for_each require in Rayon, and why must the closure be shareable between threads?)

Work Stealing

Splitting work evenly sounds easy until items differ in cost. If eight threads each get one eighth of an array, but one eighth contains the expensive items, seven threads finish early and sit idle while the eighth works alone. Static splitting also cannot adapt to a machine that is busy with other programs.

Rayon uses work stealing to balance load dynamically. Each worker thread has its own double-ended queue (deque) of tasks. A worker pushes new tasks onto the bottom of its own deque and pops from the bottom (most recent first, which keeps data warm in its cache). When a worker runs out of tasks, it picks another worker at random and steals from the top of that worker's deque, taking the oldest and therefore usually largest piece of work.

The analogy is a team of cooks, each with their own order rail. Cooks work on their own orders, but a cook with an empty rail walks over and takes the oldest ticket from a busy colleague's rail.

   Worker 1 deque         Worker 2 deque          Worker 3 deque (empty)
   top    [ 0..500K ]  <────────────── steal ───────────  idle: steals oldest,
          [ 500K..750K]                                    largest task from W1
   bottom [ 750K..1M ]  ← W1 pops own work here

The core primitive is rayon::join(a, b). It runs closure a on the current thread and pushes b onto the current worker's deque, where an idle worker may steal it. If nobody steals b before a finishes, the current thread runs b itself, so an unstolen join costs little more than two function calls. Parallel iterators are built by recursively splitting the input in halves with join until the pieces are small, which gives work stealing a supply of tasks of every size.

// Cargo.toml: rayon = "1"
use rayon::join;

// Divide and conquer: sum a slice by recursively splitting it.
fn par_sum(v: &[u64]) -> u64 {
    if v.len() <= 10_000 {
        return v.iter().sum();              // small enough: do it sequentially
    }
    let (left, right) = v.split_at(v.len() / 2);
    let (a, b) = join(|| par_sum(left), || par_sum(right));   // b may be stolen
    a + b
}

// Uneven work: cost grows with the index, so static splitting would be unbalanced.
fn work(i: u64) -> u64 {
    (0..i % 5_000).fold(i, |acc, x| acc.wrapping_mul(31).wrapping_add(x))
}

fn main() {
    let v: Vec<u64> = (1..=10_000_000).collect();
    println!("{}", par_sum(&v));            // 50000005000000

    use rayon::prelude::*;
    let par = (0..200_000u64).into_par_iter().map(work).reduce(|| 0, u64::wrapping_add);
    let seq = (0..200_000u64).map(work).fold(0, u64::wrapping_add);
    println!("{}", par == seq);             // true: same result, balanced by stealing

    // join returns both results; the closures may borrow local data.
    let text = "the quick brown fox jumps over the lazy dog";
    let (words, vowels) = join(
        || text.split_whitespace().count(),
        || text.chars().filter(|c| "aeiou".contains(*c)).count(),
    );
    println!("{words} words, {vowels} vowels");   // 9 words, 11 vowels
}

Work stealing gives Rayon two important properties: load balancing is automatic even for irregular workloads, and nested parallelism composes. A parallel iterator whose closure itself uses par_iter simply adds more tasks to the deques instead of oversubscribing the machine with extra threads.

Key Takeaways

• Each Rayon worker has its own deque; idle workers steal the oldest (largest) task from a random busy worker.
• rayon::join(a, b) makes b available for stealing while running a; if not stolen, the same thread runs b cheaply.
• Recursive splitting plus stealing balances irregular workloads automatically and lets parallel code nest safely.
• Stop splitting below a threshold, where sequential work is cheaper than task overhead.

🧪 Practice

1. Implement parallel merge sort with rayon::join, falling back to sort_unstable below 4,096 elements. Compare with slice::sort.
2. Write a recursive parallel Fibonacci with join and measure how the sequential threshold affects run time.
3. Interview: Why does a worker take tasks from the bottom of its own deque but steal from the top of others? (Hint: think about cache locality and which tasks are likely to be large.)

Parallel Sorting and Reduction

Two operations dominate data processing: sorting and reducing (combining many values into one, such as sum, max, histogram, or merged map). Both parallelize well, but reduction comes with a mathematical requirement that is easy to overlook.

Sorting. Rayon adds parallel versions of the slice sorts through ParallelSliceMut: par_sort (stable), par_sort_unstable (usually faster), and the _by and _by_key variants. They split the slice, sort the parts on different workers, and merge. For millions of elements the speedup is often several times on a multi-core machine; for small slices the sequential sort is faster.

Reduction. A parallel reduction combines partial results computed on different threads, in an order that depends on how the work was split and stolen. That only gives a correct answer if the combining operation is:

• associative: (a op b) op c == a op (b op c), so grouping does not matter, and
• given an identity element (0 for addition, 1 for multiplication, an empty map for merging), because Rayon may start a new partial result on any worker.

Rayon's reduce(identity, op) takes both explicitly. The more flexible pair is fold + reduce: fold builds one accumulator per split (of any type, for example a HashMap), and reduce merges those accumulators. Picture a vote count: each polling station tallies its own ballots (fold), and headquarters adds up the station tallies (reduce).

   items:    [a b c d | e f g h | i j k l]     split among workers
   fold:      acc1      acc2      acc3          each worker builds its own
   reduce:      \_ merge _/        |            merge in any grouping
                     \____ merge __/            -> final result

// Cargo.toml: rayon = "1"
use rayon::prelude::*;
use std::collections::HashMap;

#[derive(Debug, Clone)]
struct Order {
    id: u32,
    customer: &'static str,
    amount: u64,                            // cents
}

fn main() {
    // Parallel sorting
    let mut data: Vec<u64> =
        (0..5_000_000u64).map(|x| x.wrapping_mul(2654435761) % 1_000_003).collect();
    data.par_sort_unstable();
    println!("sorted: {}", data.windows(2).all(|w| w[0] <= w[1]));   // sorted: true

    let customers = ["ana", "ben", "cy"];
    let mut orders: Vec<Order> = (0..300_000u32)
        .map(|id| Order {
            id,
            customer: customers[id as usize % 3],
            amount: (id as u64 % 97) * 100,
        })
        .collect();
    orders.par_sort_by_key(|o| std::cmp::Reverse(o.amount));   // stable: ties keep id order
    println!("top order id {}", orders[0].id);   // top order id 96

    // Built-in reductions
    let total: u64 = orders.par_iter().map(|o| o.amount).sum();
    let biggest = orders.par_iter().max_by_key(|o| o.amount).unwrap();
    println!("total {total}, max {}", biggest.amount);   // total 1439920200, max 9600

    // reduce with identity: the operation must be associative.
    let max_amount = orders.par_iter().map(|o| o.amount).reduce(|| 0, u64::max);
    println!("{max_amount}");               // 9600

    // fold + reduce: per-thread HashMaps, then merge them.
    let per_customer: HashMap<&str, u64> = orders
        .par_iter()
        .fold(HashMap::new, |mut acc, o| {  // one map per split, no locking
            *acc.entry(o.customer).or_insert(0) += o.amount;
            acc
        })
        .reduce(HashMap::new, |mut a, b| {  // merge maps pairwise
            for (k, v) in b {
                *a.entry(k).or_insert(0) += v;
            }
            a
        });
    let mut rows: Vec<_> = per_customer.into_iter().collect();
    rows.sort();
    println!("{rows:?}");
    // [("ana", 479974100), ("ben", 479973400), ("cy", 479972700)]
}

Floating-point addition is not associative: (1e20 + 1.0) - 1e20 is 0.0, but 1e20 - 1e20 + 1.0 is 1.0. A parallel sum of f64 values can therefore differ in the last bits from the sequential sum, and from one run to the next. When reproducibility matters, sum integers (for example cents), use a compensated algorithm, or sort and sum sequentially.

Operation	Associative?	Identity	Safe for parallel reduce
Integer +, *	Yes	0, 1	Yes (watch for overflow)
min, max	Yes	Max/min value of the type	Yes
String concatenation	Yes (not commutative)	""	Yes; Rayon preserves order for indexed sources
Set/map union	Yes	Empty collection	Yes
Floating-point +	No	0.0	Approximately; results may vary slightly
Subtraction, average	No	none	No (compute sum and count instead)

Key Takeaways

• par_sort, par_sort_unstable, and their _by_key variants are drop-in parallel sorts for large slices.
• Parallel reduction requires an associative operation and an identity; reduce(identity, op) makes both explicit.
• fold + reduce builds per-thread accumulators and then merges them, avoiding locks entirely.
• Floating-point sums are not associative, so parallel results can differ slightly; use integers or compensated sums when exactness matters.

🧪 Practice

1. Compute a word-frequency histogram of a large text in parallel with fold + reduce, and check it against a sequential version.
2. Compute the mean and standard deviation of ten million f64 values in a single parallel pass by reducing (count, sum, sum_of_squares) tuples.
3. Interview: Why is a lock-protected shared HashMap updated inside for_each usually much slower than fold + reduce? (Hint: count how many times a lock is taken, and think about contention between cores.)

Thread Pools

Spawning a fresh OS thread for every small task wastes time (creation costs microseconds) and, under load, creates more threads than the machine has cores, which then compete for CPU time and memory. A thread pool creates a fixed set of worker threads once and feeds them tasks through a queue. The workers stay alive between tasks, so the per-task cost drops to "push a closure onto a queue", and the number of threads stays bounded.

A taxi rank is the analogy: rather than buying a new car for every passenger, a fixed fleet of taxis waits at the rank and each one takes the next passenger in line.

Rayon maintains a global pool that its parallel iterators use automatically. Its size defaults to the number of logical CPUs and can be set with the RAYON_NUM_THREADS environment variable or ThreadPoolBuilder:: build_global (once, at startup). For isolation, for example to stop one subsystem from monopolizing every core, build a custom pool and run work in it with install; every parallel iterator inside the closure uses that pool.

// Cargo.toml: rayon = "1"
use rayon::prelude::*;
use rayon::ThreadPoolBuilder;

fn main() {
    println!("global pool threads: {}", rayon::current_num_threads());   // e.g. 12

    // A dedicated pool with 2 named threads.
    let pool = ThreadPoolBuilder::new()
        .num_threads(2)
        .thread_name(|i| format!("img-worker-{i}"))
        .build()
        .expect("failed to build pool");

    let sum: u64 = pool.install(|| {
        // Every Rayon operation in here runs on `pool`, not the global pool.
        println!("inside: {}", rayon::current_num_threads());   // inside: 2
        (1..=1_000_000u64).into_par_iter().sum()
    });
    println!("{sum}");                      // 500000500000

    // Fire-and-forget tasks with a scope: tasks may borrow local data.
    let mut results = vec![0u64; 4];
    pool.scope(|s| {
        for (i, slot) in results.iter_mut().enumerate() {
            s.spawn(move |_| *slot = (i as u64 + 1) * 100);
        }
    });                                     // waits for all spawned tasks
    println!("{results:?}");                // [100, 200, 300, 400]
}

Understanding a pool is easier after building one. The minimal design is a channel of boxed closures shared by N workers. Because mpsc::Receiver has a single consumer, the workers share it through Arc<Mutex<Receiver<Job>>>, and dropping the sender shuts the pool down gracefully:

use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};

type Job = Box<dyn FnOnce() + Send + 'static>;

pub struct ThreadPool {
    sender: Option<Sender<Job>>,
    workers: Vec<JoinHandle<()>>,
}

impl ThreadPool {
    pub fn new(size: usize) -> Self {
        assert!(size > 0);
        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver: Arc<Mutex<Receiver<Job>>> = Arc::new(Mutex::new(receiver));
        let workers = (0..size)
            .map(|_| {
                let rx = Arc::clone(&receiver);
                thread::spawn(move || loop {
                    // The guard is a temporary, so the lock is released before
                    // the job runs: other workers can take jobs meanwhile.
                    let msg = rx.lock().unwrap().recv();
                    match msg {
                        Ok(job) => job(),
                        Err(_) => break,     // channel closed: shut down
                    }
                })
            })
            .collect();
        ThreadPool { sender: Some(sender), workers }
    }

    pub fn execute<F: FnOnce() + Send + 'static>(&self, f: F) {
        self.sender.as_ref().unwrap().send(Box::new(f)).unwrap();
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take());           // close the channel...
        for w in self.workers.drain(..) {
            w.join().unwrap();              // ...and wait for workers to finish
        }
    }
}

fn main() {
    let (tx, rx) = mpsc::channel();
    {
        let pool = ThreadPool::new(3);
        for i in 0..6u64 {
            let tx = tx.clone();
            pool.execute(move || tx.send(i * i).unwrap());
        }
    }                                       // pool dropped: all jobs completed
    drop(tx);
    let mut out: Vec<u64> = rx.iter().collect();
    out.sort();
    println!("{out:?}");                    // [0, 1, 4, 9, 16, 25]
}

Pool	Best for	Notes
Rayon global pool	CPU-bound data parallelism	Work stealing; avoid blocking calls inside
Rayon custom ThreadPool	Isolating CPU-heavy subsystems	install, scope, spawn
Hand-written channel pool	Learning; simple background jobs	Single shared queue, no stealing
Async runtime (Tokio)	Many I/O-bound tasks	Plus spawn_blocking for blocking work (Chapter 11)

Size pools for their workload: CPU-bound pools near the number of cores (more threads only add switching), I/O-bound pools larger, because their threads spend most of their time waiting.

Key Takeaways

• A thread pool reuses a fixed set of threads, cutting per-task cost and bounding the number of threads.
• Rayon's global pool serves all parallel iterators; ThreadPoolBuilder plus install runs work on a dedicated pool.
• A basic pool is a channel of Box<dyn FnOnce() + Send> jobs shared by workers; dropping the sender shuts it down.
• Size CPU-bound pools to the core count; never block a CPU pool's threads on I/O.

🧪 Practice

1. Extend the hand-written ThreadPool so that execute returns a Receiver<T> for the job's result (a minimal future).
2. Run two CPU-heavy Rayon computations concurrently, one in a 2-thread pool and one in the global pool, and observe with current_num_threads which pool each uses.
3. Interview: In the hand-written pool, why is let msg = rx.lock().unwrap().recv(); written as a separate statement instead of while let Ok(job) = rx.lock().unwrap().recv() { job() }? (Hint: how long does the temporary MutexGuard live in each version?)

Measuring Parallel Speedup

Parallel code is more complex than sequential code, and it is only worth that complexity if it is actually faster. Intuition is unreliable here: doubling the cores rarely halves the time, and sometimes a parallel version is slower. The only way to know is to measure, and to understand the limits that theory predicts.

Speedup is S(n) = T(1) / T(n), the sequential time divided by the time with n threads. Efficiency is S(n) / n, how well each extra core is used. The ceiling is given by Amdahl's law: if a fraction p of the work can run in parallel and the rest is sequential, then

   S(n) = 1 / ((1 - p) + p / n)          and as n -> infinity:  S = 1 / (1 - p)

   p = 0.90:  n=2 -> 1.82x   n=4 -> 3.08x   n=8 -> 4.71x   n=16 -> 6.40x   max 10x
   p = 0.99:  n=2 -> 1.98x   n=4 -> 3.88x   n=8 -> 7.48x   n=16 -> 13.9x   max 100x

Even 10% sequential work caps the speedup at 10x, no matter how many cores you add. A highway is the analogy: adding lanes speeds up the open road, but if every car must pass through a single-lane toll booth, the booth sets the limit.

Real programs fall short of Amdahl's ideal for further reasons:

Limiting factor	Symptom	Remedy
Task overhead	Parallel slower on small inputs	Larger chunks, with_min_len, sequential cutoff
Memory bandwidth	Simple loops stop scaling at a few cores	Do more work per byte; improve locality
Contention (locks, shared atomics)	Adding threads makes it slower	fold/reduce, per-thread state
False sharing	Per-thread counters next to each other are slow	Pad to separate cache lines
Load imbalance	Some cores idle at the end	Smaller tasks, work stealing
Other processes, frequency scaling	Noisy, unstable numbers	Repeat runs, quiet machine

// Cargo.toml: rayon = "1"
// Run with: cargo run --release   (debug builds make timing meaningless)
use rayon::prelude::*;
use std::hint::black_box;
use std::time::{Duration, Instant};

fn heavy(x: u64) -> u64 {
    (0..200).fold(x, |acc, i| acc.wrapping_mul(6364136223846793005).wrapping_add(i))
}

// Best of several runs reduces noise from the OS and CPU frequency changes.
fn best_of<F: FnMut() -> u64>(runs: usize, mut f: F) -> Duration {
    (0..runs)
        .map(|_| {
            let t = Instant::now();
            black_box(f());                 // stop the optimizer deleting the work
            t.elapsed()
        })
        .min()
        .unwrap()
}

fn main() {
    let threads = rayon::current_num_threads();
    for n in [1_000u64, 100_000, 5_000_000] {
        let seq = best_of(5, || (0..n).map(heavy).fold(0, u64::wrapping_add));
        let par = best_of(5, || {
            (0..n).into_par_iter().map(heavy).reduce(|| 0, u64::wrapping_add)
        });
        let speedup = seq.as_secs_f64() / par.as_secs_f64();
        let efficiency = 100.0 * speedup / threads as f64;
        println!(
            "n = {n:>9}: seq {seq:>10.2?}  par {par:>10.2?}  \
             speedup {speedup:5.2}x  efficiency {efficiency:3.0}%"
        );
    }
    // One run on a 12-thread machine (numbers vary by hardware):
    //   n =      1000: seq  20.41µs  par   2.47ms  speedup 0.01x  (overhead dominates)
    //   n =    100000: seq   1.98ms  par 916.49µs  speedup 2.16x
    //   n =   5000000: seq  98.40ms  par  20.07ms  speedup 4.90x  (efficiency 41%)
}

A disciplined measuring routine:

1. Build in release mode and measure the code path that matters, with realistic input sizes.
2. Establish a sequential baseline using the best sequential algorithm, not the parallel code run on one thread.
3. Repeat runs and report the minimum or median; use a benchmarking harness such as criterion (Chapter 12) for statistically sound results.
4. Scale the thread count (RAYON_NUM_THREADS=1,2,4,...) and plot speedup. A curve that flattens early suggests a sequential fraction or a memory bottleneck; a curve that drops suggests contention.
5. Profile (perf, flame graphs) to see where time goes before optimizing.

Key Takeaways

• Speedup is T(1) / T(n); efficiency is speedup per core.
• Amdahl's law caps speedup at 1 / (1 - p): the sequential fraction dominates as cores increase.
• Overhead, memory bandwidth, contention, false sharing, and imbalance push real results below the ideal.
• Measure release builds against a good sequential baseline, repeat runs, and scale the thread count to understand the curve.

🧪 Practice

1. Run the benchmark with RAYON_NUM_THREADS set to 1, 2, 4, and 8 and tabulate speedup and efficiency for the largest input. Estimate p from your numbers using Amdahl's law.
2. Replace heavy with a function that does almost no work (x ^ 1) and explain why the speedup collapses.
3. Interview: A team parallelized a step that takes 30% of a request's time, on an 8-core server. What is the best possible overall improvement, and what would you look at next? (Hint: apply Amdahl's law with p = 0.3.)

---