Concurrency

Threads

A Rust program starts with one thread, the one that runs main. std::thread::spawn starts another operating system thread that runs a closure (see Iterators and Closures) and returns a JoinHandle. Calling join() on the handle waits until the thread has finished and gives back the closure's return value, wrapped in a Result: Err if the thread panicked. The general concepts of processes and threads are on Concurrency Basics.

use std::thread;

fn main() {
    // three batches of readings in Wh, one thread per batch
    let batches = vec![vec![350, 412, 120], vec![98, 131], vec![260, 245, 110]];

    let mut handles = Vec::new();
    for batch in batches {
        handles.push(thread::spawn(move || batch.iter().sum::<u32>()));
    }

    // join in the order the threads were started, and print here in main
    let mut total = 0;
    for (n, handle) in handles.into_iter().enumerate() {
        let sum = handle.join().unwrap();
        println!("batch {n}: {sum} Wh");
        total += sum;
    }
    println!("total {total} Wh");
}

batch 0: 882 Wh
batch 1: 229 Wh
batch 2: 615 Wh
total 1726 Wh

Each thread receives its own batch: move moves the vector into the closure, so the thread owns it. The results come back through join, in the order the handles are joined, and main does the printing. That is deliberate. Threads run in whatever order the operating system schedules them, and four threads that each print a line give different orders on different runs. Collect results and print them in one place, or sort them first.

Why spawn needs move

A spawned thread can keep running after the function that started it has returned. If its closure borrowed a local variable, that borrow could outlive the variable. thread::spawn therefore requires a closure that owns everything it uses (in Rust terms, one that is 'static), and the compiler refuses a borrowing closure:

use std::thread;

fn main() {
    let meter = String::from("M-30004");
    let handle = thread::spawn(|| {
        println!("reading {meter}");
    });
    handle.join().unwrap();
}

example.rs:5:32: error[E0373]: closure may outlive the current function, but it borrows `meter`, which is owned by the current function: may outlive borrowed value `meter`

$ rustc --edition 2024 example.rs
error[E0373]: closure may outlive the current function, but it borrows `meter`, which is owned by the current function
 --> example.rs:5:32
  |
5 |     let handle = thread::spawn(|| {
  |                                ^^ may outlive borrowed value `meter`
6 |         println!("reading {meter}");
  |                            ----- `meter` is borrowed here
  |
note: function requires argument type to outlive `'static`
 --> example.rs:5:18
  |
5 |       let handle = thread::spawn(|| {
  |  __________________^
6 | |         println!("reading {meter}");
7 | |     });
  | |______^
help: to force the closure to take ownership of `meter` (and any other referenced variables), use the `move` keyword
  |
5 |     let handle = thread::spawn(move || {
  |                                ++++

error: aborting due to 1 previous error

For more information about this error, try `rustc --explain E0373`.

The help line gives the fix: move ||. The note explains why: spawn "requires argument type to outlive 'static", the lifetime of the whole program (see Lifetimes). After the move, main can no longer use meter; if both need it, clone it first or share it with an Arc.

Scoped threads

Often you want a few threads to work on data that already exists in the current function, and to wait for all of them before going on. A scoped thread, started inside std::thread::scope (stable since Rust 1.63, August 2022), does exactly that: every thread spawned in the scope is joined automatically before scope returns, so the compiler knows the threads cannot outlive the local data, and they may borrow it without move.

use std::thread;

fn main() {
    let readings = vec![350, 412, 120, 98, 131, 260];
    let (first, second) = readings.split_at(readings.len() / 2);

    // scoped threads may borrow local data: they all end before scope returns
    let (a, b) = thread::scope(|s| {
        let left = s.spawn(|| first.iter().sum::<u32>());
        let right = s.spawn(|| second.iter().sum::<u32>());
        (left.join().unwrap(), right.join().unwrap())
    });

    println!("{a} + {b} = {}", a + b);
    println!("readings is still here: {} values", readings.len());
}

882 + 489 = 1371
readings is still here: 6 values

Both threads borrow halves of readings, and the vector is still usable afterward. Scoped threads are the simplest way to split work on a slice across CPU cores. Joining them explicitly, as here, is only needed to get their results; the scope joins any thread you leave alone.

Data races, Send and Sync

A data race happens when two threads access the same memory at the same time, at least one of them writes, and nothing orders the accesses. The result depends on timing: an update can be lost, a reader can see a half-written value. In C and C++ a data race is undefined behavior (see race conditions for the general picture). Rust rules them out in safe code with the same tool as the rest of the language, the borrowing rules: many readers or one writer. Two threads that both want to change a local variable need two mutable borrows at once:

use std::thread;

fn main() {
    let mut total = 0;

    thread::scope(|s| {
        s.spawn(|| total += 350); // both threads want to change total
        s.spawn(|| total += 412);
    });
    println!("{total}");
}

example.rs:8:17: error[E0499]: cannot borrow `total` as mutable more than once at a time: second mutable borrow occurs here

$ rustc --edition 2024 example.rs
error[E0499]: cannot borrow `total` as mutable more than once at a time
 --> example.rs:8:17
  |
6 |     thread::scope(|s| {
  |                    - has type `&'1 Scope<'1, '_>`
7 |         s.spawn(|| total += 350); // both threads want to change total
  |         ------------------------
  |         |       |  |
  |         |       |  first borrow occurs due to use of `total` in closure
  |         |       first mutable borrow occurs here
  |         argument requires that `total` is borrowed for `'1`
8 |         s.spawn(|| total += 412);
  |                 ^^ ----- second borrow occurs due to use of `total` in closure
  |                 |
  |                 second mutable borrow occurs here
  |
note: requirement that the value outlives `'1` introduced here
 --> /rustc/…/library/std/src/thread/scoped.rs:203:34

error: aborting due to 1 previous error

For more information about this error, try `rustc --explain E0499`.

The Rust Book calls its concurrency chapter "Fearless Concurrency" for this reason: mistakes that are found by chance, under load, in other languages are compile errors here. Two marker traits carry the rules to every type. Marker traits have no methods; the compiler implements them automatically for a type whose parts all have them.

• The Send trait means a value of the type may be moved to another thread. Almost every type is Send. Rc is not, because two threads updating its plain counter at once could corrupt it.
• The Sync trait means a reference to the type (&T) may be used from several threads at once. Cell and RefCell are not Sync, because their changes through a shared reference are not synchronized. Mutex<T> is.

thread::spawn requires its closure to be Send, so a type that is unsafe to move to another thread cannot get there:

use std::rc::Rc;
use std::thread;

fn main() {
    let tariff = Rc::new(String::from("fixed"));
    let copy = Rc::clone(&tariff);
    let worker = thread::spawn(move || println!("worker sees {copy}"));
    worker.join().unwrap();
    println!("main sees {tariff}");
}

example.rs:7:32: error[E0277]: `Rc<String>` cannot be sent between threads safely: `Rc<String>` cannot be sent between threads safely

$ rustc --edition 2024 example.rs
error[E0277]: `Rc<String>` cannot be sent between threads safely
 --> example.rs:7:32
  |
7 |     let worker = thread::spawn(move || println!("worker sees {copy}"));
  |                  ------------- -------^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |                  |             |
  |                  |             `Rc<String>` cannot be sent between threads safely
  |                  |             within this `{closure@example.rs:7:32: 7:39}`
  |                  required by a bound introduced by this call
  |
  = help: within `{closure@example.rs:7:32: 7:39}`, the trait `Send` is not implemented for `Rc<String>`
note: required because it's used within this closure
 --> example.rs:7:32
  |
7 |     let worker = thread::spawn(move || println!("worker sees {copy}"));
  |                                ^^^^^^^
note: required by a bound in `spawn`
 --> /rustc/…/library/std/src/thread/functions.rs:125:0

error: aborting due to 1 previous error

For more information about this error, try `rustc --explain E0277`.

Sharing a RefCell between scoped threads fails the same way, on Sync, and the compiler's note suggests the thread-safe alternative ("if you want to do aliasing and mutation between multiple threads, use std::sync::RwLock instead"):

use std::cell::RefCell;
use std::thread;

fn main() {
    let total = RefCell::new(0);
    thread::scope(|s| {
        s.spawn(|| *total.borrow_mut() += 350);
        s.spawn(|| *total.borrow_mut() += 412);
    });
    println!("{}", total.borrow());
}

example.rs:7:17: error[E0277]: `RefCell<i32>` cannot be shared between threads safely: `RefCell<i32>` cannot be shared between threads safely

The one-thread types are not mistakes: they are faster because they skip synchronization, and the compiler makes sure they stay in one thread. The smart pointers themselves are explained on Smart Pointers.

Shared state: Arc and Mutex

A mutex (mutual exclusion) is a lock that lets only one thread at a time use the data it protects. Rust's std::sync::Mutex<T> differs from a mutex in C or Java in one important way: it contains the data. There is no way to reach the T without calling lock(), which waits until the lock is free and returns a guard. The guard acts like &mut T, and dropping it unlocks the mutex. Forgetting to lock cannot be written, and unlocking happens by itself when the guard goes out of scope.

To give several threads the same mutex, put it in an Arc (atomically reference counted) and clone the Arc for each thread:

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

fn main() {
    let total = Arc::new(Mutex::new(0u64)); // Wh, shared by all threads
    let batches = vec![vec![350, 412, 120], vec![98, 131], vec![260, 245, 110]];

    let mut handles = Vec::new();
    for batch in batches {
        let total = Arc::clone(&total);
        handles.push(thread::spawn(move || {
            let sum: u64 = batch.iter().sum();
            let mut guard = total.lock().unwrap(); // wait for the lock
            *guard += sum;
        })); // guard dropped at the end of the closure: lock released
    }
    for handle in handles {
        handle.join().unwrap();
    }
    println!("total {} Wh", *total.lock().unwrap());
}

total 1726 Wh

The guard is dropped at the end of the closure, which releases the lock. Keep that span short: compute the sum first, lock only for the update, as here. A thread that holds a lock while it does slow work (reading a file, waiting for the network) makes all the others wait. With scoped threads the Arc is not even needed, because the threads may borrow a local Mutex.

Atomics and RwLock

For a single counter or flag, a mutex is more than you need. The atomic types in std::sync::atomic (AtomicU64, AtomicBool, …) are changed with atomic CPU operations that other threads can never see half done, such as fetch_add:

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

fn main() {
    let telegrams = AtomicU64::new(0); // a counter many threads may bump

    thread::scope(|s| {
        for _ in 0..4 {
            s.spawn(|| {
                for _ in 0..1000 {
                    telegrams.fetch_add(1, Ordering::Relaxed);
                }
            });
        }
    });
    println!("{} telegrams counted", telegrams.load(Ordering::Relaxed));
}

4000 telegrams counted

Four threads add 1000 each, and no update is lost. The Ordering argument says what other memory accesses must be ordered around the atomic one; Relaxed is enough for a counter that is read after all threads have finished, and anything more subtle is a topic for specialist literature. RwLock<T> is the other variation: many readers at the same time, or one writer, which pays off for data that is read often and changed rarely, such as a tariff table.

Lock poisoning

What if a thread panics while it holds a lock? It may have left the data half updated. Rust's mutex remembers this: lock poisoning marks a Mutex or RwLock whose guard was dropped during a panic, and from then on lock() returns an Err instead of the guard. That is why the examples write lock().unwrap(): if another thread panicked with the lock held, this thread panics too instead of continuing with possibly broken data. The example shows the panic message of the worker (on standard error) together with the program's own output:

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

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

    let t = Arc::clone(&total);
    let result = thread::spawn(move || {
        let mut guard = t.lock().unwrap();
        *guard += 350;
        panic!("meter offline"); // panics while holding the lock
    })
    .join();

    println!("worker failed: {}", result.is_err());
    println!("poisoned: {}", total.is_poisoned());
    match total.lock() {
        Ok(guard) => println!("total {}", *guard),
        Err(poisoned) => {
            println!("lock error: {poisoned}");
            let value = *poisoned.into_inner(); // look at the data anyway
            println!("value left behind: {value}");
        }
    }
}

thread '<unnamed>' (3216140) panicked at example.rs:11:9:
meter offline
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
worker failed: true
poisoned: true
lock error: poisoned lock: another task failed inside
value left behind: 350

The panic only ended the worker thread; main learned about it through join, which returned Err, and through the poisoned lock. When you know how to repair or ignore a half-done update, into_inner() on the error gives you the guard anyway. Spawned threads are named '<unnamed>' in panic messages; thread::Builder::new().name(…) gives them a name that makes logs easier to read.

Message passing with channels

Instead of sharing a value, threads can hand values to each other. Message passing is a style of concurrency in which threads communicate only by sending messages over a channel, so each piece of data has one owner at a time. Rust's std::sync::mpsc module (multiple producer, single consumer) provides it: channel() returns a sender tx and a receiver rx. send moves a value into the channel; recv, or a for loop over rx, takes values out in the order they were sent.

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

fn main() {
    let (tx, rx) = mpsc::channel();

    let worker = thread::spawn(move || {
        for kwh in [350, 412, 120] {
            tx.send(kwh).unwrap(); // the value moves into the channel
        }
        // tx is dropped here: the channel closes once nobody can send
    });

    for kwh in rx {
        // receives in sending order, ends when the channel is closed
        println!("received {kwh} kWh");
    }
    worker.join().unwrap();
}

received 350 kWh
received 412 kWh
received 120 kWh

Sending moves the value. The numbers here are simply copied, but a String or a Vec that is sent can no longer be used by the sending thread, so two threads can never change it at the same time. The loop over rx ends when every sender has been dropped and the channel is empty. mpsc::sync_channel(n) creates a bounded channel instead: send then waits when n messages are queued, which keeps a fast producer from filling memory.

Several producers are made by cloning the sender. In this example, a kestrel-readings tool reads lines of the form meter;kwh;tariff; each worker thread parses one batch and sends the results, and main totals them per tariff. The meter numbers and amounts are example values:

use std::collections::BTreeMap;
use std::sync::mpsc;
use std::thread;

/// "M-30101;350;fixed" -> ("fixed", 350); anything else is an error
fn parse(line: &str) -> Result<(String, u64), String> {
    let mut fields = line.split(';');
    let (Some(_meter), Some(kwh), Some(tariff), None) =
        (fields.next(), fields.next(), fields.next(), fields.next())
    else {
        return Err(format!("{line:?}: expected meter;kwh;tariff"));
    };
    let kwh = kwh.parse::<u64>().map_err(|e| format!("{line:?}: {e}"))?;
    Ok((tariff.to_string(), kwh))
}

fn main() {
    // two batches of lines, as two files would deliver them
    let batches = vec![
        vec!["M-30101;350;fixed", "M-30102;412;day_night", "M-30103;12O;dynamic"],
        vec!["M-30104;98;fixed", "M-30105;131", "M-30106;260;day_night"],
    ];

    let (tx, rx) = mpsc::channel();
    for batch in batches {
        let tx = tx.clone(); // one sender per worker
        thread::spawn(move || {
            for line in batch {
                tx.send(parse(line)).unwrap();
            }
        });
    }
    drop(tx); // main's own sender: without this the loop below never ends

    let mut totals: BTreeMap<String, u64> = BTreeMap::new();
    let mut errors = Vec::new();
    for result in rx {
        match result {
            Ok((tariff, kwh)) => *totals.entry(tariff).or_insert(0) += kwh,
            Err(e) => errors.push(e),
        }
    }
    errors.sort(); // arrival order depends on the threads: sort before printing

    for (tariff, kwh) in &totals {
        println!("{tariff:<10} {kwh:>4} kWh");
    }
    for e in &errors {
        println!("rejected {e}");
    }
}

day_night   672 kWh
fixed       448 kWh
rejected "M-30103;12O;dynamic": invalid digit found in string
rejected "M-30105;131": expected meter;kwh;tariff

Three details make this output the same on every run:

• The totals go into a BTreeMap, which iterates in key order. A HashMap would print in a different order on each run (see Collections and Strings).
• The rejected lines arrive in an order that depends on the threads, so they are sorted before printing.
• drop(tx) removes main's own sender. Without it, one sender would stay alive, and the for loop would wait forever for more messages.

The let … else in parse requires exactly three fields and returns an error otherwise (see Control Flow and Pattern Matching), and errors travel through the channel as ordinary Result values (see Error Handling).

What Rust does not prevent

"Fearless" is about data races, not about every concurrency bug. The Rust Reference lists deadlocks among the behavior that is "not considered unsafe": the compiler accepts code that can deadlock, and it is up to you to avoid it.

• Deadlock: thread 1 holds lock A and waits for B while thread 2 holds B and waits for A. Avoid it by always taking locks in the same order, by holding at most one lock at a time where you can, and by keeping guards short-lived.
• Locking the same Mutex twice in one thread: the standard library documents that the second lock() will not return (it might deadlock or panic). try_lock() reports the situation instead of waiting:

use std::sync::Mutex;

fn main() {
    let total = Mutex::new(0u64);

    let guard = total.lock().unwrap();
    // A second lock() here, in the same thread, would never return.
    // try_lock() reports the problem instead of waiting forever:
    println!("try_lock while held: {}", total.try_lock().is_err());
    drop(guard);
    println!("try_lock after drop: {}", total.try_lock().is_ok());
}

try_lock while held: true
try_lock after drop: true

• Race conditions in the logic: "read the balance, decide, write the new balance" with the lock released in between is free of data races and still wrong when two threads interleave. Hold the lock for the whole check-and-update.
• Leaks (an Arc cycle, a thread that never ends) and panics are not prevented either, and a channel whose receiver has been dropped makes send return an error.

A first look at async

Threads are a good fit for work that keeps the CPU busy. A server that waits on thousands of network connections spends most of its time waiting, and one operating system thread per connection becomes expensive. Rust's answer is async/await: an async fn returns a future, a value that represents work that is not done yet, and .await inside another async function waits for a future without blocking the thread, so one thread can switch between many tasks (see asynchronous programming). async/.await is stable since Rust 1.39 (November 2019), and in the 2024 edition the Future trait is in the prelude.

Futures are lazy, like iterators: calling an async function runs none of its body. The compiler warns when a future is created and never used:

async fn read_power(home: &str) -> u32 {
    println!("reading {home}");
    1234
}

fn main() {
    read_power("h0017");
}

example.rs:7:5: error: unused implementer of `Future` that must be used

Something has to poll a future, asking it to make progress, until it reports Ready. That something is an executor, and the standard library deliberately does not include one. To show the mechanism, this example polls a future once with a do-nothing waker; it works only because nothing in it really waits:

use std::future::Future;
use std::pin::pin;
use std::task::{Context, Poll, Waker};

async fn read_power(home: &str) -> u32 {
    println!("  reading {home}");
    1234 // watts; a real version would wait for the network here
}

async fn report() -> u32 {
    let a = read_power("h0017").await;
    let b = read_power("h0042").await;
    a + b
}

/// A toy executor: polls a future once. Real programs use a runtime instead.
fn poll_once<F: Future>(future: F) -> Poll<F::Output> {
    let mut future = pin!(future);
    let mut context = Context::from_waker(Waker::noop());
    future.as_mut().poll(&mut context)
}

fn main() {
    let future = report();
    println!("created the future; nothing has run yet");
    println!("{:?}", poll_once(future));
}

created the future; nothing has run yet
  reading h0017
  reading h0042
Ready(2468)

Real programs use a runtime crate that provides the executor, timers and asynchronous network I/O, most often Tokio. A Tokio program looks like this (not run here: this site's examples have no crates.io access):

// Cargo.toml: [dependencies] tokio = { version = "1", features = ["full"] }
use std::time::Duration;

async fn read_power(home: &str) -> u32 {
    tokio::time::sleep(Duration::from_millis(100)).await; // waits without blocking
    1234
}

#[tokio::main]
async fn main() {
    let a = tokio::spawn(read_power("h0017")); // two tasks, one or a few threads
    let b = tokio::spawn(read_power("h0042"));
    let total = a.await.unwrap() + b.await.unwrap();
    println!("total {total} W");
}

#[tokio::main] wraps main in a runtime, and tokio::spawn starts a lightweight task instead of a thread. Two warnings for async code: never call a blocking function (std::thread::sleep, a long computation, a blocking file read) inside an async task, because it stops every other task on that thread; and do not hold a std::sync::Mutex guard across an .await. Use plain threads unless you have many concurrent waits; they are simpler to reason about and to debug.