Concurrency

Concurrency#

Rust offers two concurrency models. OS threads (std::thread) plus Send / Sync markers and Mutex / Arc for sharing; async (async / await) on a runtime like tokio or async-std for high-fan-out I/O. The borrow checker eliminates data races at compile time: two threads can share a value only when its type is Sync.

/functions`. For the ownership rules underneath, see Variables.

Send and Sync#

  • Send: ownership of values of this type can move across threads.

  • Sync: &T is Send (the type is safe to share by reference).

Most types are automatically both. Rc, RefCell, raw pointers, and Cell are not Sync (or Send). The compiler refuses to share them across threads.

let v = Rc::new(10);                          // !Send
// std::thread::spawn(move || println!("{v}"));  // compile error

let v = Arc::new(10);                          // Send + Sync
std::thread::spawn(move || println!("{v}"));   // ok

Spawning threads#

use std::thread;

let h = thread::spawn(|| {
    println!("hello from thread");
    42
});
let v = h.join().unwrap();                     // wait for it; get value

let n = 7;
thread::spawn(move || println!("{n}"));         // move captures n

Threads are OS threads; cheap-ish (a few MB stack), not as cheap as goroutines.

For fan-out with shared state, share via Arc<Mutex<...>> or Arc<atomic_T>.

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

let counter = Arc::new(Mutex::new(0));

let mut handles = vec![];
for _ in 0..10 {
    let c = Arc::clone(&counter);
    handles.push(thread::spawn(move || {
        *c.lock().unwrap() += 1;
    }));
}
for h in handles { h.join().unwrap(); }
println!("{}", *counter.lock().unwrap());

Mutex and RwLock#

Mutex<T> wraps a value so threads access it serially. lock() returns a guard; the lock is released when the guard drops.

let mu = Mutex::new(0);
{
    let mut guard = mu.lock().unwrap();
    *guard += 1;
}                                              // unlocks here

RwLock<T> is many-readers, one-writer.

let rw = RwLock::new(vec![1, 2, 3]);
let r = rw.read().unwrap();
println!("{r:?}");

Lock poisoning: if a thread panics while holding the lock, subsequent lock() returns Err(PoisonError). The operator either .unwrap() (assume not poisoned) or handles e.into_inner() to recover the value.

Atomics#

std::sync::atomic exposes lock-free primitives.

use std::sync::atomic::{AtomicU64, Ordering};

static HITS: AtomicU64 = AtomicU64::new(0);
HITS.fetch_add(1, Ordering::Relaxed);
let v = HITS.load(Ordering::Relaxed);

Memory orderings: Relaxed (no synchronisation), Acquire / Release (one-way fence), SeqCst (default; total order). Relaxed is fine for monotonic counters; pick Acquire / Release when atomics gate access to other memory.

Channels#

std::sync::mpsc (multi-producer, single-consumer) is in the stdlib. The crossbeam ecosystem (crossbeam-channel) adds multi-consumer, select, and unbounded / bounded variants.

use std::sync::mpsc;

let (tx, rx) = mpsc::channel();
for i in 0..3 {
    let tx = tx.clone();
    thread::spawn(move || tx.send(i).unwrap());
}
drop(tx);                                       // close
while let Ok(v) = rx.recv() { println!("{v}"); }

rayon (data parallelism)#

rayon parallelises iterator chains across the thread pool. Two-line change to add parallelism.

[dependencies]
rayon = "1"

use rayon::prelude::*;

let sum: u64 = (0..1_000_000u64).into_par_iter().map(|n| n * n).sum();
let max  = items.par_iter().filter(|x| x.is_active()).count();

async / await#

async functions return a Future. Calling them returns the future; an executor (tokio, async-std, smol) drives it.

async fn fetch(url: &str) -> reqwest::Result<String> {
    reqwest::get(url).await?.text().await
}

#[tokio::main]
async fn main() -> reqwest::Result<()> {
    let body = fetch("https://example.com").await?;
    println!("{body}");
    Ok(())
}

tokio#

The default async runtime. tokio::spawn schedules a future onto the runtime; select! waits on multiple futures.

#[tokio::main]
async fn main() {
    let a = tokio::spawn(async { fetch(u1).await });
    let b = tokio::spawn(async { fetch(u2).await });

    let (ra, rb) = tokio::join!(a, b);
}

tokio::select! {
    v = primary() => handle(v),
    _ = tokio::time::sleep(Duration::from_secs(5)) => timeout(),
    _ = tokio::signal::ctrl_c() => shutdown(),
}

Async Mutex (tokio::sync::Mutex) and async channels (tokio::sync::mpsc, broadcast, oneshot) live under tokio::sync.

Concurrency cap#

For N-at-a-time async tasks, futures::stream::iter plus buffer_unordered.

use futures::stream::{self, StreamExt};

let bodies: Vec<_> = stream::iter(urls)
    .map(|u| fetch(u))
    .buffer_unordered(8)                       // 8 in flight
    .collect()
    .await;

Cancellation#

Futures cancel by dropping. Use tokio::select! or tokio_util::sync::CancellationToken to coordinate.

tokio::select! {
    _ = fetch(url) => {}
    _ = tokio::time::sleep(Duration::from_secs(3)) => println!("timeout"),
}

References#

Contents