Shared State: Arc, Mutex & RwLock

The Threads lesson gave each thread its own data, or let scoped threads borrow it read-only. Sooner or later threads need to share and change the same data: a cache, a counter, a set of accounts. Rust has the usual building blocks (mutexes, read-write locks, condition variables), with one twist that removes a whole class of bugs:

A Rust lock owns the data it protects. You can't touch the data without holding the lock, because the only way to reach it is through the lock.

The connection between a lock and the data it guards lives in the type: Mutex<HashMap<String, User>>, not in comments and conventions.

Need	Rust
Shared ownership across threads	`Arc<T>` (atomic reference counting)
Mutual exclusion	`Mutex<T>`
Many readers or one writer	`RwLock<T>`
Wait for a condition	`Condvar`
Wait for a group of threads	`Barrier` (and channels, next lesson)
Concurrent map	`Mutex<HashMap>` / `RwLock<HashMap>`, sharded maps, or crates like `dashmap`

1. `Arc`: shared ownership across threads

Arc<T> is the thread-safe version of Rc<T>: several owners, and the value is dropped when the last one goes away. The count is updated atomically, so it's safe to clone and drop from many threads.

Rust

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

fn main() {
    let catalog = Arc::new(vec!["keyboard", "mouse", "monitor"]);

    let handles: Vec<_> = (0..3)
        .map(|i| {
            let catalog = Arc::clone(&catalog); // cheap: bumps a counter, doesn't copy the Vec
            thread::spawn(move || format!("thread {i} sees {}", catalog[i]))
        })
        .collect();

    for h in handles {
        println!("{}", h.join().unwrap());
    }
    println!("owners left: {}", Arc::strong_count(&catalog));
}

Output

thread 0 sees keyboard
thread 1 sees mouse
thread 2 sees monitor
owners left: 1

Each thread gets its own Arc (moved into the closure), and all of them point to the same Vec.

2. `Arc` alone is read-only

Arc<T> gives shared access, and shared access means &T: you can read, but not modify. Trying to change the value is a compile error.

Rust✗ does not compile · E0594

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

fn main() {
    let visits = Arc::new(0);
    let v = Arc::clone(&visits);
    thread::spawn(move || {
        *v += 1;
    })
    .join()
    .unwrap();
}

Output

error[E0594]: cannot assign to data in an `Arc`
 --> src/main.rs:8:9
  |
8 |         *v += 1;
  |         ^^^^^^^ cannot assign
  |
  = help: trait `DerefMut` is required to modify through a dereference, but it is not implemented for `Arc<i32>`

To mutate shared data you need something that makes mutation safe: a Mutex, an RwLock, or an atomic.

3. `Mutex` basics

Mutex<T> wraps a value. lock() blocks until the lock is free and returns a guard, which derefs to the data. When the guard goes out of scope, the lock is released. There's no unlock() to forget.

Rust

use std::sync::Mutex;

fn main() {
    let inventory = Mutex::new(vec![String::from("apple")]);

    {
        let mut items = inventory.lock().unwrap(); // MutexGuard<Vec<String>>
        items.push(String::from("pear"));
        items.push(String::from("plum"));
    } // guard dropped: lock released

    let items = inventory.lock().unwrap();
    println!("{} items: {:?}", items.len(), *items);
}

Output

3 items: ["apple", "pear", "plum"]

lock() returns a Result because of poisoning (example 11). unwrap() is the common choice.

4. `Arc<Mutex<T>>`: the classic shared counter

The combination you'll write most often: Arc to share, Mutex to mutate. Eight threads increment a counter a thousand times each, and no increments are lost.

Rust

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

fn main() {
    let counter = Arc::new(Mutex::new(0));

    let handles: Vec<_> = (0..8)
        .map(|_| {
            let counter = Arc::clone(&counter);
            thread::spawn(move || {
                for _ in 0..1000 {
                    *counter.lock().unwrap() += 1;
                }
            })
        })
        .collect();

    for h in handles {
        h.join().unwrap();
    }
    println!("count: {}", *counter.lock().unwrap());
}

Output

count: 8000

For a plain counter, an atomic is faster (see Atomics). Mutex is the general tool for anything bigger than one number.

5. Scoped threads don't need `Arc`

With thread::scope, threads can borrow the mutex directly: &Mutex<T> is enough, because the scope guarantees the mutex outlives the threads.

Rust

use std::sync::Mutex;
use std::thread;

fn main() {
    let totals = Mutex::new(Vec::new());
    let regions = ["north", "south", "east", "west"];

    thread::scope(|s| {
        for (i, region) in regions.iter().enumerate() {
            let totals = &totals;
            s.spawn(move || {
                let sales = (i as u32 + 1) * 1000; // pretend to compute something
                totals.lock().unwrap().push((*region, sales));
            });
        }
    });

    let mut totals = totals.into_inner().unwrap(); // no threads left: take the data out
    totals.sort();
    println!("{totals:?}");
}

Output

[("east", 3000), ("north", 1000), ("south", 2000), ("west", 4000)]

The threads push in whatever order they finish, so we sort before printing. into_inner() consumes the mutex and returns the data, with no locking needed since nobody else can have it.

6. The data is only reachable through the lock

You can't forget to lock, because the value is inside the Mutex. There's no field to access directly:

Rust✗ does not compile · E0608

use std::sync::Mutex;

fn main() {
    let scores = Mutex::new(vec![10, 20, 30]);
    println!("{}", scores[0]);
}

Output

error[E0608]: cannot index into a value of type `std::sync::Mutex<Vec<{integer}>>`
 --> src/main.rs:5:26
  |
5 |     println!("{}", scores[0]);
  |                          ^^^

The fix is scores.lock().unwrap()[0].

7. Keep critical sections short

A guard holds the lock until it's dropped. Do the slow work outside: copy what you need, release the lock, then work. Use a block or drop(guard).

Rust

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

fn send_email(to: &str) {
    thread::sleep(Duration::from_millis(20)); // slow I/O
    println!("emailed {to}");
}

fn main() {
    let subscribers = Mutex::new(vec![String::from("ana@x.io"), String::from("ben@y.io")]);

    // Bad: holding the lock while doing slow I/O blocks every other thread.
    // for s in subscribers.lock().unwrap().iter() { send_email(s); }

    // Good: take a snapshot, release the lock, then do the slow part.
    let snapshot: Vec<String> = subscribers.lock().unwrap().clone();
    for s in &snapshot {
        send_email(s);
    }

    // Or release explicitly:
    let mut guard = subscribers.lock().unwrap();
    guard.push(String::from("cy@z.io"));
    let count = guard.len();
    drop(guard); // unlock now
    println!("{count} subscribers");
}

Output

emailed ana@x.io
emailed ben@y.io
3 subscribers

8. How long does a temporary guard live?

subscribers.lock().unwrap().clone() creates a guard as a temporary, which is dropped at the end of the statement. That's usually what you want. But a guard created in a match scrutinee lives until the end of the match, so the lock is held during every arm:

Rust

use std::sync::Mutex;

fn main() {
    let queue = Mutex::new(vec![1, 2, 3]);

    // The guard lives for the whole match: the lock is held while the arm runs.
    match queue.lock().unwrap().pop() {
        Some(job) => println!("processing {job} (lock still held)"),
        None => println!("empty"),
    }

    // Pop in its own statement: the guard is dropped at the `;`.
    let job = queue.lock().unwrap().pop();
    if let Some(job) = job {
        println!("processing {job} (lock released)");
    }
    println!("remaining: {:?}", queue.lock().unwrap());
}

Output

processing 3 (lock still held)
processing 2 (lock released)
remaining: [1]

Locking the same mutex again inside that match would deadlock (next example). Pulling the value out into a let first avoids the problem.

9. A `Mutex` is not reentrant

Rust's Mutex is not reentrant: locking it again from the thread that already holds it deadlocks (or panics; the behavior is unspecified). This compiles, and hangs:

Rust

use std::sync::Mutex;

fn main() {
    let balance = Mutex::new(100);
    let guard = balance.lock().unwrap();
    let again = balance.lock().unwrap(); // blocks forever: this thread already holds the lock
    println!("{} {}", *guard, *again);
}

The usual fix is structural: lock once in the public method and pass &mut T to private helpers, instead of having every helper lock for itself.

Rust

use std::sync::Mutex;

struct Account {
    balance: i64,
    history: Vec<i64>,
}

// Helpers take &mut Account: they never lock anything.
fn apply(acc: &mut Account, amount: i64) {
    acc.balance += amount;
    acc.history.push(amount);
}

fn deposit_many(account: &Mutex<Account>, amounts: &[i64]) {
    let mut acc = account.lock().unwrap(); // lock exactly once
    for &a in amounts {
        apply(&mut acc, a);
    }
}

fn main() {
    let account = Mutex::new(Account { balance: 0, history: vec![] });
    deposit_many(&account, &[100, -30, 45]);
    let acc = account.lock().unwrap();
    println!("balance {} history {:?}", acc.balance, acc.history);
}

Output

balance 115 history [100, -30, 45]

10. `try_lock`: don't wait

try_lock() returns immediately: the guard if the lock was free, or an error if another thread holds it. Use it for optional work, like skipping a cache refresh that's already in progress.

Rust

use std::sync::Mutex;
use std::thread;

fn main() {
    let cache = Mutex::new(String::from("stale"));

    let guard = cache.lock().unwrap(); // main holds the lock...
    thread::scope(|s| {
        s.spawn(|| match cache.try_lock() {
            Ok(mut c) => *c = String::from("fresh"),
            Err(_) => println!("refresh already in progress, skipping"),
        });
    });
    drop(guard);

    if let Ok(mut c) = cache.try_lock() {
        *c = String::from("fresh");
    }
    println!("cache: {}", cache.lock().unwrap());
}

Output

refresh already in progress, skipping
cache: fresh

11. Poisoning: a thread panicked while holding the lock

If a thread panics while it holds a lock, the data might be half-updated. Rust marks the mutex poisoned, and later lock() calls return Err. You can still get the data with into_inner() on the error if you know how to repair it.

Rust

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

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

    let l = Arc::clone(&ledger);
    let result = thread::spawn(move || {
        let mut entries = l.lock().unwrap();
        entries.push(300);
        panic!("crashed halfway through an update");
    })
    .join();
    println!("worker failed: {}", result.is_err());

    match ledger.lock() {
        Ok(entries) => println!("entries: {entries:?}"),
        Err(poisoned) => {
            println!("lock is poisoned; recovering");
            let entries = poisoned.into_inner();
            println!("entries: {:?}", *entries);
        }
    }
    println!("still poisoned: {}", ledger.is_poisoned());
}

Output

worker failed: true
lock is poisoned; recovering
entries: [100, 200, 300]
still poisoned: true

Most Rust code just calls unwrap() and lets the poison spread as a panic, which is usually the right call.

12. `RwLock`: many readers or one writer

RwLock<T> allows any number of simultaneous readers (read()), or one writer (write()). It suits data that's read far more than written, like configuration.

Rust

use std::collections::HashMap;
use std::sync::RwLock;
use std::thread;

fn main() {
    let config = RwLock::new(HashMap::from([("timeout_ms", 500), ("retries", 3)]));

    thread::scope(|s| {
        // Four readers can hold read guards at the same time.
        let readers: Vec<_> = (0..4).map(|_| s.spawn(|| config.read().unwrap()["timeout_ms"])).collect();
        for r in readers {
            assert_eq!(r.join().unwrap(), 500);
        }
    });

    {
        let mut cfg = config.write().unwrap(); // exclusive: waits for all readers to finish
        cfg.insert("timeout_ms", 1000);
    }

    let cfg = config.read().unwrap();
    let mut entries: Vec<_> = cfg.iter().collect();
    entries.sort();
    println!("{entries:?}");
}

Output

[("retries", 3), ("timeout_ms", 1000)]

RwLock is not always faster than Mutex: it has more overhead per operation. Measure before switching; for short critical sections, a Mutex often wins.

13. Hot-swapping config with `RwLock<Arc<T>>`

A useful pattern for read-mostly data: store an Arc<Config> inside the lock. Readers lock only long enough to clone the Arc (a counter bump), then use the config with no lock held. A writer swaps in a whole new Arc.

Rust

use std::sync::{Arc, RwLock};

#[derive(Debug)]
struct Config {
    version: u32,
    banner: String,
}

fn current(cfg: &RwLock<Arc<Config>>) -> Arc<Config> {
    Arc::clone(&cfg.read().unwrap()) // lock held only for this line
}

fn main() {
    let cfg = RwLock::new(Arc::new(Config { version: 1, banner: String::from("Welcome!") }));

    let old = current(&cfg); // a request in flight keeps using version 1...
    *cfg.write().unwrap() = Arc::new(Config { version: 2, banner: String::from("Sale today!") });
    let new = current(&cfg);

    println!("in-flight request: v{} {:?}", old.version, old.banner);
    println!("new requests:      v{} {:?}", new.version, new.banner);
}

Output

in-flight request: v1 "Welcome!"
new requests:      v2 "Sale today!"

14. Getting data out: `Arc::into_inner` and `Mutex::into_inner`

When the threads are done, you often want the final value back without a lock. Arc::into_inner returns the value if this was the last Arc, and Mutex::into_inner unwraps the mutex.

Rust

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

fn main() {
    let log = Arc::new(Mutex::new(Vec::new()));

    let handles: Vec<_> = ["db", "cache", "queue"]
        .into_iter()
        .map(|name| {
            let log = Arc::clone(&log);
            thread::spawn(move || log.lock().unwrap().push(format!("{name} started")))
        })
        .collect();
    for h in handles {
        h.join().unwrap();
    }

    let mut lines: Vec<String> = Arc::into_inner(log).expect("other owners still alive").into_inner().unwrap();
    lines.sort();
    println!("{lines:?}");
}

Output

["cache started", "db started", "queue started"]

15. `get_mut`: no locking when you have exclusive access

If you hold &mut Mutex<T>, no other thread can have access, so get_mut() returns &mut T without locking at all. The borrow checker proves it's safe.

Rust

use std::sync::Mutex;

struct Stats {
    hits: Mutex<u64>,
}

impl Stats {
    fn record(&self) {
        *self.hits.lock().unwrap() += 1; // shared access: must lock
    }

    fn reset(&mut self) -> u64 {
        let hits = self.hits.get_mut().unwrap(); // exclusive access: no lock needed
        std::mem::take(hits)
    }
}

fn main() {
    let mut stats = Stats { hits: Mutex::new(0) };
    stats.record();
    stats.record();
    println!("reset returned {}", stats.reset());
    println!("now {}", stats.hits.lock().unwrap());
}

Output

reset returned 2
now 0

16. A shared cache: `Mutex<HashMap>`

A memoizing cache shared by worker threads. Note how the expensive computation happens outside the lock, so threads don't serialize on it.

Rust

use std::collections::HashMap;
use std::sync::Mutex;
use std::thread;

fn slow_square(n: u64) -> u64 {
    thread::sleep(std::time::Duration::from_millis(10));
    n * n
}

fn cached(cache: &Mutex<HashMap<u64, u64>>, n: u64) -> u64 {
    if let Some(&v) = cache.lock().unwrap().get(&n) {
        return v; // guard dropped at the end of the `if let`
    }
    let v = slow_square(n); // compute without holding the lock
    cache.lock().unwrap().insert(n, v);
    v
}

fn main() {
    let cache = Mutex::new(HashMap::new());
    let inputs = [3, 4, 3, 5, 4, 3];

    let results: Vec<u64> = thread::scope(|s| {
        let cache = &cache; // move this reference (and `n`) into each thread
        let handles: Vec<_> = inputs.iter().map(|&n| s.spawn(move || cached(cache, n))).collect();
        handles.into_iter().map(|h| h.join().unwrap()).collect()
    });

    println!("results: {results:?}");
    println!("distinct keys cached: {}", cache.lock().unwrap().len());
}

Output

results: [9, 16, 9, 25, 16, 9]
distinct keys cached: 3

Two threads might both miss and compute the same value. That's harmless here; when it isn't, hold the lock during computation, or use a per-key OnceLock (Atomics).

17. Locking several things: avoid deadlock with a fixed order

Deadlock happens when thread A holds lock 1 and waits for lock 2, while thread B holds 2 and waits for 1. The standard fix is to always acquire locks in one global order, for example by account id.

Rust

use std::sync::Mutex;
use std::thread;

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

fn transfer(from: &Account, to: &Account, amount: i64) -> Result<(), String> {
    if from.id == to.id {
        return Err(String::from("same account"));
    }
    // Lock the lower id first, whatever the direction of the transfer.
    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();
    let (from_bal, to_bal) = if from.id < to.id { (&mut *a, &mut *b) } else { (&mut *b, &mut *a) };

    if *from_bal < amount {
        return Err(format!("account {} has only {}", from.id, from_bal));
    }
    *from_bal -= amount;
    *to_bal += amount;
    Ok(())
}

fn main() {
    let alice = Account { id: 1, balance: Mutex::new(1000) };
    let bob = Account { id: 2, balance: Mutex::new(1000) };

    thread::scope(|s| {
        // Opposite directions at the same time: would deadlock without the ordering.
        s.spawn(|| for _ in 0..500 { transfer(&alice, &bob, 1).unwrap(); });
        s.spawn(|| for _ in 0..300 { transfer(&bob, &alice, 1).unwrap(); });
    });

    let (a, b) = (*alice.balance.lock().unwrap(), *bob.balance.lock().unwrap());
    println!("alice {a}, bob {b}, total {}", a + b);
    println!("{:?}", transfer(&alice, &bob, 5000));
}

Output

alice 800, bob 1200, total 2000
Err("account 1 has only 800")

Often the simplest answer is fewer locks. If accounts are always updated together, put them all behind one Mutex. There's then no ordering to get wrong, at the cost of less parallelism.

Rust

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

struct Bank {
    accounts: Mutex<HashMap<&'static str, i64>>,
}

impl Bank {
    fn transfer(&self, from: &str, to: &str, amount: i64) -> Result<(), String> {
        let mut accts = self.accounts.lock().unwrap();
        let from_bal = *accts.get(from).ok_or(format!("no account {from}"))?;
        if !accts.contains_key(to) {
            return Err(format!("no account {to}"));
        }
        if from_bal < amount {
            return Err(format!("{from} has only {from_bal}"));
        }
        *accts.get_mut(from).unwrap() -= amount;
        *accts.get_mut(to).unwrap() += amount;
        Ok(())
    }
}

fn main() {
    let bank = Arc::new(Bank { accounts: Mutex::new(HashMap::from([("ana", 500), ("ben", 500), ("cy", 0)])) });

    let handles: Vec<_> = [("ana", "cy"), ("ben", "cy"), ("cy", "ana")]
        .into_iter()
        .map(|(from, to)| {
            let bank = Arc::clone(&bank);
            thread::spawn(move || (0..100).filter(|_| bank.transfer(from, to, 2).is_ok()).count())
        })
        .collect();
    let succeeded: usize = handles.into_iter().map(|h| h.join().unwrap()).sum();

    let total: i64 = bank.accounts.lock().unwrap().values().sum(); // guard dropped at `;`
    println!("successful transfers: at least 200: {}", succeeded >= 200);
    println!("money is conserved: {}", total == 1000);
    println!("{:?}", bank.transfer("ana", "zed", 1).err());
}

Output

successful transfers: at least 200: true
money is conserved: true
Some("no account zed")

19. `Condvar`: waiting for a condition

A Condvar lets a thread sleep until another thread signals that something changed. It's always paired with a Mutex. wait_while re-checks the condition for you, which handles spurious wakeups.

Rust

use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::Duration;

fn main() {
    let state = Arc::new((Mutex::new(false), Condvar::new()));

    let s = Arc::clone(&state);
    let loader = thread::spawn(move || {
        thread::sleep(Duration::from_millis(50)); // load something
        let (ready, cvar) = &*s;
        *ready.lock().unwrap() = true;
        cvar.notify_all();
        "config loaded"
    });

    let (ready, cvar) = &*state;
    let guard = cvar.wait_while(ready.lock().unwrap(), |ready| !*ready).unwrap();
    println!("ready = {}", *guard);
    drop(guard);
    println!("{}", loader.join().unwrap());
}

Output

ready = true
config loaded

20. `Condvar` with a timeout

wait_timeout_while gives up after a deadline and tells you whether it timed out. Useful for "wait for the service to become healthy, but not forever".

Rust

use std::sync::{Condvar, Mutex};
use std::thread;
use std::time::Duration;

fn wait_for_workers(done: &Mutex<u32>, cvar: &Condvar, expected: u32, timeout: Duration) -> Result<(), u32> {
    let (count, result) = cvar.wait_timeout_while(done.lock().unwrap(), timeout, |n| *n < expected).unwrap();
    if result.timed_out() { Err(*count) } else { Ok(()) }
}

fn main() {
    let done = Mutex::new(0);
    let cvar = Condvar::new();

    thread::scope(|s| {
        for _ in 0..3 {
            s.spawn(|| {
                *done.lock().unwrap() += 1;
                cvar.notify_all();
            });
        }
        println!("all 3 finished: {:?}", wait_for_workers(&done, &cvar, 3, Duration::from_secs(5)));
    });

    // Waiting for a 4th worker that never comes times out.
    println!("waiting for 4: {:?}", wait_for_workers(&done, &cvar, 4, Duration::from_millis(50)));
}

Output

all 3 finished: Ok(())
waiting for 4: Err(3)

21. A blocking queue built from `Mutex` + `Condvar`

A bounded blocking queue is a lock with two conditions ("not empty" and "not full"). Here's a small version. The Channels lesson shows the ready-made alternative, but this is what's underneath.

Rust

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

struct BoundedQueue<T> {
    items: Mutex<VecDeque<T>>,
    not_empty: Condvar,
    not_full: Condvar,
    capacity: usize,
}

impl<T> BoundedQueue<T> {
    fn new(capacity: usize) -> Self {
        BoundedQueue { items: Mutex::new(VecDeque::new()), not_empty: Condvar::new(), not_full: Condvar::new(), capacity }
    }

    fn put(&self, item: T) {
        let mut items = self.not_full.wait_while(self.items.lock().unwrap(), |q| q.len() >= self.capacity).unwrap();
        items.push_back(item);
        self.not_empty.notify_one();
    }

    fn take(&self) -> T {
        let mut items = self.not_empty.wait_while(self.items.lock().unwrap(), |q| q.is_empty()).unwrap();
        let item = items.pop_front().unwrap();
        self.not_full.notify_one();
        item
    }
}

fn main() {
    let queue = BoundedQueue::new(2); // small capacity: the producer has to wait for the consumer

    let consumed = thread::scope(|s| {
        s.spawn(|| {
            for order in 1..=6 {
                queue.put(order);
            }
            queue.put(0); // 0 = "no more orders"
        });
        let consumer = s.spawn(|| {
            let mut got = Vec::new();
            loop {
                match queue.take() {
                    0 => break got,
                    order => got.push(order),
                }
            }
        });
        consumer.join().unwrap()
    });

    println!("consumed in order: {consumed:?}");
}

Output

consumed in order: [1, 2, 3, 4, 5, 6]

22. `Barrier`: wait until everyone arrives

A Barrier blocks threads until a given number of them have called wait(), then releases them all. It can be reused for several phases.

Rust

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

fn main() {
    let workers = 3;
    let barrier = Barrier::new(workers);
    let log = Mutex::new(Vec::new());

    thread::scope(|s| {
        for id in 0..workers {
            let (barrier, log) = (&barrier, &log);
            s.spawn(move || {
                log.lock().unwrap().push(format!("phase 1: worker {id} loaded data"));
                barrier.wait(); // nobody starts phase 2 until all finished phase 1
                log.lock().unwrap().push(format!("phase 2: worker {id} processing"));
            });
        }
    });

    let log = log.into_inner().unwrap();
    let last_phase_1 = log.iter().rposition(|l| l.starts_with("phase 1")).unwrap();
    let first_phase_2 = log.iter().position(|l| l.starts_with("phase 2")).unwrap();
    println!("{} log lines", log.len());
    println!("every phase 1 line comes before every phase 2 line: {}", last_phase_1 < first_phase_2);
}

Output

6 log lines
every phase 1 line comes before every phase 2 line: true

Instead of Arc<Mutex<Service>> (one big lock), put locks on the fields that change. Methods then take &self, and unrelated operations don't block each other.

Rust

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

struct UserService {
    users: RwLock<HashMap<u32, String>>, // read-mostly
    audit: Mutex<Vec<String>>,           // append-only log
}

impl UserService {
    fn name(&self, id: u32) -> Option<String> {
        self.audit.lock().unwrap().push(format!("read {id}"));
        self.users.read().unwrap().get(&id).cloned()
    }

    fn rename(&self, id: u32, name: &str) {
        self.users.write().unwrap().insert(id, name.to_string());
        self.audit.lock().unwrap().push(format!("rename {id}"));
    }
}

fn main() {
    let svc = Arc::new(UserService {
        users: RwLock::new(HashMap::from([(1, String::from("ana")), (2, String::from("ben"))])),
        audit: Mutex::new(Vec::new()),
    });

    let readers: Vec<_> = (1..=2)
        .map(|id| {
            let svc = Arc::clone(&svc);
            thread::spawn(move || svc.name(id))
        })
        .collect();
    let names: Vec<_> = readers.into_iter().map(|h| h.join().unwrap()).collect();
    svc.rename(2, "benjamin");

    println!("{names:?} -> {:?}", svc.name(2));
    println!("audit entries: {}", svc.audit.lock().unwrap().len());
}

Output

[Some("ana"), Some("ben")] -> Some("benjamin")
audit entries: 4

24. A sharded map: less contention

One Mutex<HashMap> serializes every access. Splitting the map into shards, each with its own lock, lets threads working on different keys proceed in parallel. This is the idea behind the Rust crate dashmap.

Rust

use std::collections::HashMap;
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::sync::Mutex;
use std::thread;

struct ShardedMap<V> {
    shards: Vec<Mutex<HashMap<String, V>>>,
}

impl<V: Clone> ShardedMap<V> {
    fn new(shards: usize) -> Self {
        ShardedMap { shards: (0..shards).map(|_| Mutex::new(HashMap::new())).collect() }
    }

    fn shard(&self, key: &str) -> &Mutex<HashMap<String, V>> {
        let mut h = DefaultHasher::new();
        key.hash(&mut h);
        &self.shards[h.finish() as usize % self.shards.len()]
    }

    fn insert(&self, key: &str, value: V) {
        self.shard(key).lock().unwrap().insert(key.to_string(), value);
    }

    fn get(&self, key: &str) -> Option<V> {
        self.shard(key).lock().unwrap().get(key).cloned()
    }

    fn len(&self) -> usize {
        self.shards.iter().map(|s| s.lock().unwrap().len()).sum()
    }
}

fn main() {
    let map = ShardedMap::new(8);
    thread::scope(|s| {
        for t in 0..4 {
            let map = &map;
            s.spawn(move || {
                for i in 0..250 {
                    map.insert(&format!("user-{t}-{i}"), i);
                }
            });
        }
    });
    println!("entries: {}", map.len());
    println!("user-2-99 = {:?}", map.get("user-2-99"));
    println!("user-9-0 = {:?}", map.get("user-9-0"));
}

Output

entries: 1000
user-2-99 = Some(99)
user-9-0 = None

25. Realistic: a token-bucket rate limiter

A limiter shared by all request handlers: each client gets a bucket of tokens that refills over time. The whole state sits behind one Mutex; each check is a few arithmetic operations, so the lock is held very briefly. Time is passed in to keep the example deterministic.

Rust

use std::collections::HashMap;
use std::sync::Mutex;
use std::thread;

struct RateLimiter {
    capacity: f64,
    refill_per_sec: f64,
    buckets: Mutex<HashMap<String, (f64, f64)>>, // client -> (tokens, last seen time)
}

impl RateLimiter {
    fn allow(&self, client: &str, now: f64) -> bool {
        let mut buckets = self.buckets.lock().unwrap();
        let (tokens, last) = buckets.entry(client.to_string()).or_insert((self.capacity, now));
        *tokens = (*tokens + (now - *last) * self.refill_per_sec).min(self.capacity);
        *last = now;
        if *tokens >= 1.0 {
            *tokens -= 1.0;
            true
        } else {
            false
        }
    }
}

fn main() {
    let limiter = RateLimiter { capacity: 5.0, refill_per_sec: 1.0, buckets: Mutex::new(HashMap::new()) };

    // 4 threads fire 3 requests each for the same client at t=0: only 5 may pass.
    let allowed: usize = thread::scope(|s| {
        let handles: Vec<_> = (0..4)
            .map(|_| s.spawn(|| (0..3).filter(|_| limiter.allow("client-a", 0.0)).count()))
            .collect();
        handles.into_iter().map(|h| h.join().unwrap()).sum()
    });
    println!("allowed at t=0: {allowed} of 12");

    println!("t=0.5: {}", limiter.allow("client-a", 0.5)); // half a token refilled: not enough
    println!("t=2.0: {}", limiter.allow("client-a", 2.0)); // 2 tokens refilled
    println!("other client: {}", limiter.allow("client-b", 2.0)); // separate bucket
}

Output

allowed at t=0: 5 of 12
t=0.5: false
t=2.0: true
other client: true

26. Realistic: a session store with expiry

A web session store: lookups are frequent (read lock), logins and cleanup are rare (write lock). A cleanup thread removes expired sessions while request threads keep reading.

Rust

use std::collections::HashMap;
use std::sync::RwLock;
use std::thread;

#[derive(Clone, Debug)]
struct Session {
    user: String,
    expires_at: u64,
}

#[derive(Default)]
struct SessionStore {
    sessions: RwLock<HashMap<String, Session>>,
}

impl SessionStore {
    fn login(&self, token: &str, user: &str, now: u64, ttl: u64) {
        let s = Session { user: user.to_string(), expires_at: now + ttl };
        self.sessions.write().unwrap().insert(token.to_string(), s);
    }

    fn user_for(&self, token: &str, now: u64) -> Option<String> {
        let sessions = self.sessions.read().unwrap();
        sessions.get(token).filter(|s| s.expires_at > now).map(|s| s.user.clone())
    }

    fn purge_expired(&self, now: u64) -> usize {
        let mut sessions = self.sessions.write().unwrap();
        let before = sessions.len();
        sessions.retain(|_, s| s.expires_at > now);
        before - sessions.len()
    }
}

fn main() {
    let store = SessionStore::default();
    store.login("t1", "ana", 0, 100);
    store.login("t2", "ben", 0, 10);
    store.login("t3", "cy", 0, 50);

    let (lookups, purged) = thread::scope(|s| {
        let reader = s.spawn(|| ["t1", "t2", "t3", "t9"].map(|t| store.user_for(t, 20)));
        let cleaner = s.spawn(|| store.purge_expired(20));
        (reader.join().unwrap(), cleaner.join().unwrap())
    });

    println!("lookups at t=20: {lookups:?}");
    println!("purged: {purged}");
    println!("remaining: {}", store.sessions.read().unwrap().len());
}

Output

lookups at t=20: [Some("ana"), None, Some("cy"), None]
purged: 1
remaining: 2

Whether the reader runs before or after the purge, t2 is already expired at t=20, so the output is the same either way. Designing operations so that interleavings don't matter is a big part of writing correct concurrent code.

27. Realistic: a worker pool pulling from a shared queue

A fixed set of worker threads takes jobs from a shared Mutex<VecDeque> until it's empty, and records results in a shared Mutex<Vec>. This is the core of what an ExecutorService does.

Rust

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

#[derive(Debug)]
struct Job {
    id: u32,
    payload: String,
}

fn main() {
    let jobs: VecDeque<Job> = (1..=10).map(|id| Job { id, payload: format!("image-{id}.png") }).collect();
    let queue = Arc::new(Mutex::new(jobs));
    let results = Arc::new(Mutex::new(Vec::new()));

    let workers: Vec<_> = (0..3)
        .map(|w| {
            let queue = Arc::clone(&queue);
            let results = Arc::clone(&results);
            thread::spawn(move || {
                let mut done = 0;
                loop {
                    let job = queue.lock().unwrap().pop_front(); // lock released at `;`
                    let Some(job) = job else { break };
                    let thumbnail = format!("thumb-{}", job.payload);
                    results.lock().unwrap().push((job.id, thumbnail));
                    done += 1;
                }
                (w, done)
            })
        })
        .collect();

    let total_done: u32 = workers.into_iter().map(|h| h.join().unwrap().1).sum();
    let mut results = results.lock().unwrap();
    results.sort();

    println!("jobs processed: {total_done}");
    println!("first: {:?}", results.first());
    println!("last: {:?}", results.last());
}

Output

jobs processed: 10
first: Some((1, "thumb-image-1.png"))
last: Some((10, "thumb-image-10.png"))

Note let job = queue.lock().unwrap().pop_front(); on its own line: if the pop_front were inside the let ... else, the lock would still be released at the end of the statement, but writing it this way makes it obvious that the job is processed without holding the queue lock.

28. Realistic: a thread-safe metrics registry

A registry of named counters and latency samples, shared by request handlers. Each metric has its own lock, and the registry map is behind an RwLock so that adding a new metric is rare and looking one up is cheap.

Rust

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

#[derive(Default)]
struct Metric {
    count: u64,
    total_ms: u64,
    max_ms: u64,
}

#[derive(Default)]
struct Registry {
    metrics: RwLock<HashMap<String, Arc<Mutex<Metric>>>>,
}

impl Registry {
    fn metric(&self, name: &str) -> Arc<Mutex<Metric>> {
        if let Some(m) = self.metrics.read().unwrap().get(name) {
            return Arc::clone(m);
        }
        let mut map = self.metrics.write().unwrap();
        Arc::clone(map.entry(name.to_string()).or_default()) // another thread may have added it meanwhile
    }

    fn record(&self, name: &str, ms: u64) {
        let metric = self.metric(name);
        let mut m = metric.lock().unwrap();
        m.count += 1;
        m.total_ms += ms;
        m.max_ms = m.max_ms.max(ms);
    }

    fn report(&self) -> Vec<String> {
        let map = self.metrics.read().unwrap();
        let mut lines: Vec<String> = map
            .iter()
            .map(|(name, m)| {
                let m = m.lock().unwrap();
                format!("{name}: n={} avg={}ms max={}ms", m.count, m.total_ms / m.count, m.max_ms)
            })
            .collect();
        lines.sort();
        lines
    }
}

fn main() {
    let registry = Registry::default();
    thread::scope(|s| {
        for t in 0..4u64 {
            let registry = &registry;
            s.spawn(move || {
                for i in 0..25u64 {
                    registry.record("GET /products", 10 + (t * 25 + i) % 7);
                    if i % 5 == 0 {
                        registry.record("POST /checkout", 100 + t * 10);
                    }
                }
            });
        }
    });
    for line in registry.report() {
        println!("{line}");
    }
}

Output

GET /products: n=100 avg=12ms max=16ms
POST /checkout: n=20 avg=115ms max=130ms

Summary

Arc<T> shares ownership across threads; alone it only gives read access.
Mutex<T> and RwLock<T> own the data they protect, so you can't access it without locking. The guard releases the lock when dropped.
Keep critical sections short: copy out, drop the guard, then do slow work. Watch temporaries in match scrutinees.
Rust mutexes aren't reentrant; structure code so each operation locks once.
A panic while holding a lock poisons it; lock() then returns Err.
Avoid deadlocks with a global lock order, or by using fewer, coarser locks.
Condvar waits for a condition (use wait_while); Barrier synchronizes phases.
Scoped threads can share &Mutex<T> without Arc; into_inner and get_mut get the data back without locking.

Next: Channels, where threads communicate by sending messages instead of sharing memory.

1. Arc: shared ownership across threads

2. Arc alone is read-only

3. Mutex basics

4. Arc<Mutex<T>>: the classic shared counter

5. Scoped threads don't need Arc

6. The data is only reachable through the lock

7. Keep critical sections short

8. How long does a temporary guard live?

9. A Mutex is not reentrant

10. try_lock: don't wait

11. Poisoning: a thread panicked while holding the lock

12. RwLock: many readers or one writer

13. Hot-swapping config with RwLock<Arc<T>>

14. Getting data out: Arc::into_inner and Mutex::into_inner

15. get_mut: no locking when you have exclusive access

16. A shared cache: Mutex<HashMap>

17. Locking several things: avoid deadlock with a fixed order

18. Or use one lock for related data

19. Condvar: waiting for a condition

20. Condvar with a timeout

21. A blocking queue built from Mutex + Condvar

22. Barrier: wait until everyone arrives

23. Sharing a struct with interior locks

24. A sharded map: less contention

25. Realistic: a token-bucket rate limiter

26. Realistic: a session store with expiry

27. Realistic: a worker pool pulling from a shared queue

28. Realistic: a thread-safe metrics registry

Summary

1. `Arc`: shared ownership across threads

2. `Arc` alone is read-only

3. `Mutex` basics

4. `Arc<Mutex<T>>`: the classic shared counter

5. Scoped threads don't need `Arc`

9. A `Mutex` is not reentrant

10. `try_lock`: don't wait

12. `RwLock`: many readers or one writer

13. Hot-swapping config with `RwLock<Arc<T>>`

14. Getting data out: `Arc::into_inner` and `Mutex::into_inner`

15. `get_mut`: no locking when you have exclusive access

16. A shared cache: `Mutex<HashMap>`

19. `Condvar`: waiting for a condition

20. `Condvar` with a timeout

21. A blocking queue built from `Mutex` + `Condvar`

22. `Barrier`: wait until everyone arrives