Concurrency | Rust - Wyatt's Notes

import Citations from ‘@components/Citations.astro’

OS Threads

Rust’s std::thread module provides a 1:1 mapping to OS threads. Each thread gets its own stack (default 8 MB on Linux, configurable) and is scheduled by the operating system.

Spawning Threads

use std::thread;
use std::time::Duration;

let handle = thread::spawn(|| {
    for i in 1..=5 {
        println!("spawned thread: {}", i);
        thread::sleep(Duration::from_millis(1));
    }
});

for i in 1..=3 {
    println!("main thread: {}", i);
    thread::sleep(Duration::from_millis(1));
}

handle.join().unwrap();

Moving Data into Threads

The closure passed to thread::spawn must own all captured values or borrow them for 'static. The move keyword transfers ownership into the thread’s closure:

use std::thread;

let s = String::from("hello");
let handle = thread::spawn(move || {
    println!("{}", s);  // s is moved into this closure
});

handle.join().unwrap();
// s is no longer valid here — it was moved

Without moveThe closure would attempt to borrow sBut the borrow checker cannot guarantee That the spawned thread will not outlive s (the thread might run after s is dropped).

Thread Return Values

JoinHandle<T> allows the spawning thread to receive the return value:

use std::thread;

let handle = thread::spawn(|| {
    let mut sum = 0;
    for i in 1..=100 {
        sum += i;
    }
    sum
});

let result = handle.join().unwrap();
assert_eq!(result, 5050);

join() blocks the calling thread until the spawned thread completes. If the spawned thread panics, join() returns Err containing the panic payload.

Scoped Threads

std::thread::scope (stable since Rust 1.63) allows spawning threads that can borrow data from the Parent scope without move or 'static:

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

let data = vec![1, 2, 3, 4, 5];

let mut results = Mutex::new(Vec::new());

thread::scope(|s| {
    for chunk in data.chunks(2) {
        let chunk = chunk.to_vec();
        s.spawn(|| {
            let sum: i32 = chunk.iter().sum();
            results.lock().unwrap().push(sum);
        });
    }
});  // all spawned threads are joined here

assert_eq!(results, vec![3, 7, 5]);

The key guarantee: all threads spawned within scope are joined before scope returns. This means Borrowed data is guaranteed to be valid for the lifetime of the scoped threads, eliminating the need For 'static bounds.

From the parent scope. They are safer (no 'static requirement) and more ergonomic.

Message Passing

Rust’s channel implementation is based on the actor model — threads communicate by sending messages, Not by sharing memory.

`mpsc` Channels

mpsc stands for “multiple producer, single consumer.” The standard library provides a bounded and Unbounded channel:

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

let (tx, rx) = mpsc::channel();

thread::spawn(move || {
    let val = String::from("hello");
    tx.send(val).unwrap();
    // val is moved into the channel — no longer accessible here
});

let received = rx.recv().unwrap();
assert_eq!(received, "hello");

Multiple Producers

Clone the Sender to create multiple producers:

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

let (tx, rx) = mpsc::channel();
let tx1 = tx.clone();

thread::spawn(move || {
    tx.send("from thread 1").unwrap();
});

thread::spawn(move || {
    tx1.send("from thread 2").unwrap();
});

drop(tx);  // drop the original sender

for received in rx {
    println!("got: {}", received);
}

When all senders are dropped, recv() returns Err (signaling the channel is closed). The for Loop over rx terminates when the channel is closed.

Bounded Channels

use std::sync::mpsc;

let (tx, rx) = mpsc::sync_channel(10);  // buffer size 10

tx.send(1).unwrap();  // OK — buffer not full
// If the buffer is full, send() blocks until a receiver reads

`send` and `recv` Semantics

Method	Behavior
`tx.send(val)`	Blocks if bounded channel is full. Moves `val` into the channel.
`rx.recv()`	Blocks until a message is available or all senders are dropped.
`rx.try_recv()`	Non-blocking. Returns `Ok(val)` or `Err(TryRecvError)`.
`rx.recv_timeout(dur)`	Blocks with timeout. Returns `Err(RecvTimeoutError)` on timeout.

Shared State

`Mutex<T>`

A mutual exclusion lock provides interior mutability — only one thread can access the data at a Time:

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

let counter = Arc::new(Mutex::new(0));
let mut handles = vec![];

for _ in 0..10 {
    let counter = Arc::clone(&counter);
    handles.push(thread::spawn(move || {
        let mut num = counter.lock().unwrap();
        *num += 1;
    }));
}

for handle in handles {
    handle.join().unwrap();
}

assert_eq!(*counter.lock().unwrap(), 10);

Mutex Poisoning

If a thread panics while holding a Mutex lock, the mutex becomes poisoned. Subsequent calls to lock() return Err(PoisonError). This is a deliberate safety feature — it prevents you from Accessing potentially inconsistent state.

let lock = counter.lock();
match lock {
    Ok(guard) => *guard += 1,
    Err(poisoned) => {
        // Recover the data if you know it is safe
        let guard = poisoned.into_inner();
        *guard += 1;
    }
}

Use lock().unwrap() when you are confident panics inside the critical section are impossible, or When a panic means the entire process should terminate.

`RwLock<T>`

A read-write lock allows multiple concurrent readers or a single exclusive writer:

use std::sync::RwLock;

let lock = RwLock::new(5);

// Multiple concurrent readers
{
    let r1 = lock.read().unwrap();
    let r2 = lock.read().unwrap();  // OK — multiple readers
    assert_eq!(*r1, 5);
    assert_eq!(*r2, 5);
}  // readers dropped

// Single exclusive writer
{
    let mut w = lock.write().unwrap();
    *w += 1;
}  // writer dropped

When to Use `RwLock` vs `Mutex`

Condition	Use
Mostly writes, low contention	`Mutex` — simpler, lower overhead
Mostly reads, occasional writes	`RwLock` — allows concurrent reads
Very high contention	Reconsider your design — locks are the bottleneck

`Arc<T>` — Atomic Reference Counting

Arc<T> enables shared ownership across threads. It is Send + Sync because the reference count is Maintained atomically:

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

let data = Arc::new(vec![1, 2, 3, 4, 5]);
let mut handles = vec![];

for _ in 0..3 {
    let data = Arc::clone(&data);
    handles.push(thread::spawn(move || {
        let sum: i32 = data.iter().sum();
        println!("sum: {}", sum);
    }));
}

for handle in handles {
    handle.join().unwrap();
}

Arc vs Rc:

Arc uses atomic operations for reference counting — thread-safe but slower.
Rc uses non-atomic reference counting — not thread-safe but faster.
Rc does not implement Send or Sync — the compiler prevents cross-thread use.

Atomic Types

std::sync::atomic provides lock-free atomic operations for primitive types. Atomics are the Foundation of lock-free data structures and are essential for performance-critical concurrent code.

Atomic Operations

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

static COUNTER: AtomicUsize = AtomicUsize::new(0);

COUNTER.fetch_add(1, Ordering::SeqCst);
let current = COUNTER.load(Ordering::SeqCst);
COUNTER.store(0, Ordering::SeqCst);
let prev = COUNTER.swap(42, Ordering::SeqCst);

// Compare-and-swap loop
let mut current = COUNTER.load(Ordering::SeqCst);
loop {
    let new = current + 1;
    match COUNTER.compare_exchange_weak(current, new, Ordering::SeqCst, Ordering::SeqCst) {
        Ok(_) => break,
        Err(actual) => current = actual,  // retry with actual value
    }
}

Memory Orderings

Ordering	Guarantee
`Relaxed`	No ordering — only atomicity guaranteed
`Release`	All prior writes are visible to threads that acquire this location
`Acquire`	All subsequent reads see writes from the last release on this location
`AcqRel`	Both acquire and release semantics
`SeqCst`	Sequentially consistent — total ordering across all threads

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

static READY: AtomicBool = AtomicBool::new(false);
static DATA: AtomicUsize = AtomicUsize::new(0);

// Writer thread
fn writer() {
    DATA.store(42, Ordering::Relaxed);          // write data
    READY.store(true, Ordering::Release);       // release fence
}

// Reader thread
fn reader() {
    if READY.load(Ordering::Acquire) {          // acquire fence
        assert_eq!(DATA.load(Ordering::Relaxed), 42);  // data is visible
    }
}

Where one thread’s write must be visible to another thread’s read. Use Release/Acquire pairs for Correct visibility semantics. Use SeqCst when you are unsure — it is the safest but slowest Option.

Available Atomic Types

Type	Operations
`AtomicBool`	load``store``swap``compare_exchange``fetch_and``fetch_or``fetch_xor
`AtomicU8`–AtomicU64``AtomicUsize	All integer atomics: fetch_add``fetch_sub``fetch_max``fetch_min``compare_exchange
`AtomicI8`–AtomicI64``AtomicIsize	Same as unsigned, with signed comparison
`AtomicPtr<T>`	Pointer-sized atomic operations

Deadlock Prevention

Common Deadlock Patterns

Lock ordering violation: Thread A holds lock 1 and waits for lock 2. Thread B holds lock 2 and waits for lock 1.
Non-reentrant locking: A thread attempts to acquire a lock it already holds. Rust’s Mutex is not reentrant (by design — reentrant locks hide bugs).
Resource starvation: A thread holds a lock for too long, preventing other threads from making progress.

Prevention Strategies

use std::sync::Mutex;

struct Account {
    balance: Mutex<i64>,
}

fn transfer(from: &Account, to: &Account, amount: i64) {
    // Always lock in a consistent order to prevent deadlock
    let (first, second) = if from as *const _ as usize < to as *const _ as usize {
        (&from.balance, &to.balance)
    } else {
        (&to.balance, &from.balance)
    };

    let mut first_lock = first.lock().unwrap();
    let mut second_lock = second.lock().unwrap();

    *first_lock.get_mut().unwrap() -= amount;
    *second_lock.get_mut().unwrap() += amount;
}

Lock Guard Scope

Keep lock guards in as small a scope as possible:

// Bad — lock held for the entire function body
fn process(data: &Mutex<Vec<i32>>) {
    let guard = data.lock().unwrap();
    guard.push(1);
    guard.push(2);
    // ... many more operations while lock is held
}

// Good — lock held only when needed
fn process(data: &Mutex<Vec<i32>>) {
    {
        let mut guard = data.lock().unwrap();
        guard.push(1);
        guard.push(2);
    }  // lock released here
    // ... other work without lock
}

Async/Await

The Problem with Threads

OS threads are expensive: each thread uses 8 MB of stack (default on Linux), context switches cost 1,000–10,000 nanoseconds, and creating threads has significant overhead. For I/O-bound workloads With thousands of concurrent tasks (web servers, database connections), threads do not scale Efficiently.

Async/await provides lightweight concurrency — thousands of tasks on a handful of OS threads.

`Future` Trait

pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

pub enum Poll<T> {
    Ready(T),
    Pending,
}

A Future represents an asynchronous computation. poll is called by the executor to check whether The computation has completed (Ready) or needs more time (Pending). When Pending is returned, The future registers the current Waker with the reactor, which will wake the future when the I/O Operation completes.

Async Syntax

async fn fetch_data(url: &str) -> Result<String, reqwest::Error> {
    let response = reqwest::get(url).await?;
    Ok(response.text().await?)
}

async fn main() {
    match fetch_data("https://example.com").await {
        Ok(data) => println!("received {} bytes", data.len()),
        Err(e) => eprintln!("error: {}", e),
    }
}

async fn desugars to a function that returns an impl Future<Output = T>. .await desugars to a State machine transition in the generated future.

What `.await` Does

The .await keyword yields control to the executor. When the future is not ready, it returns Poll::Pending and saves its state. When the waker fires, the executor polls the future again, and Execution resumes from the .await point.

Time →
┌──────────────────────────────────────────────────┐
│ Future execution:                                 │
│                                                   │
│  start ──► .await ──► [yield] ──► resume ──► end  │
│           (pending)   (woken)                     │
│                                                   │
│  While yielded, the executor runs OTHER futures    │
└──────────────────────────────────────────────────┘

`Pin` and Unpin

Pin is a wrapper that prevents the wrapped pointer from being moved. This is necessary because Async futures contain self-referential data (pointers from the state machine to local variables on The stack). If the future were moved, these pointers would become invalid.

use std::pin::Pin;
use std::marker::PhantomPinned;

struct SelfReferential {
    data: String,
    pointer: *const String,  // points to data
    _marker: PhantomPinned,
}

Most types are Unpin — they can be safely moved even when pinned. Types that are self-referential (like the compiler-generated state machine for async blocks) are !Unpin.

`Send` and `Sync` Bounds for Futures

A future must be Send to be spawned on an async runtime (like tokio). If a future captures a Non-Send type (like Rc or &RefCell), it cannot be spawned:

use std::rc::Rc;

async fn bad() {
    let rc = Rc::new(42);
    // This future is !Send because Rc is !Send
    // tokio::spawn(async move { println!("{}", rc) });  // ERROR
}

Tokio Runtime

Tokio is the dominant async runtime for Rust. It provides a multi-threaded executor, I/O driver, and Timer.

Basic Tokio Usage

[dependencies]
tokio = { version = "1", features = ["full"] }

use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let handle = tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        42
    });

    let result = handle.await.unwrap();
    assert_eq!(result, 42);
}

Tokio Task Spawning

#[tokio::main]
async fn main() {
    let handle1 = tokio::spawn(async {
        // runs on the tokio runtime
        tokio::time::sleep(Duration::from_millis(100)).await;
        "task 1"
    });

    let handle2 = tokio::spawn(async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        "task 2"
    });

    assert_eq!(handle2.await.unwrap(), "task 2");  // finishes first
    assert_eq!(handle1.await.unwrap(), "task 1");
}

`spawn_blocking`

For CPU-bound work that would block the async executor:

#[tokio::main]
async fn main() {
    let result = tokio::task::spawn_blocking(|| {
        // This runs on a blocking thread pool
        // Does NOT block the async executor
        expensive_computation()
    }).await.unwrap();
}

Never run CPU-intensive or blocking I/O (like std::fs::read_to_string) directly on the async Executor — it will block all other tasks on that thread. Use spawn_blocking for blocking Operations and tokio::fs for async file I/O.

Async Channels

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);

    let tx2 = tx.clone();
    tokio::spawn(async move {
        tx.send("from task 1").await.unwrap();
    });

    tokio::spawn(async move {
        tx2.send("from task 2").await.unwrap();
    });

    while let Some(msg) = rx.recv().await {
        println!("{}", msg);
    }
}

Note: tokio’s async channels use .await for send/receive, unlike std::sync::mpsc which blocks.

`select!`

select! allows waiting on multiple async operations simultaneously and handles the first one to Complete:

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);

    tokio::spawn(async move {
        sleep(Duration::from_millis(100)).await;
        tx.send("delayed").await.unwrap();
    });

    tokio::select! {
        msg = rx.recv() => {
            println!("received: {:?}", msg);
        }
        _ = sleep(Duration::from_millis(50)) => {
            println!("timeout");
        }
    }
}

Restructure your code to recreate the futures. This is a common source of confusion for developers Coming from JavaScript’s Promise.race.

Race Conditions

Data Races vs Race Conditions

A data race is undefined behavior — two threads access the same memory location concurrently, at Least one of them writes, and there is no synchronization. Rust’s type system prevents data races at Compile time (in safe code).

A race condition is a logical error where the outcome depends on the timing of concurrent Operations. Race conditions are not prevented by the type system — they are logic bugs that require Careful design to avoid.

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

let counter = Arc::new(Mutex::new(0));
let mut handles = vec![];

for _ in 0..10 {
    let counter = Arc::clone(&counter);
    handles.push(thread::spawn(move || {
        let mut num = counter.lock().unwrap();
        // Check-then-act race condition:
        // Two threads could both read 0, both increment to 1, losing one increment
        if *num < 5 {
            *num += 1;
        }
    }));
}

// This is NOT a data race (Mutex provides synchronization)
// but it IS a race condition (logical bug)

Preventing Race Conditions

Minimize shared mutable state. Prefer message passing over shared memory.
Use atomic operations for simple counters and flags.
Hold locks for the shortest possible time. Do not perform I/O while holding a lock.
Design lock protocols (always acquire locks in the same order).
Use RwLock when reads vastly outnumber writes.

Common Concurrency Patterns

Thread Pool

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

type Job = Box<dyn FnOnce() + Send + 'static>;

struct ThreadPool {
    workers: Vec<Worker>,
    sender: mpsc::Sender<Job>,
}

struct Worker {
    id: usize,
    thread: Option<thread::JoinHandle<()>>,
}

impl ThreadPool {
    fn new(size: usize) -> Self {
        let (sender, receiver) = mpsc::channel();
        let receiver = Arc::new(Mutex::new(receiver));
        let mut workers = Vec::with_capacity(size);

        for id in 0..size {
            workers.push(Worker::new(id, Arc::clone(&receiver)));
        }

        ThreadPool { workers, sender }
    }

    fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        self.sender.send(Box::new(f)).unwrap();
    }
}

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Self {
        let thread = thread::spawn(move || loop {
            let job = receiver.lock().unwrap().recv();
            match job {
                Ok(job) => job(),
                Err(_) => break,
            }
        });

        Worker {
            id,
            thread: Some(thread),
        }
    }
}

Work Stealing (with `rayon`)

[dependencies]
rayon = "1.10"

use rayon::prelude::*;

let data: Vec<i32> = (1..=1_000_000).collect();

let sum: i64 = data.par_iter().map(|&x| x as i64).sum();
let even_count = data.par_iter().filter(|&&x| x % 2 == 0).count();

let mut result = vec![0i32; 1_000_000];
data.par_iter().enumerate().for_each(|(i, &x)| {
    result[i] = x * 2;
});

Rayon uses a work-stealing scheduler: each thread has a local deque of tasks. When a thread finishes Its work, it steals tasks from other threads’ deques. This provides automatic load balancing without Central coordination.

Async Mutex (tokio)

use tokio::sync::Mutex;

#[tokio::main]
async fn main() {
    let data = Arc::new(Mutex::new(0));

    let mut handles = vec![];
    for _ in 0..10 {
        let data = Arc::clone(&data);
        handles.push(tokio::spawn(async move {
            let mut lock = data.lock().await;
            *lock += 1;
        }));
    }

    for handle in handles {
        handle.await.unwrap();
    }

    assert_eq!(*data.lock().await, 10);
}

Blocks the entire OS thread, preventing other async tasks from running. Use tokio::sync::Mutex for Async contexts. However, if the critical section is short and does not contain any .awaitA std::sync::Mutex is acceptable and has lower overhead.

`Send` and `Sync` in Depth

Auto-Implementation Rules

The compiler automatically implements Send and Sync based on the composition of the type:

A type is Send if all its members are Send.
A type is Sync if all its members are Sync (equivalently, if &T is Send).

Type	`Send`	`Sync`
i32``f64``bool	Yes	Yes
String``Vec<T> where `T: Send + Sync`	Yes	Yes
`Box<T>` where `T: Send`	Yes	No (unless `T: Sync`)
`Rc<T>`	No	No
`Arc<T>` where `T: Send + Sync`	Yes	Yes
`&T` where `T: Sync`	Yes	Yes
`&mut T` where `T: Send`	Yes	No
`Mutex<T>` where `T: Send`	Yes	Yes

Manual Implementation

struct MyType {
    pointer: *const i32,  // raw pointers are !Send and !Sync
}

unsafe impl Send for MyType {}
unsafe impl Sync for MyType {}

Automatic analysis is wrong and that your type is actually safe to send/share across threads. If Your assertion is wrong, you have undefined behavior. Only do this when you can rigorously prove Thread safety (e.g., when using platform-specific synchronization primitives that the compiler Cannot see).

Common Pitfalls

Blocking the async executor. Calling std::thread::sleep``std::fs::read_to_stringOr any blocking operation inside an async task blocks the entire OS thread. All other tasks on that thread are stalled. Use tokio::time::sleep``tokio::fsOr spawn_blocking.
Holding a std::sync::Mutex across .await. This blocks the thread even while the future is suspended. Either use tokio::sync::Mutex or restructure the code to drop the lock before awaiting.
Deadlock with Mutex in async code. Two tasks each lock one mutex and then try to lock the other — classic deadlock. This is worse in async code because the executor cannot preempt the tasks. Always acquire locks in a consistent order, or use try_lock with backoff.
Arc reference cycles. Two Arc values that reference each other will never be dropped. Use Weak<T> to break cycles, especially in graph data structures and observer patterns.
Not using scoped threads when possible. thread::spawn requires 'static bounds, forcing you to move or Arc everything. thread::scope allows borrowing and is safer and more ergonomic.
Atomic ordering mistakes. Using Ordering::Relaxed when you need Release/Acquire semantics is a data race on the visibility of writes. The data is not corrupted (the operation is atomic), but other threads may see stale values.
Over-spawning threads. Each OS thread consumes stack space (8 MB default) and kernel resources. For I/O-bound concurrency, use async/await instead of threads. For CPU-bound parallelism, use a thread pool with a fixed number of threads ( equal to the number of CPU cores).
Ignoring JoinHandle errors. If a spawned thread panics, handle.join() returns Err. Ignoring this error silently swallows panics, which may indicate serious bugs. Always check join results or use a supervision mechanism.
Using Rc in async code. Rc is not SendSo any future capturing an Rc cannot be spawned on tokio. Use Arc instead. This is one of the most common async Rust compilation errors.
select! dropping futures. When select! completes, all non-selected branches are dropped. If you need to retry a branch, you must restructure your code to loop and recreate the future. Consider using tokio::select! with a loop pattern for repeated selection.

flowchart TD
    A[Rust Concurrency] --> B[Threads]
    A --> C[Message Passing]
    A --> D[Shared State]
    A --> E[Async/Await]
    B --> F[std::thread::spawn]
    B --> G[JoinHandle]
    C --> I[mpsc channels]
    C --> J[sync_channel]
    D --> L[Mutex]
    D --> M[RwLock]
    D --> N[Atomic]
    D --> O[Arc]
    E --> P[Future trait]
    E --> Q[async fn]
    E --> R[.await]
    E --> S[tokio runtime]
    L --> T[lock().unwrap()]
    N --> V[load/store/compare_exchange]
    O --> W[Reference-counted shared ownership]

Summary

This topic covers the core concepts of concurrency, including underlying theory, practical implementation, and key applications.

Key concepts include:

core concepts and terminology
algorithms and computational thinking
practical implementation
security and ethical considerations
applications in the real world

Understanding these concepts thoroughly is essential for both examinations and practical programming, and requires both theoretical knowledge and hands-on practice.

Worked Examples

Worked examples demonstrating the application of key concepts are covered in the detailed sub-pages linked above.

Intuition

Rust’s concurrency model is like a traffic system with strict rules. Each thread is a lane, and the type system is the traffic controller. Arc and Mutex are the traffic lights: Arc lets multiple threads share ownership of data, and Mutex ensures only one thread accesses it at a time. The compiler prevents data races before your program runs, which is like having a safety inspector who checks every vehicle before it enters the highway. Channels are the postal system: threads send messages instead of sharing memory, reducing contention.

Cross-References

Mark this page as reviewed