Concurrency & shared state

// Logical race condition: compiles, but wrong
use std::sync::Mutex;
let stock = Mutex::new(1);

// Two threads do this at the same time:
let available = *stock.lock().unwrap() > 0;   // lock taken then released
if available {
    *stock.lock().unwrap() -= 1;              // the other thread may have slipped in between: stock = -1
}

// Correct: read and write under the same lock
let mut s = stock.lock().unwrap();
if *s > 0 {
    *s -= 1;
}

use std::thread;

let h = thread::spawn(|| {
    (1..=10).sum::<u32>()
});

let total = h.join().unwrap();       // wait for the end; Err if the thread panicked

let v = vec![1, 2, 3];
let h = thread::spawn(move || v.iter().sum::<i32>());   // v belongs to the thread
let s = h.join().unwrap();

let handles: Vec<_> = (0..4)
    .map(|i| thread::spawn(move || i * i))
    .collect();

let results: Vec<i32> = handles.into_iter().map(|h| h.join().unwrap()).collect();

let data = vec![1, 2, 3, 4, 5, 6];
let (left, right) = data.split_at(3);

let (a, b) = thread::scope(|s| {
    let ha = s.spawn(|| left.iter().sum::<i32>());   // borrows, with no move or Arc
    let hb = s.spawn(|| right.iter().sum::<i32>());
    (ha.join().unwrap(), hb.join().unwrap())
});

A thread spawned in a scope cannot outlive the block, so it can borrow local variables. This is often the simplest way to parallelize a computation.

let h = thread::Builder::new()
    .name("cleanup".into())
    .spawn(|| { /* ... */ })
    .unwrap();

The name appears in panic messages and profiling tools.

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

let config = Arc::new(vec!["a", "b"]);
let mut handles = vec![];
for i in 0..3 {
    let c = Arc::clone(&config);         // copy of the pointer, counter + 1
    handles.push(thread::spawn(move || println!("{i} : {:?}", c)));
}
for h in handles {
    h.join().unwrap();
}
println!("{}", Arc::strong_count(&config));   // 1: the threads released their copy

use std::sync::Mutex;

let m = Mutex::new(0);
{
    let mut guard = m.lock().unwrap();   // only one thread at a time here
    *guard += 1;
}                                        // guard dropped: lock released

lock() returns a MutexGuard: it dereferences to the value, and its Drop releases the lock. You cannot forget to unlock.

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

let counter = Arc::new(Mutex::new(0));
let handles: Vec<_> = (0..10)
    .map(|_| {
        let c = Arc::clone(&counter);
        thread::spawn(move || *c.lock().unwrap() += 1)
    })
    .collect();
for h in handles {
    h.join().unwrap();
}
assert_eq!(*counter.lock().unwrap(), 10);

// Bad: the lock is held during the whole processing
let list = Mutex::new(vec![1, 2, 3]);
{
    let guard = list.lock().unwrap();
    for x in guard.iter() {
        long_task(*x);                   // the other threads wait
    }
}

// Better: copy, release, process
let copy = list.lock().unwrap().clone();   // lock released at the end of the statement
for x in copy {
    long_task(x);
}

fn long_task(_: i32) {}

match m.try_lock() {
    Ok(mut g) => g.push(2),
    Err(_) => println!("busy: we will try again later"),
}

let mut m = Mutex::new(5);
*m.get_mut().unwrap() += 1;        // &mut m proves no other access exists: no lock
let value = m.into_inner().unwrap();   // take the data out and destroy the Mutex

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

let config = Arc::new(RwLock::new(String::from("v1")));

let reader = config.read().unwrap();      // several readers at the same time
println!("{}", *reader);
drop(reader);

let mut writer = config.write().unwrap(); // a single writer, no readers
*writer = String::from("v2");

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

let counter = Arc::new(AtomicU64::new(0));
let handles: Vec<_> = (0..10)
    .map(|_| {
        let c = Arc::clone(&counter);
        thread::spawn(move || {
            c.fetch_add(1, Ordering::Relaxed);
        })
    })
    .collect();
for h in handles {
    h.join().unwrap();
}
assert_eq!(counter.load(Ordering::Relaxed), 10);

static STOP: AtomicBool = AtomicBool::new(false);
STOP.store(true, Ordering::Release);
if STOP.load(Ordering::Acquire) { /* stop */ }

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

let (tx, rx) = mpsc::channel();
for i in 0..3 {
    let tx = tx.clone();
    thread::spawn(move || tx.send(i * 10).unwrap());
}
drop(tx);                                   // close it: otherwise rx waits forever

let total: i32 = rx.iter().sum();           // receives until the channel closes

let (tx, rx) = mpsc::sync_channel::<String>(10);   // beyond 10, send blocks

// 1. Shared memory: several threads modify the same Vec under a lock
// 2. Messages: each thread sends its results; a single owner assembles them

Messages avoid locks and deadlocks: a single task owns the data. For async code, tokio provides its own channels (mpsc, oneshot, broadcast, watch, see async-tokio). The crossbeam-channel crate offers richer std-style channels (several receivers, select!).

use std::collections::HashMap;
use std::sync::LazyLock;

static CODES: LazyLock<HashMap<u16, &'static str>> = LazyLock::new(|| {
    HashMap::from([(404, "not found"), (500, "internal error")])
});

fn message(code: u16) -> &'static str {
    CODES.get(&code).copied().unwrap_or("unknown")
}

• The closure runs only once, on first access, even if several threads access it at the same time.
• Available since Rust 1.80: it replaces lazy_static! and once_cell::sync::Lazy.

// lazy_static (crate)
// lazy_static! { static ref CODES: HashMap<u16, &'static str> = { .. }; }

// once_cell (crate)
// static CODES: once_cell::sync::Lazy<HashMap<..>> = Lazy::new(|| ..);

// Today: std only
// static CODES: std::sync::LazyLock<HashMap<..>> = LazyLock::new(|| ..);

use std::sync::OnceLock;

static BASE_URL: OnceLock<String> = OnceLock::new();

fn initialize(url: String) {
    if BASE_URL.set(url).is_err() {
        panic!("BASE_URL already initialized");
    }
}

fn base_url() -> &'static str {
    BASE_URL.get().map(String::as_str).expect("initialize() must be called at startup")
}

// Or: computed on first request, from a function
fn cache() -> &'static Vec<u32> {
    static C: OnceLock<Vec<u32>> = OnceLock::new();
    C.get_or_init(|| (1..=10).collect())
}

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

let a = Arc::new(Mutex::new(0));
let b = Arc::new(Mutex::new(0));

let (a1, b1) = (Arc::clone(&a), Arc::clone(&b));
let t1 = thread::spawn(move || {
    let _ga = a1.lock().unwrap();
    let _gb = b1.lock().unwrap();      // waits for b
});

let (a2, b2) = (Arc::clone(&a), Arc::clone(&b));
let t2 = thread::spawn(move || {
    let _gb = b2.lock().unwrap();
    let _ga = a2.lock().unwrap();      // waits for a: if t1 holds a and waits for b, both wait forever
});

use rayon::prelude::*;

let v: Vec<u64> = (1..=1_000_000).collect();

let total: u64 = v.iter().sum();                      // one core
let total_par: u64 = v.par_iter().sum();              // all cores

let squares: Vec<u64> = v.par_iter().map(|x| x * x).collect();
let mut to_sort = v.clone();
to_sort.par_sort_unstable();

// runique/src/utils/config/env.rs (excerpt)
static DEBUG: LazyLock<bool> = LazyLock::new(|| debug_from(read_after_dotenv("DEBUG").as_deref()));

pub fn is_debug() -> bool {
    *DEBUG
}

// Version token for static files (?v=...), computed only once:
// a fingerprint of the contents of the project's and Runique's .css/.js files.
static CSS_TOKEN: LazyLock<String> = LazyLock::new(|| { /* ... */ });

read_after_dotenv loads .env before reading the variable: this is the answer to the pitfall in section 8.5. And since the value is computed only once, a CSS file changed in production is only picked up when the server restarts.

use std::collections::HashMap;
use std::sync::{LazyLock, Mutex};

static ATTEMPTS: LazyLock<Mutex<HashMap<String, u32>>> = LazyLock::new(|| Mutex::new(HashMap::new()));

fn login_failure(ip: &str) {
    *ATTEMPTS.lock().unwrap().entry(ip.to_string()).or_insert(0) += 1;
}
// Every IP address that gets it wrong adds an entry, never removed.

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

fn count(t: &str) -> HashMap<String, usize> {
    let mut m = HashMap::new();
    for word in t.split_whitespace() {
        *m.entry(word.to_lowercase()).or_insert(0) += 1;
    }
    m
}

fn merge(total: &mut HashMap<String, usize>, partial: HashMap<String, usize>) {
    for (k, v) in partial {
        *total.entry(k).or_insert(0) += v;
    }
}

// 1. Shared memory: each thread writes into a common table under a lock
fn v1(texts: Vec<String>) -> HashMap<String, usize> {
    let total = Arc::new(Mutex::new(HashMap::new()));
    let hs: Vec<_> = texts.into_iter().map(|t| {
        let total = Arc::clone(&total);
        thread::spawn(move || {
            let partial = count(&t);                         // computation outside the lock
            merge(&mut total.lock().unwrap(), partial);      // lock only to merge
        })
    }).collect();
    hs.into_iter().for_each(|h| h.join().unwrap());
    Arc::try_unwrap(total).unwrap().into_inner().unwrap()
}

// 2. Channel: each thread sends its result, the main thread merges
fn v2(texts: Vec<String>) -> HashMap<String, usize> {
    let (tx, rx) = mpsc::channel();
    for t in texts {
        let tx = tx.clone();
        thread::spawn(move || tx.send(count(&t)).unwrap());
    }
    drop(tx);
    let mut total = HashMap::new();
    for partial in rx {
        merge(&mut total, partial);
    }
    total
}

// 3. scope + join: the threads borrow the texts, return their result
fn v3(texts: &[String]) -> HashMap<String, usize> {
    thread::scope(|s| {
        let hs: Vec<_> = texts.iter().map(|t| s.spawn(move || count(t))).collect();
        let mut total = HashMap::new();
        for h in hs {
            merge(&mut total, h.join().unwrap());
        }
        total
    })
}

// 4. rayon: parallel map, then reduction
fn v4(texts: &[String]) -> HashMap<String, usize> {
    use rayon::prelude::*;
    texts
        .par_iter()
        .map(|t| count(t))
        .reduce(HashMap::new, |mut a, b| { merge(&mut a, b); a })
}

thread::spawn(move || ..).join()         thread::scope(|s| { s.spawn(|| ..); })
Arc::new(v)   Arc::clone(&a)             Arc::try_unwrap(a)
Mutex::new(v)   m.lock().unwrap()        m.lock().unwrap_or_else(|e| e.into_inner())
m.try_lock()   m.get_mut()   m.into_inner()
RwLock::new(v)   r.read()   r.write()
AtomicU64::new(0)   a.fetch_add(1, Ordering::Relaxed)   a.load(..)   a.fetch_update(..)
mpsc::channel()   mpsc::sync_channel(n)   drop(tx)   rx.iter()
static X: LazyLock<T> = LazyLock::new(|| ..);
static Y: OnceLock<T> = OnceLock::new();   Y.set(v)   Y.get()   Y.get_or_init(|| ..)
v.par_iter().map(f).sum()                // rayon