Concurrency

A program so far has done one thing at a time. A thread lets part of it run at the same time as the rest: on a machine with several cores, threads really do run in parallel, so work split across four threads can finish close to four times faster. The cost is that two threads can touch the same data at the same moment, and in most languages that leads to data races, bugs that only show up now and then and are very hard to reproduce.

Rust's ownership and borrowing rules extend to threads. Code that could race on data does not compile. The Rust community calls this fearless concurrency. This lesson covers the tools in the standard library: spawning threads, passing messages between them with channels, and sharing data with Arc and Mutex.

std::thread::spawn takes a closure and runs it on a new thread. The thread that runs main carries on with its own work in the meantime:

use std::thread;
use std::time::Duration;
 
fn main() {
    let handle = thread::spawn(|| {
        for step in 1..=3 {
            println!("spawned thread: step {step}");
            thread::sleep(Duration::from_millis(1));
        }
    });
 
    for step in 1..=3 {
        println!("main thread: step {step}");
        thread::sleep(Duration::from_millis(1));
    }
 
    handle.join().unwrap();
}

One run printed:

main thread: step 1
spawned thread: step 1
main thread: step 2
spawned thread: step 2
main thread: step 3
spawned thread: step 3

Run it a few times and the order changes. The operating system decides when each thread runs, and your program can't rely on any particular interleaving.

spawn returns a JoinHandle. Calling join() on it blocks until that thread has finished. Without the join, main could return first, and when main returns the whole process exits, taking any unfinished threads with it.

The closure's return value is not lost: join() hands it back. A thread that returns a usize gives you a JoinHandle<usize>:

use std::thread;
 
fn main() {
    let handle = thread::spawn(|| (1..=1_000_000_u64).filter(|n| n % 7 == 0).count());
 
    println!("counting in the background...");
    let count = handle.join().unwrap();
    println!("{count} multiples of 7");
}

counting in the background...
142857 multiples of 7

join() returns a Result. It is Err when the thread panicked, so a panic in one thread does not bring down the thread that joins it:

use std::thread;
 
fn main() {
    let handle = thread::spawn(|| {
        let readings: Vec<u32> = Vec::new();
        readings[0]
    });
 
    match handle.join() {
        Ok(first) => println!("first reading: {first}"),
        Err(_) => println!("the worker panicked"),
    }
    println!("main keeps going");
}

the worker panicked
main keeps going

The panic message (index out of bounds) still goes to stderr. Most code calls .unwrap() on join(), which passes a worker's panic on to the thread that joined it.

A spawned thread can run for longer than the function that created it, so its closure is not allowed to borrow that function's local variables. This closure only reads words, so it tries to borrow it:

use std::thread;
 
fn main() {
    let words = vec!["ferris", "crab", "rustacean"];
 
    let handle = thread::spawn(|| {
        let longest = words.iter().max_by_key(|w| w.len()).unwrap();
        println!("longest word: {longest}");
    });
 
    handle.join().unwrap();
}

cargo run fails with E0373: closure may outlive the current function, but it borrows words, which is owned by the current function, with help to use the move keyword

The compiler doesn't look at the join() below; as far as the type of spawn is concerned, the thread might still be running after main has dropped words. The fix is the one it suggests: a move closure takes ownership of what it captures, so the data goes to the thread and lives as long as it does.

use std::thread;
 
fn main() {
    let words = vec!["ferris", "crab", "rustacean"];
 
    let handle = thread::spawn(move || {
        let longest = words.iter().max_by_key(|w| w.len()).unwrap();
        println!("longest word: {longest}");
    });
 
    handle.join().unwrap();
}

longest word: rustacean

After the move, words belongs to the thread, and main can't use it any more. If both need it, clone it before spawning, or share it with Arc (below).

This rule is in the signature of spawn:

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
where
    F: FnOnce() -> T + Send + 'static,
    T: Send + 'static,

'static means the closure holds no borrowed data that could go away, and Send means it is safe to hand to another thread (more on Send at the end). The same bounds apply to the value it returns.

To split work, spawn a thread per piece, keep the handles in a Vec, and join them all. Here four threads each count the primes in a quarter of a range:

use std::thread;
 
fn is_prime(n: u64) -> bool {
    n >= 2 && (2..).take_while(|d| d * d <= n).all(|d| n % d != 0)
}
 
fn main() {
    let mut handles = Vec::new();
 
    for chunk in 0..4 {
        let start = chunk * 25_000;
        let end = start + 25_000;
        handles.push(thread::spawn(move || {
            (start..end).filter(|&n| is_prime(n)).count()
        }));
    }
 
    let mut total = 0;
    for (chunk, handle) in handles.into_iter().enumerate() {
        let count = handle.join().unwrap();
        println!("chunk {chunk}: {count} primes");
        total += count;
    }
    println!("total: {total} primes below 100000");
}

chunk 0: 2762 primes
chunk 1: 2371 primes
chunk 2: 2260 primes
chunk 3: 2199 primes
total: 9592 primes below 100000

The threads finish in any order, but the results come out in order, because the handles are joined in the order they were spawned. move copies start and end into each closure, so every thread gets its own.

Threads are not free: each has its own stack and costs some time to start. A thread per core is a good rule of thumb, and thread::available_parallelism() tells you how many that is. Spawning one per tiny piece of work is usually slower than a plain loop.

Sometimes the threads only need to borrow data for a while, and moving or cloning it is a waste. thread::scope spawns threads that are guaranteed to finish before scope returns, so they can borrow local variables:

use std::thread;
 
fn main() {
    let temperatures = vec![18.5, 21.0, 19.5, 23.0, 22.5, 17.0];
    let (morning, evening) = temperatures.split_at(3);
 
    let (low, high) = thread::scope(|s| {
        let low = s.spawn(|| morning.iter().copied().fold(f64::INFINITY, f64::min));
        let high = s.spawn(|| evening.iter().copied().fold(f64::NEG_INFINITY, f64::max));
        (low.join().unwrap(), high.join().unwrap())
    });
 
    println!("morning low: {low}, evening high: {high}");
    println!("still usable: {} readings", temperatures.len());
}

morning low: 18.5, evening high: 23
still usable: 6 readings

No move, no clone: both threads borrow slices of temperatures, and main still owns the vector afterwards. Any scoped thread you don't join is joined automatically at the end of the scope. Reach for thread::scope when the work fits inside one function call, and thread::spawn when a thread has to outlive it.

One way for threads to cooperate is to not share data at all, and send it instead. A channel has two ends: a sender (tx) and a receiver (rx). Values sent into one end come out of the other in the same order. std::sync::mpsc stands for multiple producer, single consumer: there can be many senders but only one receiver.

use std::sync::mpsc;
use std::thread;
 
fn main() {
    let (tx, rx) = mpsc::channel();
 
    thread::spawn(move || {
        for file in ["intro.md", "setup.md", "faq.md"] {
            tx.send(format!("uploaded {file}")).unwrap();
        }
    });
 
    for message in rx {
        println!("{message}");
    }
    println!("all uploads done");
}

uploaded intro.md
uploaded setup.md
uploaded faq.md
all uploads done

A few things are going on:

• mpsc::channel() returns the pair (Sender<T>, Receiver<T>). T is inferred from what you send, here String.
• send moves the value into the channel. Ownership goes with it, so the sending thread can't use the value afterwards, and the two threads never hold it at the same time.
• for message in rx waits for each message in turn. The loop ends when every sender has been dropped, which happens here when the thread's closure finishes.

The receiver has more than the loop: rx.recv() waits for one value and returns Err once all senders are gone, rx.try_recv() returns straight away whether or not something has arrived, and rx.recv_timeout(duration) waits at most that long. On the other side, send returns Err if the receiver has been dropped, since nobody would ever read the value.

To have several threads send to the same receiver, clone the sender, one clone per thread:

use std::sync::mpsc;
use std::thread;
 
fn main() {
    let (tx, rx) = mpsc::channel();
 
    let stations = [
        ("north", vec![2, 0, 5]),
        ("south", vec![1, 1, 1]),
        ("east", vec![7, 3, 0]),
    ];
 
    for (station, samples) in stations {
        let tx = tx.clone();
        thread::spawn(move || {
            let total: u32 = samples.iter().sum();
            tx.send((station, total)).unwrap();
        });
    }
    drop(tx);
 
    let mut totals: Vec<(&str, u32)> = rx.iter().collect();
    totals.sort();
 
    for (station, mm) in totals {
        println!("{station}: {mm} mm of rain");
    }
}

east: 10 mm of rain
north: 7 mm of rain
south: 3 mm of rain

Each thread gets its own tx (the let tx = tx.clone(); shadows the original inside the loop body). The messages arrive in whatever order the threads finish, so the program sorts them before printing.

Note the drop(tx). The receiver stops only when every sender is gone, and main still holds the original one. Leave that line out and rx.iter() waits forever for a message that will never come: the program compiles, prints nothing and hangs.

mpsc::channel is unbounded: senders never wait. mpsc::sync_channel(n) makes a channel that holds at most n values, and send blocks when it is full, which keeps a fast producer from running far ahead of a slow consumer.

Channels move data between threads. Sometimes several threads really do need to read and change the same value. A Mutex<T> (mutual exclusion) guards a value so that only one thread at a time can reach it:

use std::sync::Mutex;
 
fn main() {
    let shelf = Mutex::new(vec!["apples", "pears"]);
 
    {
        let mut items = shelf.lock().unwrap();
        items.push("plums");
    }
 
    println!("{:?}", shelf.lock().unwrap());
}

["apples", "pears", "plums"]

lock() waits until no other thread holds the lock, then returns a MutexGuard. The guard derefs to the value inside, so you use it like a &mut Vec. When the guard is dropped, here at the end of the inner block, the lock is released. There is no unlock to forget.

lock() returns a Result because a mutex can be poisoned: if a thread panics while holding the lock, the value might be half-updated, and later lock() calls return Err. Calling unwrap() on it is the usual choice, since it turns that into a panic in the next thread too.

Note that shelf is not mut. Like RefCell, a Mutex gives you mutable access through a shared reference, and checks at runtime that access is exclusive. The difference is that a RefCell panics when the rule is broken, and a Mutex makes the second thread wait.

Hold the lock for the whole of an operation that must not be interrupted. Here, checking the balance and taking money out happen under one lock, so two threads can't both see enough money and both withdraw it:

use std::sync::Mutex;
use std::thread;
 
fn withdraw(balance: &Mutex<u32>, amount: u32) -> bool {
    let mut balance = balance.lock().unwrap();
    if *balance >= amount {
        *balance -= amount;
        true
    } else {
        false
    }
}
 
fn main() {
    let balance = Mutex::new(100);
 
    let succeeded = thread::scope(|s| {
        let handles: Vec<_> = (0..5).map(|_| s.spawn(|| withdraw(&balance, 30))).collect();
        let results = handles.into_iter().map(|h| h.join().unwrap());
        results.filter(|&ok| ok).count()
    });
 
    println!("{succeeded} withdrawals succeeded");
    println!("left: {}", balance.lock().unwrap());
}

3 withdrawals succeeded
left: 10

Five threads try to take 30 from 100. Whichever order they run in, exactly three succeed. Scoped threads could borrow balance directly, because they all finish inside thread::scope.

Threads from thread::spawn can't borrow, so each one needs to own the mutex. The smart pointers lesson solved "several owners" with Rc, but Rc updates its count with plain arithmetic, which two threads could do at the same moment and get wrong. Arc<T> (atomically reference counted) is the same idea with a thread-safe count. Arc::clone gives each thread its own pointer to one shared value:

use std::sync::{Arc, Mutex};
use std::thread;
 
fn main() {
    let done = Arc::new(Mutex::new(Vec::new()));
    let mut handles = Vec::new();
 
    for job in ["resize", "compress", "upload"] {
        let done = Arc::clone(&done);
        handles.push(thread::spawn(move || {
            // the job's real work would happen here
            done.lock().unwrap().push(job);
        }));
    }
 
    for handle in handles {
        handle.join().unwrap();
    }
 
    let mut finished = done.lock().unwrap();
    finished.sort();
    println!("finished: {finished:?}");
    println!("owners left: {}", Arc::strong_count(&done));
}

finished: ["compress", "resize", "upload"]
owners left: 1

The pattern is always the same: clone the Arc before the move closure, so the closure takes the clone and the original stays in main. When a thread finishes, its clone is dropped, which is why only one owner is left at the end.

Arc on its own gives shared, read-only access, like Rc. For data that threads only read, such as settings or a lookup table, Arc<T> is all you need. Add the Mutex when they also write.

Other tools in std::sync for the same job:

A thread that holds one lock and waits for another, while a second thread does the opposite, waits forever: a deadlock. The compiler does not catch those. Keep guards short-lived, and when a thread needs two locks, take them in the same order everywhere.

How does the compiler know Arc is safe to send to a thread and Rc is not? Through two marker traits, which have no methods:

• Send: a value of this type can be moved to another thread.
• Sync: a value of this type can be shared between threads, that is, &T is Send.

Both are implemented automatically: a struct is Send if all its fields are. Almost every type is both. The exceptions are the ones that would be unsound across threads:

thread::spawn requires Send for the closure and its return value, as its signature above shows. So a generic function that spawns threads with a T has to promise the same, with a bound like T: Send + 'static. You will rarely implement Send or Sync yourself (doing so is unsafe); mostly you meet them in error messages, when a type is not allowed to cross a thread.

use std::sync::mpsc;
use std::thread;
 
fn main() {
    let (tx, rx) = mpsc::channel();
 
    for worker in 1..=3 {
        thread::spawn(move || {
            tx.send(worker * 10).unwrap();
        });
    }
 
    let total: u32 = rx.iter().sum();
    println!("{total}");
}

cargo run fails with E0382: use of moved value tx, value moved into closure here, in previous iteration of loop

The first iteration's move closure takes tx, and the second has nothing left to take. Give each thread its own sender with let tx = tx.clone(); at the top of the loop body, and drop(tx) after the loop so that rx.iter() ends.

use std::rc::Rc;
use std::thread;
 
fn main() {
    let settings = Rc::new(String::from("dark mode"));
 
    let for_thread = Rc::clone(&settings);
    let handle = thread::spawn(move || {
        println!("theme: {for_thread}");
    });
 
    handle.join().unwrap();
}

cargo run fails with E0277: Rc of String cannot be sent between threads safely, with a note that the trait Send is not implemented for Rc of String

Rc is not Send, and spawn requires it. Swap Rc for Arc (use std::sync::Arc;); the rest of the code stays the same.

use std::sync::{Arc, Mutex};
use std::thread;
 
fn main() {
    let log = Arc::new(Mutex::new(Vec::new()));
 
    let writer = Arc::clone(&log);
    thread::spawn(move || {
        writer.push("started");
    })
    .join()
    .unwrap();
 
    println!("{:?}", log.lock().unwrap());
}

cargo run fails with E0599: no method named push found for struct Arc of Mutex of Vec in the current scope

An Arc derefs to what it holds, which here is the Mutex, not the Vec. The only way to the Vec is through a lock: writer.lock().unwrap().push("started"). That is the point of a mutex: you can't forget to lock it.

use std::sync::Mutex;
 
fn main() {
    let queue = Mutex::new(vec![3, 1, 2]);
 
    let guard = queue.lock().unwrap();
    guard.sort();
 
    println!("{guard:?}");
}

cargo run fails with E0596: cannot borrow guard as mutable, as it is not declared as mutable, with help to add mut

The mutex itself doesn't need mut, but changing the value goes through the guard's DerefMut, and that needs a mutable binding: let mut guard.

• thread::spawn(closure) runs the closure on a new thread and returns a JoinHandle. join() waits for the thread and returns its value, or Err if it panicked. When main returns, unfinished threads are stopped.
• Threads run in an order you don't control. Don't rely on it; join the handles or sort the results.
• A spawned closure must be 'static, so use move to give it ownership of what it uses. thread::scope lets threads borrow local data instead.
• Channels (mpsc::channel) send values from any number of senders to one receiver. Clone the sender per thread, and drop every sender so the receiver's loop ends.
• Mutex<T> gives one thread at a time access to a value; the lock is released when the guard is dropped. Arc<T> shares ownership across threads, and Arc<Mutex<T>> is shared, mutable data.
• Send and Sync mark the types that may cross threads. The compiler checks them, which is why Rc and RefCell are rejected where Arc and Mutex are accepted.