Threads, atomics, and locks
This chapter covers
- Starting threads with
thread::spawnandthread::scope, and collecting their results - The
SendandSynctraits, which let the compiler reject data races - Atomic integers, and why
count += 1from two threads can lose an update - Splitting a sum across threads, measuring it, and Amdahl’s law
- A spin lock built from one atomic flag, with acquire and release ordering
- A counting semaphore built from a
Mutexand aCondvar - What happens to a
Mutexwhen its holder panics, and how to detect a deadlock
Chapter 15 showed that a laptop CPU has several cores. A program that runs on one thread uses one of them. To use the others, the program must split its work across several threads: independent sequences of instructions that the operating system schedules onto cores.
Threads share the process’s memory, so they can all read the same data. That sharing also makes them dangerous. When two threads change the same memory at the same time, the result can be wrong in ways that are hard to reproduce.
Rust’s promise here is narrow and specific. Safe Rust code cannot contain a data race. In a data race, two threads access the same memory at the same time, and at least one of them writes.
Nothing orders the two accesses. The
compiler checks this with two traits, Send and Sync. The chapter starts there.
16.1 Starting threads
16.1.1 What a thread is
A process owns memory: the program’s code, its static data, and its heap. A thread is one path of execution through that code. Each thread has its own stack, which holds the local variables of the functions it is running. Each also has its own copy of the CPU registers, including the instruction pointer that says which instruction runs next. Everything else is shared. Every thread reads the same code and can reach the same heap (figure 16.1).
The kernel’s scheduler decides which threads run. A core runs one thread at a time. When there are more runnable threads than cores, the scheduler gives each a short slice of time. Between slices it saves one thread’s registers and loads another’s. That swap is a context switch, and it costs a few microseconds.
A thread that waits for I/O or for a lock is not runnable. It uses no CPU time while it waits, only the memory of its stack.
Because the heap is shared, two threads can reach the same value at the same moment. Section 16.2 shows what goes wrong then, and the rest of the chapter builds the tools that prevent it.
16.1.2 spawn and join
std::thread::spawn starts a new thread that runs a closure. It returns a JoinHandle. Calling join() on the
handle waits for the thread to finish and returns the closure’s result:
let handle = thread::spawn(|| 6 * 7);
assert_eq!(handle.join().expect("worker panicked"), 42);join returns a Result, because the thread might have panicked. Ok carries the closure’s return value.
A spawned thread may outlive the function that started it. So the closure cannot borrow local variables, which
might be gone by the time the thread uses them. The compiler requires the closure to own everything it uses, or
to borrow only data that lives for the whole program. That requirement is written 'static. To share data with
a spawned thread, you move an Arc into the closure, as section 8.6 did.
16.1.3 Scoped threads
thread::scope removes that restriction for work that finishes before the function continues. It takes a
closure that receives a scope value. Threads started with scope.spawn may borrow local data. The scope waits
for every one of them before thread::scope returns, so the borrowed data is still alive while they run.
The file threads.rs uses a scope to sum a slice on several threads (figure 16.2).
Listing 16.1 available_workers and parallel_sum (lines 31 to 69). src/problems/threads.rs
use std::{
sync::{
OnceLock,
atomic::{AtomicUsize, Ordering},
},
thread,
};
/// Number of hardware threads the runtime can use, defaulting to 1.
pub fn available_workers() -> usize {
thread::available_parallelism()
.map(|count| count.get())
.unwrap_or(1)
}
/// Sum a slice by splitting it into chunks that run on scoped threads.
///
/// `thread::scope` is what makes this ergonomic: the workers borrow `values`
/// directly, so there is no `Arc` and no `'static` bound. The scope waits for
/// every thread, which removes the forgotten-join bug.
pub fn parallel_sum(values: &[i32]) -> i64 {
if values.is_empty() {
return 0;
}
let workers = available_workers().min(values.len());
let chunk_size = values.len().div_ceil(workers);
thread::scope(|scope| {
let handles: Vec<_> = values
.chunks(chunk_size)
.map(|chunk| {
scope.spawn(move || {
chunk
.iter()
.map(|value| i64::from(*value))
.sum::<i64>()
})
})
.collect();
handles
.into_iter()
.map(|handle| handle.join().expect("worker panicked"))
.sum()
})
}thread::available_parallelism() asks the operating system how many threads can run at the same time. On this
laptop it returns 8: four cores, each running two hardware threads. It returns a Result, so the code falls back
to 1 if the question cannot be answered.
The function never starts more workers than there are values. values.len().div_ceil(workers) computes a chunk
size that rounds up, so the chunks cover every value. values.chunks(chunk_size) then yields sub-slices of that
size, with a shorter last one.
Each worker converts its values to i64 with i64::from(*value) before adding. The sum of many i32 values can
exceed the i32 range, and an i64 has room for it.
The order of two steps decides whether the threads run in parallel. The workers are all started, and their handles collected into a Vec,
before any is joined. Iterators are lazy. A single chain like .map(spawn).map(join) would start one thread, wait
for it, then start the next. The work would run one chunk at a time. Collecting first starts every thread before
waiting on any of them.
The closure passed to scope.spawn is a move closure. It moves chunk, a &[i32], into the thread. Moving a
reference copies the reference, not the data.
16.1.4 Send and Sync
How does the compiler know which values may be used from another thread? Two marker traits describe it:
- A type is
Sendif a value of it can be moved to another thread. - A type is
Syncif a reference to it,&T, can be shared with other threads. Put another way,TisSyncexactly when&TisSend.
These traits have no methods. They are auto traits: the compiler implements them for a type automatically
when all of the type’s fields have them. i32, String, Vec<i32>, and Arc<Mutex<T>> are both Send and
Sync.
Rc is neither. Its reference count is a plain integer, so two threads cloning the same Rc could lose a count,
as section 16.2 shows. RefCell is Send but not Sync: its borrow flag is not safe to change from two threads.
thread::spawn and scope.spawn require their closures to be Send. A closure that captures an Rc is
therefore rejected at compile time.
Listing 16.2 A compile-time check (lines 121 to 125).
/// A compile-time assertion that `T` may cross thread boundaries.
///
/// Calling `assert_send_sync::<Client>()` fails to compile if `Client` is not
/// thread-safe, which turns a concurrency assumption into a build error.
pub fn assert_send_sync<T: Send + Sync>() {}assert_send_sync has an empty body. Calling assert_send_sync::<AtomicCounter>() compiles only if
AtomicCounter is Send + Sync. If a later change added an Rc field to it, the call would stop compiling. The
check costs nothing at run time.
16.2 Atomic integers
16.2.1 The lost update
count += 1 looks like one step, but the CPU performs three: read the value, add one, and write the result back.
If two threads do this at the same time, their steps can interleave (figure 16.3).
Safe Rust does not compile this: sharing a plain usize for writing between threads needs &mut in two places
at once. Other languages compile it, and the program loses counts only under load.
An atomic type performs the read, the change, and the write as one indivisible operation. No other thread can
see or change the value in between. AtomicUsize::fetch_add(1, ordering) adds 1 and returns the old value.
16.2.2 A counter without a lock
Listing 16.3 AtomicCounter and scoped_increment (lines 71 to 110).
/// A counter that several threads may increment without a lock.
///
/// `Relaxed` is correct here because the counter is the only value being
/// communicated; there is no other data whose visibility depends on it.
#[derive(Default)]
pub struct AtomicCounter {
value: AtomicUsize,
}
impl AtomicCounter {
pub const fn new() -> Self {
Self {
value: AtomicUsize::new(0),
}
}
/// Increment and return the new value.
pub fn increment(&self) -> usize {
self.value
.fetch_add(1, Ordering::Relaxed)
+ 1
}
pub fn get(&self) -> usize {
self.value.load(Ordering::Relaxed)
}
}
/// Increment a counter from several scoped threads that share it by reference.
pub fn scoped_increment(counter: &AtomicCounter, threads: usize, per_thread: usize) {
thread::scope(|scope| {
for _ in 0..threads {
scope.spawn(|| {
for _ in 0..per_thread {
counter.increment();
}
});
}
});
}increment takes &self, not &mut self. An atomic can be changed through a shared reference, the same way a
RefCell can, but safely across threads. So many threads can hold &AtomicCounter and increment it at once.
fetch_add returns the value before the addition, so increment adds 1 to report the new value.
scoped_increment starts threads threads that each increment the counter per_thread times. The closure has
no move, so it borrows counter, which the scope allows. The test runs 8 threads × 500 increments and checks
that the total is exactly 4,000. No increment is lost.
16.2.3 Memory ordering
Every atomic operation takes an Ordering argument. The ordering answers a question beyond the atomic value itself. When this thread changes the atomic, what
may other threads assume about the other memory this thread changed?
Compilers and CPUs reorder memory operations to run faster. Within one thread, the reordering is invisible. Another thread watching the memory can see writes in a different order from the one in the source code. The ordering argument limits that reordering around the atomic operation.
Relaxedmakes the operation itself atomic, and promises nothing about other memory. That is enough for a counter whose value no other data depends on, which is the case here.Release, used when writing, andAcquire, used when reading, work as a pair. Section 16.4 uses them to build a lock.SeqCstis the strongest. AllSeqCstoperations appear in one order that every thread agrees on. It is the simplest to reason about, and sometimes slower.
16.2.4 Initializing a value once
Listing 16.4 OnceLock (lines 112 to 119).
/// A value initialized once and shared by every caller.
///
/// `OnceLock` is the primitive behind lazy process-wide configuration, and it is
/// safe because initialization happens at most once even under concurrent calls.
pub fn build_channel() -> &'static str {
static CHANNEL: OnceLock<&'static str> = OnceLock::new();
CHANNEL.get_or_init(|| "stable")
}A static is a single value shared by the whole program. OnceLock holds a value that starts empty and is set at
most once. get_or_init runs the closure the first time, and returns the stored value after that. If several
threads call it at the same moment, one runs the closure and the others wait for its result.
Listing 16.5 The complete file, with tests. src/problems/threads.rs
//! Threads and fearless concurrency.
//!
//! Rust's claim about concurrency is narrow and checkable: safe code cannot
//! produce a data race, because `Send` and `Sync` describe what may cross a
//! thread boundary and the compiler refuses programs that violate the rules.
//!
//! - `Send` means a value may be moved to another thread.
//! - `Sync` means `&T` may be shared across threads.
//!
//! Both are auto traits: a type is `Send`/`Sync` when its fields are. `Rc` and
//! `RefCell` are the standard counter-examples, which is precisely why sharing
//! them across threads does not compile. The compile error is the feature, not an
//! obstacle.
//!
//! Three ways to get work onto other threads, in increasing order of ceremony:
//!
//! 1. `thread::scope` for work that borrows local data and finishes with the
//! scope. No `'static`, no `Arc`.
//! 2. `thread::spawn` for work that must own its data and outlive the caller.
//! 3. A thread pool when the number of units of work is large and threads should
//! be reused.
use std::{
sync::{
OnceLock,
atomic::{AtomicUsize, Ordering},
},
thread,
};
/// Number of hardware threads the runtime can use, defaulting to 1.
pub fn available_workers() -> usize {
thread::available_parallelism()
.map(|count| count.get())
.unwrap_or(1)
}
/// Sum a slice by splitting it into chunks that run on scoped threads.
///
/// `thread::scope` is what makes this ergonomic: the workers borrow `values`
/// directly, so there is no `Arc` and no `'static` bound. The scope waits for
/// every thread, which removes the forgotten-join bug.
pub fn parallel_sum(values: &[i32]) -> i64 {
if values.is_empty() {
return 0;
}
let workers = available_workers().min(values.len());
let chunk_size = values.len().div_ceil(workers);
thread::scope(|scope| {
let handles: Vec<_> = values
.chunks(chunk_size)
.map(|chunk| {
scope.spawn(move || {
chunk
.iter()
.map(|value| i64::from(*value))
.sum::<i64>()
})
})
.collect();
handles
.into_iter()
.map(|handle| handle.join().expect("worker panicked"))
.sum()
})
}
/// A counter that several threads may increment without a lock.
///
/// `Relaxed` is correct here because the counter is the only value being
/// communicated; there is no other data whose visibility depends on it.
#[derive(Default)]
pub struct AtomicCounter {
value: AtomicUsize,
}
impl AtomicCounter {
pub const fn new() -> Self {
Self {
value: AtomicUsize::new(0),
}
}
/// Increment and return the new value.
pub fn increment(&self) -> usize {
self.value
.fetch_add(1, Ordering::Relaxed)
+ 1
}
pub fn get(&self) -> usize {
self.value.load(Ordering::Relaxed)
}
}
/// Increment a counter from several scoped threads that share it by reference.
pub fn scoped_increment(counter: &AtomicCounter, threads: usize, per_thread: usize) {
thread::scope(|scope| {
for _ in 0..threads {
scope.spawn(|| {
for _ in 0..per_thread {
counter.increment();
}
});
}
});
}
/// A value initialized once and shared by every caller.
///
/// `OnceLock` is the primitive behind lazy process-wide configuration, and it is
/// safe because initialization happens at most once even under concurrent calls.
pub fn build_channel() -> &'static str {
static CHANNEL: OnceLock<&'static str> = OnceLock::new();
CHANNEL.get_or_init(|| "stable")
}
/// A compile-time assertion that `T` may cross thread boundaries.
///
/// Calling `assert_send_sync::<Client>()` fails to compile if `Client` is not
/// thread-safe, which turns a concurrency assumption into a build error.
pub fn assert_send_sync<T: Send + Sync>() {}
#[cfg(test)]
mod tests {
use std::thread;
use super::*;
#[test]
fn parallel_sum_matches_the_sequential_sum() {
let values: Vec<i32> = (1..=10_000).collect();
let expected: i64 = values
.iter()
.map(|value| i64::from(*value))
.sum();
assert_eq!(parallel_sum(&values), expected);
}
#[test]
fn parallel_sum_handles_small_and_empty_inputs() {
assert_eq!(parallel_sum(&[]), 0);
assert_eq!(parallel_sum(&[7]), 7);
assert_eq!(parallel_sum(&[-1, 1, -2, 2]), 0);
}
#[test]
fn scoped_threads_share_a_counter_without_arc() {
let counter = AtomicCounter::new();
scoped_increment(&counter, 8, 500);
assert_eq!(counter.get(), 4_000);
}
#[test]
fn spawn_returns_a_join_handle_that_carries_the_result() {
let handle = thread::spawn(|| 6 * 7);
assert_eq!(handle.join().expect("worker panicked"), 42);
}
#[test]
fn once_lock_initializes_a_single_value() {
assert_eq!(build_channel(), "stable");
assert_eq!(build_channel(), build_channel());
}
#[test]
fn counter_is_thread_safe_by_construction() {
assert_send_sync::<AtomicCounter>();
assert!(available_workers() >= 1);
}
}$ cargo test --lib problems::threads
running 6 tests
test problems::threads::tests::counter_is_thread_safe_by_construction ... ok
test problems::threads::tests::once_lock_initializes_a_single_value ... ok
test problems::threads::tests::spawn_returns_a_join_handle_that_carries_the_result ... ok
test problems::threads::tests::scoped_threads_share_a_counter_without_arc ... ok
test problems::threads::tests::parallel_sum_handles_small_and_empty_inputs ... ok
test problems::threads::tests::parallel_sum_matches_the_sequential_sum ... ok
test result: ok. 6 passed; 0 failed; 0 ignored; 0 measured; 131 filtered out
16.3 How much faster do threads make it?
16.3.1 A billion elements
parallel_sum.rs times the same idea at a large scale. It sums a billion elements three times, as i32, i64,
and i128, once on one thread and once on several.
Listing 16.6 The parallel sum for i32 (lines 15 to 38). src/bin/parallel_sum.rs
use std::{
env,
thread,
time::{Duration, Instant},
};
fn par_i32(data: &[i32], n: usize) -> (i32, Duration) {
let t = Instant::now();
let sum = thread::scope(|s| {
let handles: Vec<_> = (0..n)
.map(|i| {
s.spawn(move || {
let chunk = data.len() / n;
let start = i * chunk;
let end = if i == n - 1 {
data.len()
} else {
start + chunk
};
data[start..end].iter().sum::<i32>()
})
})
.collect();
handles
.into_iter()
.map(|h| h.join().unwrap())
.sum::<i32>()
});
(sum, t.elapsed())
}This version divides the work by index instead of with chunks. Worker i sums the range from i * chunk to the
next boundary. The last worker takes everything to the end, so the leftover elements from the integer division
are not lost. The handles are collected before joining, for the reason in section 16.1.3.
Listing 16.7 The complete program. src/bin/parallel_sum.rs
use std::{
env,
thread,
time::{Duration, Instant},
};
const SIZE: usize = 1_000_000_000;
fn seq_i32(data: &[i32]) -> (i32, Duration) {
let t = Instant::now();
let sum = data.iter().sum::<i32>();
(sum, t.elapsed())
}
fn par_i32(data: &[i32], n: usize) -> (i32, Duration) {
let t = Instant::now();
let sum = thread::scope(|s| {
let handles: Vec<_> = (0..n)
.map(|i| {
s.spawn(move || {
let chunk = data.len() / n;
let start = i * chunk;
let end = if i == n - 1 {
data.len()
} else {
start + chunk
};
data[start..end].iter().sum::<i32>()
})
})
.collect();
handles
.into_iter()
.map(|h| h.join().unwrap())
.sum::<i32>()
});
(sum, t.elapsed())
}
fn seq_i64(data: &[i64]) -> (i64, Duration) {
let t = Instant::now();
let sum = data.iter().sum::<i64>();
(sum, t.elapsed())
}
fn par_i64(data: &[i64], n: usize) -> (i64, Duration) {
let t = Instant::now();
let sum = thread::scope(|s| {
let handles: Vec<_> = (0..n)
.map(|i| {
s.spawn(move || {
let chunk = data.len() / n;
let start = i * chunk;
let end = if i == n - 1 {
data.len()
} else {
start + chunk
};
data[start..end].iter().sum::<i64>()
})
})
.collect();
handles
.into_iter()
.map(|h| h.join().unwrap())
.sum::<i64>()
});
(sum, t.elapsed())
}
fn seq_i128(data: &[i128]) -> (i128, Duration) {
let t = Instant::now();
let sum = data.iter().sum::<i128>();
(sum, t.elapsed())
}
fn par_i128(data: &[i128], n: usize) -> (i128, Duration) {
let t = Instant::now();
let sum = thread::scope(|s| {
let handles: Vec<_> = (0..n)
.map(|i| {
s.spawn(move || {
let chunk = data.len() / n;
let start = i * chunk;
let end = if i == n - 1 {
data.len()
} else {
start + chunk
};
data[start..end].iter().sum::<i128>()
})
})
.collect();
handles
.into_iter()
.map(|h| h.join().unwrap())
.sum::<i128>()
});
(sum, t.elapsed())
}
fn main() {
let n: usize = env::args()
.nth(1)
.and_then(|s| s.parse().ok())
.unwrap_or(8);
{
let data: Vec<i32> = vec![1; SIZE];
let (s1, t1) = seq_i32(&data);
let (s2, t2) = par_i32(&data, n);
assert_eq!(s1, s2);
println!("i32 seq sum = {s1}, elapsed = {t1:?}");
println!("i32 par sum = {s2}, elapsed = {t2:?} (n = {n})");
}
{
let data: Vec<i64> = vec![1; SIZE];
let (s1, t1) = seq_i64(&data);
let (s2, t2) = par_i64(&data, n);
assert_eq!(s1, s2);
println!("i64 seq sum = {s1}, elapsed = {t1:?}");
println!("i64 par sum = {s2}, elapsed = {t2:?} (n = {n})");
}
{
let data: Vec<i128> = vec![1; SIZE];
let (s1, t1) = seq_i128(&data);
let (s2, t2) = par_i128(&data, n);
assert_eq!(s1, s2);
println!("i128 seq sum = {s1}, elapsed = {t1:?}");
println!("i128 par sum = {s2}, elapsed = {t2:?} (n = {n})");
}
}The thread count comes from the first command-line argument, and defaults to 8. Each block creates its vector, times both sums, and drops the vector at the closing brace before the next block starts.
WARNING: The three vectors take 4 GB, 8 GB, and 16 GB of memory. On a machine with 16 GB, the larger two do not fit next to everything else in memory. The operating system
then compresses pages or swaps them to disk. Lower
SIZE before you run this on a small machine.
Here is one run on the 16 GB laptop:
$ cargo run --release --bin parallel_sum
i32 seq sum = 1000000000, elapsed = 522.646188ms
i32 par sum = 1000000000, elapsed = 142.427408ms (n = 8)
i64 seq sum = 1000000000, elapsed = 7.750109265s
i64 par sum = 1000000000, elapsed = 2.591278253s (n = 8)
i128 seq sum = 1000000000, elapsed = 16.173010504s
i128 par sum = 1000000000, elapsed = 10.198263426s (n = 8)
Read these numbers with care. The i32 rows are the only ones that measure the sum. Eight threads made it about
3.7 times faster, not 8 times. The i64 and i128 rows are dominated by the memory pressure from the warning. Reading 8 GB should take
well under a second, so 7.7 seconds means pages were being brought back from compression or disk.
The runs were not stable, either. In an earlier run, the sequential i32 sum took 4.6 seconds instead of 0.52.
To check the i32 result, I wrote a separate test that fills the vector once and times each sum several times.
The sequential sum took 0.25 to 0.29 seconds, and eight threads took 0.08 seconds, about 3.2 times faster.
Why not 8 times? Summing is almost no work per byte, so the loop is limited by how fast memory delivers bytes, not by how fast cores add. One core cannot use all of the memory bandwidth, so several cores help, but only until the memory bus is full. The laptop also has four cores, not eight. Its eight hardware threads share those cores in pairs.
16.3.2 Amdahl’s law
Most programs also contain work that cannot be split: reading input, starting threads, combining results. Amdahl’s law states how much that serial part limits the speedup.
Suppose a fraction s of the run time is serial, and p processors share the rest perfectly. The run time
becomes s + (1 - s) / p of the original, and the speedup is one divided by that:
speedup(s, p) = 1 / (s + (1 - s) / p)
Work through one case by hand: s = 0.5 and p = 4. The run time is 0.5 + 0.5 / 4 = 0.625 of the original,
so the speedup is 1 / 0.625 = 1.6.
Listing 16.8 The complete program. src/bin/concurrency_amdahl.rs
//! Amdahl's law: the serial fraction of the work caps the speedup from adding
//! processors. A program that is 50 percent serial cannot go faster than 2x, no
//! matter how many cores it gets.
//!
//! Run with: cargo run --bin concurrency_amdahl
/// Speedup with `processors` when `serial` is the non-parallel fraction.
fn speedup(serial: f64, processors: f64) -> f64 {
1.0 / (serial + (1.0 - serial) / processors)
}
fn main() {
let processor_counts = [1.0, 2.0, 4.0, 8.0, 16.0];
let serial_fractions = [0.0, 0.05, 0.10, 0.25, 0.50];
print!("{:>8} |", "serial");
for processors in processor_counts {
print!(" {processors:>6.0}");
}
println!(" <- processors");
println!("{:->9}+{}", "", "-".repeat(7 * processor_counts.len()));
for serial in serial_fractions {
print!("{serial:>8.2} |");
for processors in processor_counts {
print!(" {:>6.2}", speedup(serial, processors));
}
println!();
}
println!("\nlimit as processors grow: 1 / serial");
for serial in serial_fractions {
let limit = if serial == 0.0 {
f64::INFINITY
} else {
1.0 / serial
};
println!(" {serial:>4.2} -> {limit:>6.1}");
}
assert!((speedup(0.0, 4.0) - 4.0).abs() < 1e-9);
assert!((speedup(0.5, 4.0) - 1.6).abs() < 1e-9);
assert!(speedup(0.5, 16.0) < 1.9, "2x is the ceiling for 50% serial");
println!("\nall checks passed");
}$ cargo run --bin concurrency_amdahl
serial | 1 2 4 8 16 <- processors
---------+-----------------------------------
0.00 | 1.00 2.00 4.00 8.00 16.00
0.05 | 1.00 1.90 3.48 5.93 9.14
0.10 | 1.00 1.82 3.08 4.71 6.40
0.25 | 1.00 1.60 2.29 2.91 3.37
0.50 | 1.00 1.33 1.60 1.78 1.88
limit as processors grow: 1 / serial
0.00 -> inf
0.05 -> 20.0
0.10 -> 10.0
0.25 -> 4.0
0.50 -> 2.0
all checks passed
As p grows, (1 - s) / p shrinks toward zero, and the speedup approaches 1 / s. A program that is 5% serial
can never run more than 20 times faster, however many cores it gets.
16.4 A spin lock
A Mutex from the standard library lets one thread at a time use a value. This section builds a simpler lock
from one atomic bool, to show what a lock must do.
A spin lock waits by looping. A thread that wants the lock checks it over and over until it is free. This
works well when the lock is held for a very short time. When it is held longer, the waiting threads waste whole
cores doing nothing useful. std::sync::Mutex instead asks the operating system to put a waiting thread to sleep.
16.4.1 The type
Listing 16.9 The lock, the guard, and new (lines 24 to 48). src/problems/spin_lock.rs
use std::{
cell::UnsafeCell,
hint::spin_loop,
ops::{Deref, DerefMut},
sync::atomic::{AtomicBool, Ordering},
};
/// A mutual-exclusion lock that busy-waits instead of parking.
pub struct SpinLock<T> {
locked: AtomicBool,
value: UnsafeCell<T>,
}
// SAFETY: `value` is only accessed while `locked` is held, and the acquire/release
// ordering on `locked` makes a hand-off between threads well defined. Sharing
// `&SpinLock` across threads therefore only requires the protected value to be
// `Send`, which lets one thread move data in and another move it out.
unsafe impl<T: Send> Sync for SpinLock<T> {}
/// Grants access to the protected value and releases the lock when dropped.
pub struct SpinGuard<'a, T> {
lock: &'a SpinLock<T>,
}
impl<T> SpinLock<T> {
/// Create an unlocked spin lock.
pub const fn new(value: T) -> Self {
Self {
locked: AtomicBool::new(false),
value: UnsafeCell::new(value),
}
}
// ...
}SpinLock<T> holds a flag, locked, and the protected value inside an UnsafeCell<T>. UnsafeCell is the one
type through which Rust allows changing data behind a shared reference. RefCell, Mutex, and the atomic types
are all built on it. Its get() method returns a raw pointer, *mut T. Using that pointer requires unsafe,
and the code must make sure no two threads use it at once.
UnsafeCell is not Sync, so SpinLock would not be Sync either. The line unsafe impl<T: Send> Sync for SpinLock<T> {} declares that it is. The compiler cannot check this claim, which is why the implementation needs
unsafe. The SAFETY comment above it gives the reason the claim holds: the value is used only while the lock is
held.
The bound is T: Send, not T: Sync. Only one thread uses the value at a time, but different threads use it one
after another, so the value effectively moves between threads.
SpinGuard is returned by lock. It holds a reference to the lock, gives access to the value, and releases the
lock when it is dropped.
16.4.2 Taking and releasing the lock
Listing 16.10 lock, try_lock, and unlock (lines 52 to 87).
impl<T> SpinLock<T> {
// ...
/// Acquire the lock, spinning until it is free.
pub fn lock(&self) -> SpinGuard<'_, T> {
loop {
if self
.locked
.compare_exchange_weak(false, true, Ordering::Acquire, Ordering::Relaxed)
.is_ok()
{
return SpinGuard { lock: self };
}
// Spin on a relaxed load so the contention loop does not keep
// bouncing the cache line with exclusive accesses.
while self.locked.load(Ordering::Relaxed) {
spin_loop();
}
}
}
/// Acquire the lock only if it is free right now.
pub fn try_lock(&self) -> Option<SpinGuard<'_, T>> {
self.locked
.compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
.ok()
.map(|_| SpinGuard { lock: self })
}
/// Whether the lock is currently held.
pub fn is_locked(&self) -> bool {
self.locked.load(Ordering::Relaxed)
}
fn unlock(&self) {
self.locked
.store(false, Ordering::Release);
}
}Taking the lock must be one atomic step: check that it is free and mark it taken, with no other thread slipping
in between. compare_exchange(false, true, ...) does that. If the flag is false, it sets it to true and
returns Ok. If it is already true, it changes nothing and returns Err.
lock uses compare_exchange_weak, which may fail even when the flag is false. The weak form is cheaper on
some CPUs, and inside a loop a false failure costs one more pass.
When the exchange fails, the thread waits in an inner loop that only reads the flag with a Relaxed load. A read
lets the cache line stay shared between the waiting cores. A compare-exchange must take the line for writing, as
section 15.4 described, so repeating it would bounce the line between cores. When the read sees false, the outer
loop tries the exchange again. spin_loop() tells the CPU that this is a wait loop. The CPU can then save power and let the other
hardware thread on the core run.
try_lock makes one attempt and returns None if the lock is taken.
16.4.3 Acquire and release
The orderings in lock and unlock make the lock correct (figure 16.5).
unlock pairs with the acquire exchange in the next lock.unlockstoresfalsewithRelease. Every write the thread made before the store happens before the store, as far as other threads can tell. That includes its changes to the value.lockexchanges withAcquire. Every read the thread makes after the exchange happens after it.
When B’s acquire reads the false that A’s release wrote, the two operations synchronize. Everything A wrote
before unlocking is then visible to B after locking. With Relaxed in both places, B could take the lock and
still read an old copy of the value.
lock from listing 16.10. B stays on the CPU while it spins. The Release store in unlock pairs with B's Acquire exchange.The second ordering argument to compare_exchange, Relaxed, applies when the exchange fails. A failed attempt
does not take the lock, so it needs no ordering.
16.4.4 The guard
Listing 16.11 The guard's traits (lines 96 to 116).
impl<T> Drop for SpinGuard<'_, T> {
fn drop(&mut self) {
self.lock.unlock();
}
}
impl<T> Deref for SpinGuard<'_, T> {
type Target = T;
fn deref(&self) -> &T {
// SAFETY: holding the guard means this thread has exclusive access.
unsafe { &*self.lock.value.get() }
}
}
impl<T> DerefMut for SpinGuard<'_, T> {
fn deref_mut(&mut self) -> &mut T {
// SAFETY: holding the guard means this thread has exclusive access.
unsafe { &mut *self.lock.value.get() }
}
}The guard implements three traits:
Dropreleases the lock. The lock is released when the guard goes out of scope, even if the code between panics. This pattern, tying a resource’s release to a value’s drop, is called RAII.Dereflets*guardand method calls on the guard reach the value, as&T.DerefMutgives&mut T, so*lock.lock() += 1changes the value.
Each method turns the raw pointer from UnsafeCell::get into a reference. That is sound only because holding the
guard means holding the lock.
WARNING: SpinGuard has a gap. Its auto traits come from its field &SpinLock<T>, so the guard is Sync
whenever T is Send. Take a type such as Cell<u64>, which is Send but not Sync. Safe code can share
&SpinGuard<Cell<u64>> between two scoped threads, and both can call Cell::set at once. That is a data race. I compiled
such a program to confirm it. std::sync::MutexGuard avoids this by being Sync only when T is Sync.
Exercise 3 asks you to close the gap.
Listing 16.12 The complete file, with tests. src/problems/spin_lock.rs
//! A spin lock built from a single `AtomicBool`.
//!
//! The loop is short. The memory ordering and the ownership argument are what
//! make the unsafe block sound.
//!
//! - `lock`/`try_lock` use a compare-exchange with `Acquire` ordering, so the
//! critical section observes every write the previous holder made before it
//! released.
//! - `unlock` uses a `Release` store, so writes inside the critical section are
//! visible to whoever acquires next.
//! - The guard derefs to `&T`/`&mut T`. Mutual exclusion is what makes handing out
//! `&mut T` sound, and that is exactly what the lock guarantees.
//!
//! A spin lock is the wrong choice when the critical section can block or is
//! long: every waiting thread burns a core. Use it for very short sections on
//! dedicated cores, and prefer `std::sync::Mutex` (which parks the thread) in
//! general code.
use std::{
cell::UnsafeCell,
hint::spin_loop,
ops::{Deref, DerefMut},
sync::atomic::{AtomicBool, Ordering},
};
/// A mutual-exclusion lock that busy-waits instead of parking.
pub struct SpinLock<T> {
locked: AtomicBool,
value: UnsafeCell<T>,
}
// SAFETY: `value` is only accessed while `locked` is held, and the acquire/release
// ordering on `locked` makes a hand-off between threads well defined. Sharing
// `&SpinLock` across threads therefore only requires the protected value to be
// `Send`, which lets one thread move data in and another move it out.
unsafe impl<T: Send> Sync for SpinLock<T> {}
/// Grants access to the protected value and releases the lock when dropped.
pub struct SpinGuard<'a, T> {
lock: &'a SpinLock<T>,
}
impl<T> SpinLock<T> {
/// Create an unlocked spin lock.
pub const fn new(value: T) -> Self {
Self {
locked: AtomicBool::new(false),
value: UnsafeCell::new(value),
}
}
/// Acquire the lock, spinning until it is free.
pub fn lock(&self) -> SpinGuard<'_, T> {
loop {
if self
.locked
.compare_exchange_weak(false, true, Ordering::Acquire, Ordering::Relaxed)
.is_ok()
{
return SpinGuard { lock: self };
}
// Spin on a relaxed load so the contention loop does not keep
// bouncing the cache line with exclusive accesses.
while self.locked.load(Ordering::Relaxed) {
spin_loop();
}
}
}
/// Acquire the lock only if it is free right now.
pub fn try_lock(&self) -> Option<SpinGuard<'_, T>> {
self.locked
.compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
.ok()
.map(|_| SpinGuard { lock: self })
}
/// Whether the lock is currently held.
pub fn is_locked(&self) -> bool {
self.locked.load(Ordering::Relaxed)
}
fn unlock(&self) {
self.locked
.store(false, Ordering::Release);
}
}
impl<T: Default> Default for SpinLock<T> {
fn default() -> Self {
Self::new(T::default())
}
}
impl<T> Drop for SpinGuard<'_, T> {
fn drop(&mut self) {
self.lock.unlock();
}
}
impl<T> Deref for SpinGuard<'_, T> {
type Target = T;
fn deref(&self) -> &T {
// SAFETY: holding the guard means this thread has exclusive access.
unsafe { &*self.lock.value.get() }
}
}
impl<T> DerefMut for SpinGuard<'_, T> {
fn deref_mut(&mut self) -> &mut T {
// SAFETY: holding the guard means this thread has exclusive access.
unsafe { &mut *self.lock.value.get() }
}
}
#[cfg(test)]
mod tests {
use std::{sync::Arc, thread};
use super::*;
#[test]
fn serializes_concurrent_increments() {
let lock = Arc::new(SpinLock::new(0usize));
let mut handles = Vec::new();
for _ in 0..8 {
let lock = Arc::clone(&lock);
handles.push(thread::spawn(move || {
for _ in 0..1_000 {
*lock.lock() += 1;
}
}));
}
for handle in handles {
handle.join().expect("worker panicked");
}
assert_eq!(*lock.lock(), 8_000);
}
#[test]
fn try_lock_reports_contention() {
let lock = SpinLock::new(1u32);
let guard = lock.lock();
assert!(lock.try_lock().is_none());
assert!(lock.is_locked());
drop(guard);
assert!(lock.try_lock().is_some());
assert!(!lock.is_locked());
}
#[test]
fn guard_gives_mutable_access() {
let lock = SpinLock::new(vec![1, 2, 3]);
{
let mut guard = lock.lock();
guard.push(4);
}
assert_eq!(*lock.lock(), vec![1, 2, 3, 4]);
}
}The first test shares the lock between 8 threads with an Arc, and each adds 1,000. The total must be 8,000.
$ cargo test --lib problems::spin_lock
running 3 tests
test problems::spin_lock::tests::try_lock_reports_contention ... ok
test problems::spin_lock::tests::guard_gives_mutable_access ... ok
test problems::spin_lock::tests::serializes_concurrent_increments ... ok
test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 134 filtered out
16.5 A counting semaphore
A lock lets one thread at a time proceed. A semaphore lets up to N threads proceed at once. It holds a count of permits. A thread takes a permit before it starts, and returns it when it finishes. When no permit is left, the next thread waits (figure 16.6).
Semaphores limit how much of something runs at once. Examples are open connections to a database, requests to a server, and open files.
16.5.1 Waiting without spinning
A waiting thread should sleep, not spin, because the wait may be long. A condition variable, Condvar, lets a
thread sleep until another thread signals it. It always works together with a Mutex:
- The waiting thread locks the mutex and checks the condition.
- If the condition is false, it calls
condvar.wait(guard). That call releases the mutex and puts the thread to sleep, in one step. - Another thread locks the mutex, changes the state, and calls
notify_one(). - The sleeping thread wakes up, holding the mutex again, and checks the condition again.
The check in step 4 is needed for two reasons. Another thread may have taken the permit first. And a thread can
sometimes wake up with no notification at all, which is called a spurious wakeup. So the wait always sits in a
while loop, not an if.
Listing 16.13 The state, the guard, and acquire_guard. src/problems/semaphore.rs
use std::{
sync::{Condvar, Mutex},
time::Duration,
};
struct State {
available: usize,
}
/// A counting semaphore. Cloning is not provided; share one behind `Arc`.
pub struct Semaphore {
state: Mutex<State>,
released: Condvar,
}
/// Releases one permit when dropped.
pub struct SemaphoreGuard<'a> {
semaphore: &'a Semaphore,
}
impl Drop for SemaphoreGuard<'_> {
fn drop(&mut self) {
self.semaphore.release();
}
}
impl Semaphore {
/// Create a semaphore with `permits` permits available.
pub fn new(permits: usize) -> Self {
Self {
state: Mutex::new(State { available: permits }),
released: Condvar::new(),
}
}
/// Block until a permit is available.
pub fn acquire(&self) {
let _guard = self.acquire_guard();
}
/// Block until a permit is available, returning a guard that releases it.
pub fn acquire_guard(&self) -> SemaphoreGuard<'_> {
let mut state = self
.state
.lock()
.expect("semaphore mutex poisoned");
while state.available == 0 {
state = self
.released
.wait(state)
.expect("semaphore mutex poisoned");
}
state.available -= 1;
SemaphoreGuard { semaphore: self }
}
// ...
}acquire_guard returns a SemaphoreGuard. The guard holds only a reference to its semaphore, and its Drop calls release. A permit goes back even if
the holder returns early or panics.
acquire_guard follows the four steps. self.released.wait(state) takes the guard, sleeps, and returns a new
guard when the thread wakes. When the loop ends, a permit is free and the thread holds the mutex, so it takes the
permit by subtracting 1.
The permit is returned by a guard, the same RAII pattern as in section 16.4.
Listing 16.14 The other ways to acquire, and release (lines 47 to 103).
impl Semaphore {
// ...
/// Take a permit only if one is free right now.
pub fn try_acquire(&self) -> Option<SemaphoreGuard<'_>> {
let mut state = self
.state
.lock()
.expect("semaphore mutex poisoned");
if state.available == 0 {
None
} else {
state.available -= 1;
Some(SemaphoreGuard { semaphore: self })
}
}
/// Take a permit, waiting at most `timeout`.
pub fn try_acquire_timeout(&self, timeout: Duration) -> Option<SemaphoreGuard<'_>> {
let mut state = self
.state
.lock()
.expect("semaphore mutex poisoned");
if state.available > 0 {
state.available -= 1;
return Some(SemaphoreGuard { semaphore: self });
}
let (mut state, _timed_out) = self
.released
.wait_timeout(state, timeout)
.expect("semaphore mutex poisoned");
if state.available == 0 {
None
} else {
state.available -= 1;
Some(SemaphoreGuard { semaphore: self })
}
}
/// Permits currently free. Useful for metrics, not for control flow.
pub fn available_permits(&self) -> usize {
self.state
.lock()
.expect("semaphore mutex poisoned")
.available
}
fn release(&self) {
let mut state = self
.state
.lock()
.expect("semaphore mutex poisoned");
state.available += 1;
self.released.notify_one();
}
}
/// Releases one permit when dropped.
pub struct SemaphoreGuard<'a> {
semaphore: &'a Semaphore,
}
impl Drop for SemaphoreGuard<'_> {
fn drop(&mut self) {
self.semaphore.release();
}
}release adds a permit and wakes one sleeping thread with notify_one. It is private, and only the guard’s Drop
calls it. A caller therefore cannot release a permit it never took.
Two methods have problems:
acquirecallsacquire_guardand stores the guard in_guard. The guard is dropped whenacquirereturns, so the permit is released at once. A caller ofacquirenever actually holds a permit.acquire_guardis the method to use.try_acquire_timeoutwaits once. If it wakes early, from a spurious wakeup or because another thread took the permit first, it returnsNonebefore the timeout has passed. A correct version loops, and waits again for the time remaining. Exercise 4 asks you to write it.
Listing 16.15 The complete file, with tests. src/problems/semaphore.rs
//! Counting semaphore built from `Mutex` + `Condvar`.
//!
//! A semaphore bounds how many callers may proceed at once without protecting any
//! data. It is the primitive behind connection pools, per-account limits, and
//! bounded fan-out in a client library or command-line tool.
//!
//! The RAII guard returns the permit on drop, so a panicking worker cannot leak a
//! permit and slowly deadlock the pool.
use std::{
sync::{Condvar, Mutex},
time::Duration,
};
struct State {
available: usize,
}
/// A counting semaphore. Cloning is not provided; share one behind `Arc`.
pub struct Semaphore {
state: Mutex<State>,
released: Condvar,
}
impl Semaphore {
/// Create a semaphore with `permits` permits available.
pub fn new(permits: usize) -> Self {
Self {
state: Mutex::new(State { available: permits }),
released: Condvar::new(),
}
}
/// Block until a permit is available.
pub fn acquire(&self) {
let _guard = self.acquire_guard();
}
/// Block until a permit is available, returning a guard that releases it.
pub fn acquire_guard(&self) -> SemaphoreGuard<'_> {
let mut state = self
.state
.lock()
.expect("semaphore mutex poisoned");
while state.available == 0 {
state = self
.released
.wait(state)
.expect("semaphore mutex poisoned");
}
state.available -= 1;
SemaphoreGuard { semaphore: self }
}
/// Take a permit only if one is free right now.
pub fn try_acquire(&self) -> Option<SemaphoreGuard<'_>> {
let mut state = self
.state
.lock()
.expect("semaphore mutex poisoned");
if state.available == 0 {
None
} else {
state.available -= 1;
Some(SemaphoreGuard { semaphore: self })
}
}
/// Take a permit, waiting at most `timeout`.
pub fn try_acquire_timeout(&self, timeout: Duration) -> Option<SemaphoreGuard<'_>> {
let mut state = self
.state
.lock()
.expect("semaphore mutex poisoned");
if state.available > 0 {
state.available -= 1;
return Some(SemaphoreGuard { semaphore: self });
}
let (mut state, _timed_out) = self
.released
.wait_timeout(state, timeout)
.expect("semaphore mutex poisoned");
if state.available == 0 {
None
} else {
state.available -= 1;
Some(SemaphoreGuard { semaphore: self })
}
}
/// Permits currently free. Useful for metrics, not for control flow.
pub fn available_permits(&self) -> usize {
self.state
.lock()
.expect("semaphore mutex poisoned")
.available
}
fn release(&self) {
let mut state = self
.state
.lock()
.expect("semaphore mutex poisoned");
state.available += 1;
self.released.notify_one();
}
}
/// Releases one permit when dropped.
pub struct SemaphoreGuard<'a> {
semaphore: &'a Semaphore,
}
impl Drop for SemaphoreGuard<'_> {
fn drop(&mut self) {
self.semaphore.release();
}
}
#[cfg(test)]
mod tests {
use std::{
sync::{
Arc,
atomic::{AtomicUsize, Ordering},
},
thread,
};
use super::*;
#[test]
fn never_exceeds_the_permit_count() {
let permits = 4;
let semaphore = Arc::new(Semaphore::new(permits));
let active = Arc::new(AtomicUsize::new(0));
let peak = Arc::new(AtomicUsize::new(0));
let mut handles = Vec::new();
for _ in 0..32 {
let semaphore = Arc::clone(&semaphore);
let active = Arc::clone(&active);
let peak = Arc::clone(&peak);
handles.push(thread::spawn(move || {
let _guard = semaphore.acquire_guard();
let now = active.fetch_add(1, Ordering::SeqCst) + 1;
peak.fetch_max(now, Ordering::SeqCst);
thread::sleep(Duration::from_millis(1));
active.fetch_sub(1, Ordering::SeqCst);
}));
}
for handle in handles {
handle.join().expect("worker panicked");
}
assert!(peak.load(Ordering::SeqCst) <= permits);
assert_eq!(semaphore.available_permits(), permits);
}
#[test]
fn guard_returns_the_permit_on_drop() {
let semaphore = Semaphore::new(1);
{
let _guard = semaphore.acquire_guard();
assert_eq!(semaphore.available_permits(), 0);
}
assert_eq!(semaphore.available_permits(), 1);
}
#[test]
fn try_acquire_reports_contention() {
let semaphore = Semaphore::new(1);
let first = semaphore
.try_acquire()
.expect("first permit is free");
assert!(semaphore.try_acquire().is_none());
drop(first);
assert!(semaphore.try_acquire().is_some());
}
#[test]
fn timeout_gives_up_when_no_permit_arrives() {
let semaphore = Semaphore::new(0);
assert!(
semaphore
.try_acquire_timeout(Duration::from_millis(5))
.is_none()
);
}
}The test never_exceeds_the_permit_count starts 32 threads on a semaphore with 4 permits. Each thread counts
itself in active while it holds a permit, and records the highest count ever seen in peak with fetch_max.
The peak must not exceed 4.
$ cargo test --lib problems::semaphore
running 4 tests
test problems::semaphore::tests::guard_returns_the_permit_on_drop ... ok
test problems::semaphore::tests::try_acquire_reports_contention ... ok
test problems::semaphore::tests::timeout_gives_up_when_no_permit_arrives ... ok
test problems::semaphore::tests::never_exceeds_the_permit_count ... ok
test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 133 filtered out
16.6 When a lock holder panics
A lock protects data that must stay consistent. A bank transfer, for example, takes money from one account and adds it to another. Between those two steps, the data is briefly inconsistent. The lock hides that moment from other threads.
What if the thread panics between the two steps? The guard is dropped during the panic, so the lock is released.
The data, though, stays half-changed. The standard Mutex records that this happened, and the mutex becomes
poisoned (figure 16.7). Every later lock() returns Err(PoisonError) instead of Ok(guard).
mutex_poisoning.rs.The error does not hide the data. PoisonError::into_inner() returns the guard anyway, for code that can check
and repair the data.
Listing 16.16 The ledger and the recovery function. src/bin/mutex_poisoning.rs
use std::{
error::Error,
sync::{Arc, Mutex, MutexGuard, PoisonError},
thread,
};
/// Shared ledger state. In a real system this would be behind a transactional
/// store; here it is plain in-memory state to illustrate lock poisoning.
#[derive(Debug, Clone)]
struct Ledger {
cash_on_hand: i64,
stock_value: i64,
transaction_id: u32,
is_settled: bool,
}
/// Recovers from a poisoned mutex by taking ownership of the guard that the
/// panicked thread left behind.
///
/// `PoisonError::into_inner()` gives us the `MutexGuard`, which means we still
/// hold the lock while repairing state. The guard is dropped when this function
/// returns.
fn recover_ledger(poisoned: PoisonError<MutexGuard<'_, Ledger>>) -> Ledger {
// into_inner() returns the MutexGuard. We own the lock exclusively even
// though the previous holder panicked.
let mut guard = poisoned.into_inner();
println!("[Reconciliation] Physical audit reveals a cash discrepancy.");
println!(
"[Reconciliation] Digital cash: {} | Corrected cash: {}",
guard.cash_on_hand,
guard.cash_on_hand - 200
);
// Correct the data while we still hold the guard.
guard.cash_on_hand -= 200;
guard.is_settled = true;
guard.transaction_id += 1;
println!("[Reconciliation] Ledger corrected. Transaction marked failed and closed.");
// Clone the inner data to return an owned value. The guard drops here and
// releases the lock.
guard.clone()
}Ledger is the shared state: the cash, the stock value, a transaction counter, and a flag that says whether
the books balance. A panic between two of its updates leaves them inconsistent.
recover_ledger receives the PoisonError and takes the guard out of it. It holds the lock while it corrects the
ledger. It returns a clone of the corrected data, and the guard is dropped at the end of the function.
Listing 16.17 The complete program. src/bin/mutex_poisoning.rs
// ============================================================================
// Mutex poisoning: failed transaction and audit recovery
//
// Ported from the Rust playground repo (`scam_transaction.rs`).
//
// Story:
// A branch executes a financial transaction that mutates shared state under
// a `Mutex`. Mid-transaction the thread panics. In Rust, a `Mutex` whose
// guard was held during a panic becomes **poisoned**. Future `lock()` calls
// return `Err(PoisonError)` instead of silently letting another thread
// observe an inconsistent partial update.
//
// Where this appears in practice:
// - Controllers/reconcilers mutate shared state; a panic while holding a lock
// is a real failure mode in worker pools, caches, and rate limiters.
// - Rust chooses fail-fast over data corruption: a poisoned mutex tells
// callers that the protected invariant may be broken. The caller must
// decide whether to abort, recover, or reset state.
// - `PoisonError::into_inner()` returns the `MutexGuard`, so a recovery path
// can inspect/correct the last known state without waiting for the lock to
// be released by the panicked thread.
//
// Run:
// cargo run --bin mutex_poisoning
// ============================================================================
use std::{
error::Error,
sync::{Arc, Mutex, MutexGuard, PoisonError},
thread,
};
/// Shared ledger state. In a real system this would be behind a transactional
/// store; here it is plain in-memory state to illustrate lock poisoning.
#[derive(Debug, Clone)]
struct Ledger {
cash_on_hand: i64,
stock_value: i64,
transaction_id: u32,
is_settled: bool,
}
impl Ledger {
fn new() -> Self {
Ledger {
cash_on_hand: 10_000,
stock_value: 5_000,
transaction_id: 0,
is_settled: false,
}
}
}
/// Recovers from a poisoned mutex by taking ownership of the guard that the
/// panicked thread left behind.
///
/// `PoisonError::into_inner()` gives us the `MutexGuard`, which means we still
/// hold the lock while repairing state. The guard is dropped when this function
/// returns.
fn recover_ledger(poisoned: PoisonError<MutexGuard<'_, Ledger>>) -> Ledger {
// into_inner() returns the MutexGuard. We own the lock exclusively even
// though the previous holder panicked.
let mut guard = poisoned.into_inner();
println!("[Reconciliation] Physical audit reveals a cash discrepancy.");
println!(
"[Reconciliation] Digital cash: {} | Corrected cash: {}",
guard.cash_on_hand,
guard.cash_on_hand - 200
);
// Correct the data while we still hold the guard.
guard.cash_on_hand -= 200;
guard.is_settled = true;
guard.transaction_id += 1;
println!("[Reconciliation] Ledger corrected. Transaction marked failed and closed.");
// Clone the inner data to return an owned value. The guard drops here and
// releases the lock.
guard.clone()
}
fn main() -> Result<(), Box<dyn Error>> {
let ledger = Arc::new(Mutex::new(Ledger::new()));
// Thread A: the faulty transaction. It mutates one side of the ledger and
// then panics before marking the transaction settled.
let ledger_a = Arc::clone(&ledger);
let handle_a = thread::spawn(move || {
// SAFETY-free normal code: if the mutex is already poisoned, unwrap()
// panics; in a controlled demo this is the first lock so it is fine.
let mut guard = ledger_a.lock().unwrap();
println!(
"[Branch A] Transaction {} commenced.",
guard.transaction_id + 1
);
guard.stock_value -= 1_000;
println!(
"[Branch A] Stock deducted. New value: {}",
guard.stock_value
);
guard.cash_on_hand += 1_200;
println!(
"[Branch A] Cash updated. New balance: {}",
guard.cash_on_hand
);
// Panic while `guard` is still alive -> the Mutex becomes poisoned.
panic!("[Branch A] Settlement failure: arithmetic overflow in reconciliation.");
});
// A panic in a spawned thread does not abort the process unless the panic
// strategy is set to abort. `join()` returns Err, which is expected here.
let _ = handle_a.join();
// Thread B: the forensic audit. It sees Err(PoisonError), inspects the
// recovered state, and repairs the ledger.
let ledger_b = Arc::clone(&ledger);
let handle_b = thread::spawn(move || {
println!("\n[Audit] Attempting to access the ledger...");
match ledger_b.lock() {
Ok(guard) => {
println!("[Audit] Ledger is not poisoned: {:?}", *guard);
}
Err(poisoned) => {
println!("[Audit] Mutex is poisoned. Recovered state before correction:");
println!("[Audit] {:?}", *poisoned.get_ref());
let corrected = recover_ledger(poisoned);
println!("[Audit] Corrected ledger: {:?}", corrected);
}
}
});
handle_b.join().unwrap();
// Important: recovering the data does NOT clear the poison bit on the
// Mutex itself. Subsequent lock() calls still return Err(PoisonError) so
// callers can decide whether the repaired state is trustworthy.
// Uncommenting the next lines would panic:
//
// let _guard = ledger.lock().unwrap();
Ok(())
}
// ============================================================================
// Tests (cargo test --bin mutex_poisoning)
// ============================================================================
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ledger_starts_in_expected_state() {
let ledger = Ledger::new();
assert_eq!(ledger.cash_on_hand, 10_000);
assert_eq!(ledger.stock_value, 5_000);
assert!(!ledger.is_settled);
}
#[test]
fn recovery_repairs_a_poisoned_ledger() {
let ledger = Arc::new(Mutex::new(Ledger::new()));
// Poison the mutex from a helper thread, exactly like the demo.
let ledger_a = Arc::clone(&ledger);
let handle = thread::spawn(move || {
let _guard = ledger_a.lock().unwrap();
panic!("boom");
});
let _ = handle.join();
let lock_result = ledger.lock();
assert!(lock_result.is_err(), "mutex should be poisoned");
let recovered = recover_ledger(lock_result.unwrap_err());
assert!(recovered.is_settled);
assert_eq!(recovered.transaction_id, 1);
assert_eq!(recovered.cash_on_hand, 10_000 - 200);
}
}In main, thread A changes the stock and the cash and then panics before marking the transaction settled. The
panic does not stop the process. It ends only thread A, and handle_a.join() returns an Err, which main
ignores on purpose with let _ =.
Thread B then locks the ledger. match on the result separates the two cases. poisoned.get_ref() shows the data
without taking the guard, and recover_ledger repairs it.
$ cargo run --bin mutex_poisoning
[Branch A] Transaction 1 commenced.
[Branch A] Stock deducted. New value: 4000
[Branch A] Cash updated. New balance: 11200
thread '<unnamed>' (5795284) panicked at src/bin/mutex_poisoning.rs:109:9:
[Branch A] Settlement failure: arithmetic overflow in reconciliation.
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
[Audit] Attempting to access the ledger...
[Audit] Mutex is poisoned. Recovered state before correction:
[Audit] Ledger { cash_on_hand: 11200, stock_value: 4000, transaction_id: 0, is_settled: false }
[Reconciliation] Physical audit reveals a cash discrepancy.
[Reconciliation] Digital cash: 11200 | Corrected cash: 11000
[Reconciliation] Ledger corrected. Transaction marked failed and closed.
[Audit] Corrected ledger: Ledger { cash_on_hand: 11000, stock_value: 4000, transaction_id: 1, is_settled: true }
The panic message goes to standard error, so in a terminal it can appear between the other lines. The number in parentheses is the thread’s ID and changes on every run.
Recovering the data does not clear the poison flag. A later lock() on the same mutex still returns Err, as the
comment at the end of main notes. Rust 1.77 added Mutex::clear_poison for code that has repaired the data and
wants to say so.
16.7 Deadlock
A deadlock happens when threads wait for each other in a circle, so none can continue. The classic case needs two locks. Thread 1 holds lock A and waits for lock B. Thread 2 holds lock B and waits for lock A. Neither will ever release what it holds.
To analyze a deadlock, draw a wait-for graph. Each thread is a node. An edge from T0 to T1 means T0 is waiting for a lock that T1 holds. The threads are deadlocked exactly when the graph has a cycle (figure 16.8). Finding a cycle is the three-state DFS from section 12.6.
has_deadlock takes the edges as (waiter, holder) pairs. It first turns them into an adjacency list:
/// Iterative-friendly depth-first search over a small adjacency list.
fn has_deadlock(waits_for: &[(usize, usize)]) -> bool {
let Some(highest) = waits_for
.iter()
.flat_map(|(a, b)| [*a, *b])
.max()
else {
return false;
};
let mut edges = vec![Vec::new(); highest + 1];
for (waiter, holder) in waits_for {
edges[*waiter].push(*holder);
}
// ...
}It finds the largest thread number with a let ... else, so an empty list returns false at once. The
adjacency list then has one entry per thread, holding the threads it waits for.
The search is the DFS of section 12.6. The three states are plain numbers: 0 for unvisited, 1 for on the current path, and 2 for fully explored:
fn has_deadlock(waits_for: &[(usize, usize)]) -> bool {
// ...
// 1 = on the current path, 2 = fully explored.
fn visit(node: usize, edges: &[Vec<usize>], state: &mut [u8]) -> bool {
match state[node] {
1 => return true,
2 => return false,
_ => state[node] = 1,
}
for &next in &edges[node] {
if visit(next, edges, state) {
return true;
}
}
state[node] = 2;
false
}
let mut state = vec![0u8; edges.len()];
(0..edges.len()).any(|node| visit(node, &edges, &mut state))
}visit returns true as soon as it reaches a node in state 1. That node is on the path that led here, so the
edge closes a cycle. A node in state 2 was explored already and is known to lead to no cycle. any starts a
search from every node, so a cycle among threads that no other thread waits on is found too.
main checks three graphs. The third, (0, 0), is a thread waiting for itself, as happens when a thread tries
to lock a Mutex it already holds.
Listing 16.18 The complete program. src/bin/concurrency_deadlock.rs
//! Deadlock detection: a deadlock exists exactly when the wait-for graph has a
//! cycle. Each edge `(waiter, holder)` means the first task is blocked on the
//! second.
//!
//! Run with: cargo run --bin concurrency_deadlock
/// Iterative-friendly depth-first search over a small adjacency list.
fn has_deadlock(waits_for: &[(usize, usize)]) -> bool {
let Some(highest) = waits_for
.iter()
.flat_map(|(a, b)| [*a, *b])
.max()
else {
return false;
};
let mut edges = vec![Vec::new(); highest + 1];
for (waiter, holder) in waits_for {
edges[*waiter].push(*holder);
}
// 1 = on the current path, 2 = fully explored.
fn visit(node: usize, edges: &[Vec<usize>], state: &mut [u8]) -> bool {
match state[node] {
1 => return true,
2 => return false,
_ => state[node] = 1,
}
for &next in &edges[node] {
if visit(next, edges, state) {
return true;
}
}
state[node] = 2;
false
}
let mut state = vec![0u8; edges.len()];
(0..edges.len()).any(|node| visit(node, &edges, &mut state))
}
fn main() {
let acyclic = [(0, 1), (1, 2), (0, 2)];
let cycle = [(0, 1), (1, 2), (2, 0)];
let self_wait = [(0, 0)];
for (label, graph) in [
("0->1, 1->2, 0->2", &acyclic[..]),
("0->1, 1->2, 2->0", &cycle[..]),
("0->0", &self_wait[..]),
] {
println!("{label:<20} deadlock: {}", has_deadlock(graph));
}
// The four Coffman conditions must all hold for a deadlock; removing any one
// of them prevents it.
println!("\nCoffman conditions:");
for condition in [
"mutual exclusion",
"hold and wait",
"no preemption",
"circular wait",
] {
println!(" - {condition}");
}
assert!(!has_deadlock(&acyclic));
assert!(has_deadlock(&cycle));
assert!(has_deadlock(&self_wait));
assert!(!has_deadlock(&[]));
println!("\nall checks passed");
}$ cargo run --bin concurrency_deadlock
0->1, 1->2, 0->2 deadlock: false
0->1, 1->2, 2->0 deadlock: true
0->0 deadlock: true
Coffman conditions:
- mutual exclusion
- hold and wait
- no preemption
- circular wait
all checks passed
The program ends with the four Coffman conditions. A deadlock needs all four:
- Mutual exclusion: a lock can be held by only one thread.
- Hold and wait: a thread holds one lock while waiting for another.
- No preemption: a lock cannot be taken away from its holder.
- Circular wait: the waiting forms a cycle.
Removing any one prevents deadlock. The most practical rule removes the fourth: give every lock a fixed order, and always take locks in that order. If every thread takes lock A before lock B, no thread can hold B while waiting for A, and no cycle can form.
m1 first, the waiting thread holds nothing, and no cycle can form.Summary
thread::spawnneeds closures that own their data.thread::scopelets threads borrow local data, and waits for them.- Collect the join handles before joining, or lazy iterators run the threads one at a time.
SendandSyncare auto traits.RcandRefCelllack them, so the compiler rejects sharing them across threads.- An atomic performs read, change, and write as one step.
Relaxedsuffices for a standalone counter.ReleaseandAcquirepair up to pass other data between threads. - Eight threads summed a large array about 3 to 4 times faster, limited by memory bandwidth and four physical
cores. Amdahl’s law caps speedup at
1 / sfor a serial fractions. - A spin lock is an atomic flag, a compare-exchange loop, and a guard that releases on drop. Its
unsafe impl Syncis a promise the code must keep. - A semaphore built from a
Mutexand aCondvarwaits in awhileloop, because wakeups can be spurious. - A panic while holding a
MutexGuardpoisons the mutex, andinto_innergives the data to recovery code. - Threads are deadlocked when their wait-for graph has a cycle. A global lock order prevents it.
Chapter 17 uses these tools to build queues that pass work between threads, and makes producers wait when the queue is full.
Exercises
- Write a version of
parallel_sumfrom listing 16.1 as a single lazy chain,.map(spawn).map(join).sum(), and time both on a large slice. Explain the difference. - Make the three pairs of functions in
parallel_sum.rsone generic function overT: Sum<T> + Copy + Send + Sync. Which extra bound doessum::<T>()over&Titems need? - Close the gap in
SpinGuard. Add a field_marker: PhantomData<&'a mut T>, and explain why this makes the guardSynconly whenT: Sync. - Rewrite
try_acquire_timeoutto loop until a permit is free or the whole timeout has passed. UseInstant::now()to compute the time left for each wait. - Write a program that deadlocks with two
Mutexvalues and two threads. Then fix it by taking the locks in a fixed order.