M07 Concurrency

Concurrency

Layer 1: Language Mechanics

Core Question

Is this CPU-bound or I/O-bound, and what's the sharing model?

Before choosing concurrency primitives:

• What's the workload type?
• What data needs to be shared?
• What's the thread safety requirement?

---

Error → Design Question

---

Thinking Prompt

Before adding concurrency:

1. What's the workload?

• CPU-bound → threads (std::thread, rayon)
• I/O-bound → async (tokio, async-std)
• Mixed → hybrid approach

2. What's the sharing model?

• No sharing → message passing (channels)
• Immutable sharing → Arc<T>
• Mutable sharing → Arc<Mutex<T>> or Arc<RwLock<T>>

3. What are the Send/Sync requirements?

• Cross-thread ownership → Send
• Cross-thread references → Sync
• Single-thread async → spawn_local

---

Trace Up ↑ (MANDATORY)

CRITICAL: Don't just fix the error. Trace UP to find domain constraints.

Domain Detection Table

Example: Web API + Rc Error

"Rc cannot be sent between threads" in Web API context
    ↑ DETECT: "Web API" → Load domain-web
    ↑ FIND: domain-web says "Shared state must be thread-safe"
    ↑ FIND: domain-web says "Rc in state" is Common Mistake
    ↓ DESIGN: Use Arc<T> with State extractor
    ↓ IMPL: axum::extract::State<Arc<AppConfig>>

Generic Trace

"Send not satisfied for my type"
    ↑ Ask: What domain is this? Load domain-* skill
    ↑ Ask: Does this type need to cross thread boundaries?
    ↑ Check: m09-domain (is the data model correct?)

---

Trace Down ↓

From design to implementation:

"Need parallelism for CPU work"
    ↓ Use: std::thread or rayon

"Need concurrency for I/O"
    ↓ Use: async/await with tokio

"Need to share immutable data across threads"
    ↓ Use: Arc<T>

"Need to share mutable data across threads"
    ↓ Use: Arc<Mutex<T>> or Arc<RwLock<T>>
    ↓ Or: channels for message passing

"Need simple atomic operations"
    ↓ Use: AtomicBool, AtomicUsize, etc.

---

Send/Sync Markers

Quick Reference

Decision Flowchart

What type of work?
├─ CPU-bound → std::thread or rayon
├─ I/O-bound → async/await
└─ Mixed → hybrid (spawn_blocking)

Need to share data?
├─ No → message passing (channels)
├─ Immutable → Arc<T>
└─ Mutable →
   ├─ Read-heavy → Arc<RwLock<T>>
   └─ Write-heavy → Arc<Mutex<T>>
   └─ Simple counter → AtomicUsize

Async context?
├─ Type is Send → tokio::spawn
├─ Type is !Send → spawn_local
└─ Blocking code → spawn_blocking

---

Common Errors

---

Anti-Patterns

---

Async-Specific Patterns

Avoid MutexGuard Across Await

// Bad: guard held across await
let guard = mutex.lock().await;
do_async().await;  // guard still held!

// Good: scope the lock
{
    let guard = mutex.lock().await;
    // use guard
}  // guard dropped
do_async().await;

Non-Send Types in Async

// Rc is !Send, can't cross await in spawned task
// Option 1: use Arc instead
// Option 2: use spawn_local (single-thread runtime)
// Option 3: ensure Rc is dropped before .await

---

Related Skills

Concurrency: Comparison with Other Languages

Rust vs Go

Concurrency Model

Goroutines vs Rust Tasks

// Rust: explicit about thread safety
use std::sync::Arc;
use tokio::sync::Mutex;

let data = Arc::new(Mutex::new(vec![]));
let data_clone = Arc::clone(&data);

tokio::spawn(async move {
    let mut guard = data_clone.lock().await;
    guard.push(1);  // Safe: Mutex protects access
});

// Go: implicit sharing (potential race)
// data := []int{}
// go func() {
//     data = append(data, 1)  // RACE CONDITION!
// }()

Channel Comparison

// Rust: typed channels with ownership
use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel::<String>(100);

tokio::spawn(async move {
    tx.send("hello".to_string()).await.unwrap();
    // tx is moved, can't be used elsewhere
});

// Go: channels are more flexible but less safe
// ch := make(chan string, 100)
// go func() {
//     ch <- "hello"
//     // ch can still be used anywhere
// }()

---

Rust vs Java

Thread Safety Model

Synchronization Comparison

// Rust: lock is tied to data
use std::sync::Mutex;

let data = Mutex::new(vec![1, 2, 3]);
{
    let mut guard = data.lock().unwrap();
    guard.push(4);
}  // lock released automatically

// Java: lock and data are separate
// List<Integer> data = new ArrayList<>();
// synchronized(data) {
//     data.add(4);
// }  // easy to forget synchronization elsewhere

Thread Pool Comparison

// Rust: rayon for data parallelism
use rayon::prelude::*;

let sum: i32 = (0..1000)
    .into_par_iter()
    .map(|x| x * x)
    .sum();

// Java: Stream API
// int sum = IntStream.range(0, 1000)
//     .parallel()
//     .map(x -> x * x)
//     .sum();

---

Rust vs C++

Safety Guarantees

Atomic Comparison

// Rust: clear memory ordering
use std::sync::atomic::{AtomicI32, Ordering};

let counter = AtomicI32::new(0);
counter.fetch_add(1, Ordering::SeqCst);
let value = counter.load(Ordering::Acquire);

// C++: similar but without safety
// std::atomic<int> counter{0};
// counter.fetch_add(1, std::memory_order_seq_cst);
// int value = counter.load(std::memory_order_acquire);

Mutex Comparison

// Rust: data protected by Mutex
use std::sync::Mutex;

struct SafeCounter {
    count: Mutex<i32>,  // Mutex contains the data
}

impl SafeCounter {
    fn increment(&self) {
        *self.count.lock().unwrap() += 1;
    }
}

// C++: mutex separate from data (error-prone)
// class Counter {
//     std::mutex mtx;
//     int count;  // NOT protected by type system
// public:
//     void increment() {
//         std::lock_guard<std::mutex> lock(mtx);
//         count++;
//     }
//     void unsafe_increment() {
//         count++;  // Compiles! But wrong.
//     }
// };

---

Async Models Comparison

Rust vs JavaScript Async

// Rust: async requires explicit runtime, can use multiple threads
#[tokio::main]
async fn main() {
    let results = tokio::join!(
        fetch("url1"),  // runs concurrently
        fetch("url2"),
    );
}

// JavaScript: single-threaded event loop
// async function main() {
//     const results = await Promise.all([
//         fetch("url1"),
//         fetch("url2"),
//     ]);
// }

Rust vs Python Async

// Rust: true parallelism possible
#[tokio::main(flavor = "multi_thread")]
async fn main() {
    let handles: Vec<_> = urls
        .into_iter()
        .map(|url| tokio::spawn(fetch(url)))  // spawns on thread pool
        .collect();

    for handle in handles {
        let _ = handle.await;
    }
}

// Python: asyncio is single-threaded (use ProcessPoolExecutor for CPU)
# async def main():
#     tasks = [asyncio.create_task(fetch(url)) for url in urls]
#     await asyncio.gather(*tasks)  # all on same thread

---

Send and Sync: Rust's Unique Feature

No other mainstream language has compile-time thread safety markers:

Why This Matters

// Rust PREVENTS this at compile time:
use std::rc::Rc;

let rc = Rc::new(42);
std::thread::spawn(move || {
    println!("{}", rc);  // ERROR: Rc is not Send
});

// In other languages, this would be a runtime bug:
// - Go: race detector might catch it
// - Java: undefined behavior
// - Python: GIL usually saves you
// - C++: undefined behavior

---

Performance Characteristics

When to Use What

---

Mental Model Shifts

From Go

Before: "Just use goroutines and channels"
After:  "Explicitly declare what can be shared and how"

Key shifts:

• Arc<Mutex<T>> instead of implicit sharing
• Compiler enforces thread safety
• Async needs explicit runtime

From Java

Before: "synchronized everywhere, hope for the best"
After:  "Types encode thread safety, compiler enforces"

Key shifts:

• No need for synchronized keyword
• Mutex contains data, not separate
• No GC pauses in critical sections

From C++

Before: "Be careful, read the docs, use sanitizers"
After:  "Compiler catches data races, trust the type system"

Key shifts:

• Send/Sync replace convention
• RAII locks are mandatory, not optional
• Much harder to write incorrect concurrent code

Thread-Based Concurrency Patterns

Thread Spawning Best Practices

Basic Thread Spawn

use std::thread;

fn main() {
    let handle = thread::spawn(|| {
        println!("Hello from thread!");
        42  // return value
    });

    let result = handle.join().unwrap();
    println!("Thread returned: {}", result);
}

Named Threads for Debugging

use std::thread;

let builder = thread::Builder::new()
    .name("worker-1".to_string())
    .stack_size(32 * 1024);  // 32KB stack

let handle = builder.spawn(|| {
    println!("Thread name: {:?}", thread::current().name());
}).unwrap();

Scoped Threads (No 'static Required)

use std::thread;

fn process_data(data: &[u32]) -> Vec<u32> {
    thread::scope(|s| {
        let handles: Vec<_> = data
            .chunks(2)
            .map(|chunk| {
                s.spawn(|| {
                    chunk.iter().map(|x| x * 2).collect::<Vec<_>>()
                })
            })
            .collect();

        handles
            .into_iter()
            .flat_map(|h| h.join().unwrap())
            .collect()
    })
}

fn main() {
    let data = vec![1, 2, 3, 4, 5, 6];
    let result = process_data(&data);  // No 'static needed!
    println!("{:?}", result);
}

---

Shared State Patterns

Arc + Mutex (Read-Write)

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

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

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

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

    println!("Result: {}", *counter.lock().unwrap());
}

Arc + RwLock (Read-Heavy)

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

fn read_heavy_cache() {
    let cache = Arc::new(RwLock::new(vec![1, 2, 3]));

    // Many readers
    for i in 0..5 {
        let cache = Arc::clone(&cache);
        thread::spawn(move || {
            let data = cache.read().unwrap();
            println!("Reader {}: {:?}", i, *data);
        });
    }

    // Occasional writer
    {
        let cache = Arc::clone(&cache);
        thread::spawn(move || {
            let mut data = cache.write().unwrap();
            data.push(4);
            println!("Writer: added element");
        });
    }
}

Atomic for Simple Types

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

fn atomic_counter() {
    let counter = Arc::new(AtomicUsize::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        handles.push(thread::spawn(move || {
            for _ in 0..1000 {
                counter.fetch_add(1, Ordering::SeqCst);
            }
        }));
    }

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

    println!("Result: {}", counter.load(Ordering::SeqCst));
}

---

Channel Patterns

MPSC Channel

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

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

    // Multiple producers
    for i in 0..3 {
        let tx = tx.clone();
        thread::spawn(move || {
            for j in 0..5 {
                tx.send(format!("msg {}-{}", i, j)).unwrap();
            }
        });
    }
    drop(tx);  // Drop original sender

    // Single consumer
    for received in rx {
        println!("Got: {}", received);
    }
}

Sync Channel (Bounded)

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

fn bounded_channel() {
    let (tx, rx) = mpsc::sync_channel(2);  // buffer size 2

    thread::spawn(move || {
        for i in 0..5 {
            println!("Sending {}", i);
            tx.send(i).unwrap();  // blocks if buffer full
            println!("Sent {}", i);
        }
    });

    thread::sleep(std::time::Duration::from_millis(500));
    for received in rx {
        println!("Received: {}", received);
        thread::sleep(std::time::Duration::from_millis(100));
    }
}

---

Thread Pool Patterns

Using rayon for Parallel Iteration

use rayon::prelude::*;

fn parallel_map() {
    let numbers: Vec<i32> = (0..1000).collect();

    let squares: Vec<i32> = numbers
        .par_iter()  // parallel iterator
        .map(|x| x * x)
        .collect();

    println!("Processed {} items", squares.len());
}

fn parallel_filter_map() {
    let data: Vec<String> = get_data();

    let results: Vec<_> = data
        .par_iter()
        .filter(|s| !s.is_empty())
        .map(|s| expensive_process(s))
        .collect();
}

Custom Thread Pool with crossbeam

use crossbeam::channel;
use std::thread;

fn custom_pool(num_workers: usize) {
    let (tx, rx) = channel::bounded::<Box<dyn FnOnce() + Send>>(100);

    // Spawn workers
    let workers: Vec<_> = (0..num_workers)
        .map(|_| {
            let rx = rx.clone();
            thread::spawn(move || {
                while let Ok(task) = rx.recv() {
                    task();
                }
            })
        })
        .collect();

    // Submit tasks
    for i in 0..100 {
        tx.send(Box::new(move || {
            println!("Processing task {}", i);
        })).unwrap();
    }

    drop(tx);  // Close channel

    for worker in workers {
        worker.join().unwrap();
    }
}

---

Synchronization Primitives

Barrier (Wait for All)

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

fn barrier_example() {
    let barrier = Arc::new(Barrier::new(3));
    let mut handles = vec![];

    for i in 0..3 {
        let barrier = Arc::clone(&barrier);
        handles.push(thread::spawn(move || {
            println!("Thread {} starting", i);
            thread::sleep(std::time::Duration::from_millis(i as u64 * 100));

            barrier.wait();  // All threads wait here

            println!("Thread {} after barrier", i);
        }));
    }

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

Condvar (Condition Variable)

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

fn condvar_example() {
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    let pair_clone = Arc::clone(&pair);

    // Waiter thread
    let waiter = thread::spawn(move || {
        let (lock, cvar) = &*pair_clone;
        let mut started = lock.lock().unwrap();
        while !*started {
            started = cvar.wait(started).unwrap();
        }
        println!("Waiter: condition met!");
    });

    // Notifier
    thread::sleep(std::time::Duration::from_millis(100));
    let (lock, cvar) = &*pair;
    {
        let mut started = lock.lock().unwrap();
        *started = true;
    }
    cvar.notify_one();

    waiter.join().unwrap();
}

Once (One-Time Initialization)

use std::sync::Once;

static INIT: Once = Once::new();
static mut CONFIG: Option<Config> = None;

fn get_config() -> &'static Config {
    INIT.call_once(|| {
        unsafe {
            CONFIG = Some(load_config());
        }
    });
    unsafe { CONFIG.as_ref().unwrap() }
}

// Better: use once_cell or lazy_static
use once_cell::sync::Lazy;

static CONFIG: Lazy<Config> = Lazy::new(|| {
    load_config()
});

---

Error Handling in Threads

Handling Panics

use std::thread;

fn handle_panic() {
    let handle = thread::spawn(|| {
        panic!("Thread panicked!");
    });

    match handle.join() {
        Ok(_) => println!("Thread completed successfully"),
        Err(e) => {
            if let Some(s) = e.downcast_ref::<&str>() {
                println!("Thread panicked with: {}", s);
            } else if let Some(s) = e.downcast_ref::<String>() {
                println!("Thread panicked with: {}", s);
            } else {
                println!("Thread panicked with unknown error");
            }
        }
    }
}

Catching Panics

use std::panic;

fn catch_panic() {
    let result = panic::catch_unwind(|| {
        risky_operation()
    });

    match result {
        Ok(value) => println!("Success: {:?}", value),
        Err(_) => println!("Operation panicked, continuing..."),
    }
}

Async Patterns in Rust

Task Spawning

Basic Spawn

use tokio::task;

#[tokio::main]
async fn main() {
    // Spawn a task that runs concurrently
    let handle = task::spawn(async {
        expensive_computation().await
    });

    // Do other work while task runs
    other_work().await;

    // Wait for result
    let result = handle.await.unwrap();
}

Spawn with Shared State

use std::sync::Arc;
use tokio::sync::Mutex;

async fn process_with_state() {
    let state = Arc::new(Mutex::new(vec![]));

    let handles: Vec<_> = (0..10)
        .map(|i| {
            let state = Arc::clone(&state);
            tokio::spawn(async move {
                let mut guard = state.lock().await;
                guard.push(i);
            })
        })
        .collect();

    // Wait for all tasks
    for handle in handles {
        handle.await.unwrap();
    }
}

---

Select Pattern

Racing Multiple Futures

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

async fn first_response() {
    select! {
        result = fetch_from_server_a() => {
            println!("A responded first: {:?}", result);
        }
        result = fetch_from_server_b() => {
            println!("B responded first: {:?}", result);
        }
    }
}

Select with Timeout

use tokio::time::timeout;

async fn with_timeout() -> Result<Data, Error> {
    select! {
        result = fetch_data() => result,
        _ = sleep(Duration::from_secs(5)) => {
            Err(Error::Timeout)
        }
    }
}

// Or use timeout directly
async fn with_timeout2() -> Result<Data, Error> {
    timeout(Duration::from_secs(5), fetch_data())
        .await
        .map_err(|_| Error::Timeout)?
}

Select with Channel

use tokio::sync::mpsc;

async fn process_messages(mut rx: mpsc::Receiver<Message>) {
    loop {
        select! {
            Some(msg) = rx.recv() => {
                handle_message(msg).await;
            }
            _ = tokio::signal::ctrl_c() => {
                println!("Shutting down...");
                break;
            }
        }
    }
}

---

Channel Patterns

MPSC (Multi-Producer, Single-Consumer)

use tokio::sync::mpsc;

async fn producer_consumer() {
    let (tx, mut rx) = mpsc::channel(100);

    // Spawn producers
    for i in 0..3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(format!("Message from {}", i)).await.unwrap();
        });
    }

    // Drop original sender so channel closes
    drop(tx);

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

Oneshot (Single-Shot Response)

use tokio::sync::oneshot;

async fn request_response() {
    let (tx, rx) = oneshot::channel();

    tokio::spawn(async move {
        let result = compute_something().await;
        tx.send(result).unwrap();
    });

    // Wait for response
    let response = rx.await.unwrap();
}

Broadcast (Multi-Consumer)

use tokio::sync::broadcast;

async fn pub_sub() {
    let (tx, _) = broadcast::channel(16);

    // Subscribe multiple consumers
    let mut rx1 = tx.subscribe();
    let mut rx2 = tx.subscribe();

    tokio::spawn(async move {
        while let Ok(msg) = rx1.recv().await {
            println!("Consumer 1: {}", msg);
        }
    });

    tokio::spawn(async move {
        while let Ok(msg) = rx2.recv().await {
            println!("Consumer 2: {}", msg);
        }
    });

    // Publish
    tx.send("Hello").unwrap();
}

Watch (Single Latest Value)

use tokio::sync::watch;

async fn config_updates() {
    let (tx, mut rx) = watch::channel(Config::default());

    // Consumer watches for changes
    tokio::spawn(async move {
        while rx.changed().await.is_ok() {
            let config = rx.borrow();
            apply_config(&config);
        }
    });

    // Update config
    tx.send(Config::new()).unwrap();
}

---

Structured Concurrency

JoinSet for Task Groups

use tokio::task::JoinSet;

async fn parallel_fetch(urls: Vec<String>) -> Vec<Result<Response, Error>> {
    let mut set = JoinSet::new();

    for url in urls {
        set.spawn(async move {
            fetch(&url).await
        });
    }

    let mut results = vec![];
    while let Some(res) = set.join_next().await {
        results.push(res.unwrap());
    }
    results
}

Scoped Tasks (no 'static)

// Using tokio-scoped or async-scoped crate
use async_scoped::TokioScope;

async fn scoped_example(data: &[u32]) {
    let results = TokioScope::scope_and_block(|scope| {
        for item in data {
            scope.spawn(async move {
                process(item).await
            });
        }
    });
}

---

Cancellation Patterns

Using CancellationToken

use tokio_util::sync::CancellationToken;

async fn cancellable_task(token: CancellationToken) {
    loop {
        select! {
            _ = token.cancelled() => {
                println!("Task cancelled");
                break;
            }
            _ = do_work() => {
                // Continue working
            }
        }
    }
}

async fn main_with_cancellation() {
    let token = CancellationToken::new();
    let task_token = token.clone();

    let handle = tokio::spawn(cancellable_task(task_token));

    // Cancel after some condition
    tokio::time::sleep(Duration::from_secs(5)).await;
    token.cancel();

    handle.await.unwrap();
}

Graceful Shutdown

async fn serve_with_shutdown(shutdown: impl Future) {
    let server = TcpListener::bind("0.0.0.0:8080").await.unwrap();

    loop {
        select! {
            Ok((socket, _)) = server.accept() => {
                tokio::spawn(handle_connection(socket));
            }
            _ = &mut shutdown => {
                println!("Shutting down...");
                break;
            }
        }
    }
}

#[tokio::main]
async fn main() {
    let ctrl_c = async {
        tokio::signal::ctrl_c().await.unwrap();
    };

    serve_with_shutdown(ctrl_c).await;
}

---

Backpressure Patterns

Bounded Channels

use tokio::sync::mpsc;

async fn with_backpressure() {
    // Buffer of 10 - producers will wait if full
    let (tx, mut rx) = mpsc::channel(10);

    let producer = tokio::spawn(async move {
        for i in 0..1000 {
            // This will wait if channel is full
            tx.send(i).await.unwrap();
        }
    });

    let consumer = tokio::spawn(async move {
        while let Some(item) = rx.recv().await {
            // Slow consumer
            tokio::time::sleep(Duration::from_millis(10)).await;
            process(item);
        }
    });

    let _ = tokio::join!(producer, consumer);
}

Semaphore for Rate Limiting

use tokio::sync::Semaphore;
use std::sync::Arc;

async fn rate_limited_requests(urls: Vec<String>) {
    let semaphore = Arc::new(Semaphore::new(10));  // max 10 concurrent

    let handles: Vec<_> = urls
        .into_iter()
        .map(|url| {
            let sem = Arc::clone(&semaphore);
            tokio::spawn(async move {
                let _permit = sem.acquire().await.unwrap();
                fetch(&url).await
            })
        })
        .collect();

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

---

Error Handling in Async

Propagating Errors

async fn fetch_and_parse(url: &str) -> Result<Data, Error> {
    let response = fetch(url).await?;
    let data = parse(response).await?;
    Ok(data)
}

Handling Task Panics

async fn robust_spawn() {
    let handle = tokio::spawn(async {
        risky_operation().await
    });

    match handle.await {
        Ok(result) => println!("Success: {:?}", result),
        Err(e) if e.is_panic() => {
            println!("Task panicked: {:?}", e);
        }
        Err(e) => {
            println!("Task cancelled: {:?}", e);
        }
    }
}

Try-Join for Multiple Results

use tokio::try_join;

async fn fetch_all() -> Result<(A, B, C), Error> {
    // All must succeed, or first error returned
    try_join!(
        fetch_a(),
        fetch_b(),
        fetch_c(),
    )
}

Common Concurrency Errors & Fixes

E0277: Cannot Send Between Threads

Error Pattern

use std::rc::Rc;

let data = Rc::new(42);
std::thread::spawn(move || {
    println!("{}", data);  // ERROR: Rc<i32> cannot be sent between threads
});

Fix Options

Option 1: Use Arc instead

use std::sync::Arc;

let data = Arc::new(42);
let data_clone = Arc::clone(&data);
std::thread::spawn(move || {
    println!("{}", data_clone);  // OK: Arc is Send
});

Option 2: Move owned data

let data = 42;  // i32 is Copy and Send
std::thread::spawn(move || {
    println!("{}", data);  // OK
});

---

E0277: Cannot Share Between Threads (Not Sync)

Error Pattern

use std::cell::RefCell;
use std::sync::Arc;

let data = Arc::new(RefCell::new(42));
// ERROR: RefCell is not Sync

Fix Options

Option 1: Use Mutex for thread-safe interior mutability

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

let data = Arc::new(Mutex::new(42));
let data_clone = Arc::clone(&data);
std::thread::spawn(move || {
    let mut guard = data_clone.lock().unwrap();
    *guard += 1;
});

Option 2: Use RwLock for read-heavy workloads

use std::sync::{Arc, RwLock};

let data = Arc::new(RwLock::new(42));
let data_clone = Arc::clone(&data);
std::thread::spawn(move || {
    let guard = data_clone.read().unwrap();
    println!("{}", *guard);
});

---

Deadlock Patterns

Pattern 1: Lock Ordering Deadlock

// DANGER: potential deadlock
use std::sync::{Arc, Mutex};

let a = Arc::new(Mutex::new(1));
let b = Arc::new(Mutex::new(2));

// Thread 1: locks a then b
let a1 = Arc::clone(&a);
let b1 = Arc::clone(&b);
std::thread::spawn(move || {
    let _a = a1.lock().unwrap();
    let _b = b1.lock().unwrap();  // waits for b
});

// Thread 2: locks b then a (opposite order!)
let a2 = Arc::clone(&a);
let b2 = Arc::clone(&b);
std::thread::spawn(move || {
    let _b = b2.lock().unwrap();
    let _a = a2.lock().unwrap();  // waits for a - DEADLOCK
});

Fix: Consistent Lock Ordering

// SAFE: always lock in same order (a before b)
std::thread::spawn(move || {
    let _a = a1.lock().unwrap();
    let _b = b1.lock().unwrap();
});

std::thread::spawn(move || {
    let _a = a2.lock().unwrap();  // same order
    let _b = b2.lock().unwrap();
});

Pattern 2: Self-Deadlock

// DANGER: locking same mutex twice
let m = Mutex::new(42);
let _g1 = m.lock().unwrap();
let _g2 = m.lock().unwrap();  // DEADLOCK on std::Mutex

// FIX: use parking_lot::ReentrantMutex if needed
// or restructure code to avoid double locking

---

Mutex Guard Across Await

Error Pattern

use std::sync::Mutex;
use tokio::time::sleep;

async fn bad_async() {
    let m = Mutex::new(42);
    let guard = m.lock().unwrap();
    sleep(Duration::from_secs(1)).await;  // WARNING: guard held across await
    println!("{}", *guard);
}

Fix Options

Option 1: Scope the lock

async fn good_async() {
    let m = Mutex::new(42);
    let value = {
        let guard = m.lock().unwrap();
        *guard  // copy value
    };  // guard dropped here
    sleep(Duration::from_secs(1)).await;
    println!("{}", value);
}

Option 2: Use tokio::sync::Mutex

use tokio::sync::Mutex;

async fn good_async() {
    let m = Mutex::new(42);
    let guard = m.lock().await;  // async lock
    sleep(Duration::from_secs(1)).await;  // OK with tokio::Mutex
    println!("{}", *guard);
}

---

Data Race Prevention

Pattern: Missing Synchronization

// This WON'T compile - Rust prevents data races
use std::sync::Arc;

let data = Arc::new(0);
let d1 = Arc::clone(&data);
let d2 = Arc::clone(&data);

std::thread::spawn(move || {
    // *d1 += 1;  // ERROR: cannot mutate through Arc
});

std::thread::spawn(move || {
    // *d2 += 1;  // ERROR: cannot mutate through Arc
});

Fix: Add Synchronization

use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicI32, Ordering};

// Option 1: Mutex
let data = Arc::new(Mutex::new(0));
let d1 = Arc::clone(&data);
std::thread::spawn(move || {
    *d1.lock().unwrap() += 1;
});

// Option 2: Atomic (for simple types)
let data = Arc::new(AtomicI32::new(0));
let d1 = Arc::clone(&data);
std::thread::spawn(move || {
    d1.fetch_add(1, Ordering::SeqCst);
});

---

Channel Errors

Disconnected Channel

use std::sync::mpsc;

let (tx, rx) = mpsc::channel();
drop(tx);  // sender dropped
match rx.recv() {
    Ok(v) => println!("{}", v),
    Err(_) => println!("channel disconnected"),  // this happens
}

Fix: Handle Disconnection

// Use try_recv for non-blocking
loop {
    match rx.try_recv() {
        Ok(msg) => handle(msg),
        Err(TryRecvError::Empty) => continue,
        Err(TryRecvError::Disconnected) => break,
    }
}

// Or iterate (stops on disconnect)
for msg in rx {
    handle(msg);
}

---

Async Common Errors

Forgetting to Spawn

// WRONG: future not polled
async fn fetch_data() -> Result<Data, Error> { ... }

fn process() {
    fetch_data();  // does nothing! returns Future that's dropped
}

// RIGHT: await or spawn
async fn process() {
    let data = fetch_data().await;  // awaited
}

fn process_sync() {
    tokio::spawn(fetch_data());  // spawned
}

Blocking in Async Context

// WRONG: blocks the executor
async fn bad() {
    std::thread::sleep(Duration::from_secs(1));  // blocks!
    std::fs::read_to_string("file.txt").unwrap();  // blocks!
}

// RIGHT: use async versions
async fn good() {
    tokio::time::sleep(Duration::from_secs(1)).await;
    tokio::fs::read_to_string("file.txt").await.unwrap();
}

// Or spawn_blocking for CPU-bound work
async fn compute() {
    let result = tokio::task::spawn_blocking(|| {
        heavy_computation()  // OK to block here
    }).await.unwrap();
}

---

Thread Panic Handling

Unhandled Panic

let handle = std::thread::spawn(|| {
    panic!("oops");
});

// Main thread continues, might miss the error
handle.join().unwrap();  // panics here

Proper Error Handling

let handle = std::thread::spawn(|| {
    panic!("oops");
});

match handle.join() {
    Ok(result) => println!("Success: {:?}", result),
    Err(e) => println!("Thread panicked: {:?}", e),
}

// For async: use catch_unwind
use std::panic;

async fn safe_task() {
    let result = panic::catch_unwind(|| {
        risky_operation()
    });

    match result {
        Ok(v) => use_value(v),
        Err(_) => log_error("task panicked"),
    }
}