Send, Sync & Concurrency Patterns

Rust's claim of "fearless concurrency" rests on two marker traits. Once you understand them, compiler errors about threads stop being mysterious.

What Send and Sync mean

• T: Send — a value of type T can be moved to another thread.
• T: Sync — a &T can be shared between threads. Formally, T is Sync exactly when &T is Send.

Both are auto traits: the compiler implements them for your struct automatically if every field is Send/Sync. You almost never implement them by hand (doing so is unsafe).

thread::spawn and tokio::spawn require Send. That is how the compiler stops you from, say, cloning an Rc into two threads and racing on its count:

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

fn main() {
    let data = Rc::new(5);
    let handle = thread::spawn(move || println!("{data}"));
    handle.join().unwrap();
}

error[E0277]: `Rc<i32>` cannot be sent between threads safely

Switching to Arc fixes it. The check is purely at compile time; there is no runtime cost.

Atomics

For a single counter or flag, a mutex is heavier than necessary. The types in std::sync::atomic provide lock-free operations on integers, booleans and pointers:

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

fn main() {
    let hits = AtomicUsize::new(0);
    let saw_error = AtomicBool::new(false);

    thread::scope(|s| {
        for worker in 0..4 {
            let hits = &hits;
            let saw_error = &saw_error;
            s.spawn(move || {
                for _ in 0..1_000 {
                    hits.fetch_add(1, Ordering::Relaxed);
                }
                if worker == 3 {
                    saw_error.store(true, Ordering::Relaxed);
                }
            });
        }
    });

    println!("hits = {}", hits.load(Ordering::Relaxed));
    println!("error = {}", saw_error.load(Ordering::Relaxed));
}

Expected output:

hits = 4000
error = true

Ordering controls how an atomic operation orders other memory accesses around it. Relaxed is right for independent counters and statistics. When one thread publishes data and another consumes it, pair Release on the store with Acquire on the load. If you are not sure, SeqCst is the strongest and simplest to reason about. Don't build your own lock-free data structures unless you have to; reach for Mutex, channels or crates such as crossbeam first.

Data parallelism with rayon

For CPU-bound work over collections, rayon turns an iterator chain into a parallel one by changing iter() to par_iter(). It uses a work-stealing thread pool sized to your CPU count.

[dependencies]
rayon = "1"

use rayon::prelude::*;

fn is_prime(n: u64) -> bool {
    n >= 2 && (2..).take_while(|d| d * d <= n).all(|d| n % d != 0)
}

fn main() {
    let count = (1..2_000_000u64).into_par_iter().filter(|&n| is_prime(n)).count();
    println!("{count} primes below 2,000,000");

    let mut words = vec!["pear", "fig", "apple", "kiwi"];
    words.par_sort_unstable();
    println!("{words:?}");
}

Expected output:

148933 primes below 2,000,000
["apple", "fig", "kiwi", "pear"]

Rayon only helps when each item does enough work to outweigh coordination. The closure must be Send + Sync — the compiler enforces this, so you cannot accidentally mutate shared state without a lock or atomic.

A worker pool with std

When you need a fixed number of long-lived workers pulling jobs from a queue, share one receiver behind Arc<Mutex<_>>:

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

type Job = Box<dyn FnOnce() -> u64 + Send>;

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<Job>();
    let (result_tx, result_rx) = mpsc::channel::<u64>();
    let job_rx = Arc::new(Mutex::new(job_rx));

    let workers: Vec<_> = (0..3)
        .map(|_| {
            let job_rx = Arc::clone(&job_rx);
            let result_tx = result_tx.clone();
            thread::spawn(move || {
                loop {
                    // The guard is dropped at the end of this statement,
                    // so the lock is not held while the job runs.
                    let job = job_rx.lock().unwrap().recv();
                    match job {
                        Ok(job) => result_tx.send(job()).unwrap(),
                        Err(_) => break, // channel closed: no more jobs
                    }
                }
            })
        })
        .collect();
    drop(result_tx);

    for n in 1..=10u64 {
        job_tx.send(Box::new(move || n * n)).unwrap();
    }
    drop(job_tx); // lets the workers exit

    let total: u64 = result_rx.iter().sum();
    for w in workers {
        w.join().unwrap();
    }
    println!("sum of squares = {total}");
}

Expected output:

sum of squares = 385

Job is a boxed FnOnce + Send trait object, so any closure that owns thread-safe data can be queued.

Common mistakes

• Wrapping everything in Arc<Mutex<_>> by reflex. Prefer passing ownership through channels, thread::scope for borrowed data, or atomics for a single number.
• Writing let guard = rx.lock().unwrap(); let job = guard.recv(); in a worker. The lock is then held while waiting and while working, so the pool runs one job at a time.
• Using Relaxed to publish data to another thread. It orders nothing but the atomic itself; use Release/Acquire.
• Adding unsafe impl Send to silence an error. The error is usually right.

Practice

Use thread::scope and an AtomicU64 to sum the numbers 1..=1_000_000 across four threads, each handling a quarter of the range. Print the total.

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

fn main() {
    let total = AtomicU64::new(0);
    let chunk = 250_000u64;

    thread::scope(|s| {
        for i in 0..4 {
            let total = &total;
            s.spawn(move || {
                let start = i * chunk + 1;
                let end = (i + 1) * chunk;
                let partial: u64 = (start..=end).sum();
                total.fetch_add(partial, Ordering::Relaxed);
            });
        }
    });

    println!("{}", total.load(Ordering::Relaxed));
}