# Rust Concurrency

> When to activate: Rust threads, concurrency, Arc, Mutex, RwLock, channels, rayon, crossbeam, shared state, parallel processing

- Skill: `mattakushi432/rust-concurrency` (Agent Skill)
- Install (CLI): `npx skillmds@latest add mattakushi432/rust-concurrency`
- Raw SKILL.md: https://api.skillmd.com/api/skills/mattakushi432/rust-concurrency/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: Mattakushi432 (https://skillmd.com/u/mattakushi432)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/mattakushi432/rust-concurrency

---


# Rust Concurrency Patterns

## Threads and move Closures

```rust
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

```toml
[dependencies]
rayon = "1"
```

```rust
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<T>>

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

```rust
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

```rust
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

```toml
[dependencies]
crossbeam = "0.8"
```

```rust
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

```rust
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

```rust
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