Threads, Channels and Shared State

After this lesson you can run work on several threads, get each thread's result back, and share a counter between them without a data race. In Smart Pointers you met Arc as the thread-safe Rc, and in Closures you met move; here both do real work.

Spawn, return a value, join

thread::spawn starts a thread running a closure and returns a JoinHandle<T>, where T is whatever the closure returns. join() waits for the thread and gives Result<T, _>, which is Err if the thread panicked.

let ok = thread::spawn(|| 6 * 7);
let bad = thread::spawn(|| { let v: Vec<i32> = Vec::new(); v[0] });
println!("{}", ok.join().unwrap());   // 42
println!("{}", bad.join().is_err());  // true

The panic message goes to stderr and the main thread carries on.

The closure must own what it uses

A spawned thread can outlive the function that started it, so the closure has to be 'static: it may not borrow locals.

let words = vec!["a", "b"];
let h = thread::spawn(|| println!("{}", words.len()));

error[E0373]: closure may outlive the current function, but it borrows `words`

move fixes it by transferring words into the thread, but then no other thread can have it. To share read-only data, put it in an Arc and give each thread its own Arc::clone; the clone bumps an atomic count, it does not copy the data.

Send and Sync

Two marker traits decide what may cross into a thread. Send: ownership can move to another thread. Sync: &T can be used from several threads. The compiler infers both from a type's fields; you never write them. Rc has a non-atomic count so it is not Send, and RefCell is not Sync:

let rc = std::rc::Rc::new(1);
thread::spawn(move || println!("{}", rc));

error[E0277]: `std::rc::Rc<i32>` cannot be sent between threads safely

Channels and Mutex together

mpsc::channel() (multiple producer, single consumer) returns a (Sender, Receiver). Clone the Sender once per worker. A for v in rx loop ends only when every Sender is dropped, so main must drop its own. For a shared counter use Arc<Mutex<T>>: lock() returns a guard that unlocks when it is dropped, so keep it alive for as short a time as possible.

This program counts primes in four ranges of 25 numbers:

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

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

fn main() {
    let tested = Arc::new(Mutex::new(0u32));
    let (tx, rx) = mpsc::channel();
    for id in 0..4u32 {
        let tx = tx.clone();
        let tested = Arc::clone(&tested);
        thread::spawn(move || {
            let (lo, hi) = (id * 25, id * 25 + 25);
            let found = (lo..hi).filter(|&n| is_prime(n)).count();
            *tested.lock().unwrap() += hi - lo;   // guard dropped at the ;
            tx.send((id, found)).unwrap();
        });
    }
    drop(tx);
    let mut rows: Vec<(u32, usize)> = rx.iter().collect();
    rows.sort();
    for (id, found) in &rows { println!("range {}: {} primes", id, found); }
    println!("tested: {}", *tested.lock().unwrap());
}

range 0: 9 primes
range 1: 6 primes
range 2: 6 primes
range 3: 4 primes
tested: 100

Without the rows.sort() the same program prints the rows in the order the threads finish, for example range 2, range 0, range 3, range 1, and a different order on the next run. Without drop(tx) the program prints nothing and never exits, because rx.iter() is still waiting on the sender that main holds. For the same reason, never print from inside a worker: send results to main and print there.

Common mistakes

• Keeping main's Sender alive. The receive loop never ends and the program hangs.
• Printing in arrival order. Output that matches on one run and differs on the next. Collect, sort by a worker id, then print.
• Rounding a split down. Giving 8 workers 100 / 8 = 12 numbers each covers 96 of 100; the last four numbers are silently skipped. Round up and clamp the range to the data length.
• Locking twice in one thread. A second lock() while the first guard is alive waits forever. Keep the guard in a short statement or inner block.

Your exercise

Split a list of numbers across k worker threads and report per-worker statistics, plus a processed count kept in a shared Arc<Mutex<usize>>. The starter does the parsing and the printing; the likeliest mistake is mishandling workers whose range is empty.