Rust Concurrency

Rust Concurrency Patterns

Threads and move Closures

use std::thread;

fn spawn_threads() {
    let data = vec![1, 2, 3];

    let handle = thread::spawn(move || {
        // `move` captures `data` by value; thread owns it
        println!("from thread: {:?}", data);
        data.len()
    });

    let result = handle.join().expect("thread panicked");
    println!("thread returned: {result}");
}

Parallel Processing with rayon

[dependencies]
rayon = "1"

use rayon::prelude::*;

fn parallel_sum(data: &[i64]) -> i64 {
    data.par_iter().sum()
}

fn parallel_transform(items: Vec<String>) -> Vec<String> {
    items.par_iter()
        .filter(|s| !s.is_empty())
        .map(|s| s.to_uppercase())
        .collect()
}

Shared State: Arc<Mutex>

Use Arc for shared ownership across threads, Mutex for exclusive mutable access.

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

fn shared_counter() {
    let counter = Arc::new(Mutex::new(0u64));

    let handles: Vec<_> = (0..10).map(|_| {
        let counter = Arc::clone(&counter);
        thread::spawn(move || {
            let mut count = counter.lock().unwrap();
            *count += 1;
        })
    }).collect();

    for h in handles { h.join().unwrap(); }
    println!("final count: {}", *counter.lock().unwrap());
}

// RwLock for multiple readers / exclusive writer
use std::sync::RwLock;

struct Cache {
    data: Arc<RwLock<std::collections::HashMap<String, String>>>,
}

impl Cache {
    fn get(&self, key: &str) -> Option<String> {
        self.data.read().unwrap().get(key).cloned()
    }

    fn set(&self, key: String, value: String) {
        self.data.write().unwrap().insert(key, value);
    }
}

Channels for Message Passing

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

fn pipeline() {
    let (tx, rx) = mpsc::channel::<String>();

    let producer = thread::spawn(move || {
        for i in 0..5 {
            tx.send(format!("item {i}")).unwrap();
        }
    });

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

    producer.join().unwrap();
}

// Multiple producers
fn multi_producer() {
    let (tx, rx) = mpsc::channel::<u64>();

    for i in 0..4 {
        let tx = tx.clone();
        thread::spawn(move || tx.send(i * i).unwrap());
    }
    drop(tx); // close original sender

    let results: Vec<u64> = rx.into_iter().collect();
    println!("results: {:?}", results);
}

crossbeam for Advanced Patterns

[dependencies]
crossbeam = "0.8"

use crossbeam::channel::{bounded, select};
use std::thread;

fn work_stealing_pool() {
    let (work_tx, work_rx) = bounded::<String>(100);
    let (result_tx, result_rx) = bounded::<String>(100);

    for _ in 0..4 {
        let work_rx = work_rx.clone();
        let result_tx = result_tx.clone();
        thread::spawn(move || {
            for task in work_rx {
                result_tx.send(process(task)).unwrap();
            }
        });
    }
    drop(result_tx);

    for i in 0..20 {
        work_tx.send(format!("task {i}")).unwrap();
    }
    drop(work_tx);

    for result in result_rx {
        println!("{result}");
    }
}

Atomic Operations

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

fn atomic_flag() {
    let running = Arc::new(AtomicBool::new(true));
    let counter = Arc::new(AtomicUsize::new(0));

    let r = Arc::clone(&running);
    let c = Arc::clone(&counter);

    let handle = std::thread::spawn(move || {
        while r.load(Ordering::Relaxed) {
            c.fetch_add(1, Ordering::SeqCst);
        }
    });

    std::thread::sleep(std::time::Duration::from_millis(10));
    running.store(false, Ordering::SeqCst);
    handle.join().unwrap();
    println!("count: {}", counter.load(Ordering::SeqCst));
}

OnceLock for Lazy Initialization

use std::sync::OnceLock;

static CONFIG: OnceLock<Config> = OnceLock::new();

fn get_config() -> &'static Config {
    CONFIG.get_or_init(|| {
        Config::load_from_env().expect("failed to load config")
    })
}

Common Anti-Patterns

• Lock poisoning ignored with .unwrap() — at least log it; consider .unwrap_or_else(|e| e.into_inner())
• Holding a lock across I/O or long computation — keep critical sections minimal
• Shared Mutex<Vec<T>> where channels suffice — channels are often clearer than shared state
• thread::sleep for synchronization — use channels, Condvar, or atomics instead
• Using unsafe to share raw pointers — wrap in Arc<Mutex<T>> or redesign the API