19 Januari 2026 22 min read Software Engineering

Concurrency and Race Conditions in Rust: A Complete Guide from Ownership to Async Tokio

Rust takes a fundamentally different approach from all other languages when handling concurrency. In Go, race conditions are prevented by conventions and a race detector that runs at runtime. In Java and Kotlin, developers are responsible for using synchronized, volatile, or locks correctly — and if they forget, bugs appear in production. In Rust, race conditions are impossible — not because of a runtime monitoring them, not because of conventions that must be followed, but because the compiler refuses to compile code that contains a data race.

This claim sounds too good to be true, but this is what Rust calls “fearless concurrency”. Rust’s ownership and borrowing system, which already guarantees memory safety without a garbage collector, also naturally guarantees thread safety. The Send and Sync traits let the compiler statically verify whether a type is safe to move across threads or access from multiple threads simultaneously. If your code violates these guarantees, the program won’t compile — race conditions are detected at compile time, not runtime.

This doesn’t mean concurrency in Rust is easy. Precisely because the compiler is very strict, the learning curve is quite steep — especially the first time you face error messages about lifetimes, the borrow checker, and trait bounds. But when the code finally compiles, you can be confident there’s no data race in it. This article covers the entire spectrum of concurrency in Rust: from primitive threading and synchronization primitives, to async/await and Tokio (current stable version 1.52.x) which has become the de-facto standard for async I/O in the Rust ecosystem.

Ownership and Borrowing: The Foundation of Thread Safety in Rust

To understand why Rust can guarantee thread safety, we first need to understand Rust’s two core rules.

Rule 1 — Ownership: Every value in Rust is owned by exactly one variable (the owner) at a time. When the owner goes out of scope, the value is dropped (memory freed). A value can be moved to a new owner, but the old owner can no longer be accessed.

Rule 2 — Borrowing: You can borrow references to a value without taking ownership. At any one time, a value may only have one of the following:

One or more immutable references (&T) — many may read simultaneously
Exactly one mutable reference (&mut T) — only one may write, and no one can read simultaneously

This borrowing rule feels very familiar from a concurrency perspective: “many readers or one writer” is the same principle as RwLock. The difference is that in Rust, this is guaranteed by the compiler statically without runtime overhead.

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

    // OK: many immutable references at once
    let r1 = &data;
    let r2 = &data;
    println!("{:?} {:?}", r1, r2);

    // ERROR at compile time: can't have a mutable and immutable reference together
    let r3 = &mut data;
    // error[E0502]: cannot borrow `data` as mutable because it is also borrowed as immutable
    println!("{:?}", r1); // r1 is still active here
}

If the same code ran from two threads without synchronization, this would be exactly the race condition that occurs — one thread reading while another thread writes. Rust prevents this not with a runtime check, but by refusing to compile code with that potential.

Basic Threads: std::thread

Rust provides threading primitives in the standard library via std::thread:

use std::thread;
use std::time::Duration;

fn main() {
    // Spawn a new thread — returns a JoinHandle
    let handle = thread::spawn(|| {
        for i in 1..=5 {
            println!("Child thread: {}", i);
            thread::sleep(Duration::from_millis(10));
        }
    });

    // The main thread also runs concurrently
    for i in 1..=3 {
        println!("Main thread: {}", i);
        thread::sleep(Duration::from_millis(10));
    }

    handle.join().unwrap(); // wait for the child thread to finish
}

Move Closures: Moving Ownership to a Thread

When a thread accesses data from outside its closure, Rust forces you to move ownership of that data into the thread — not borrow a reference to it. This prevents situations where the thread still uses data while its original owner is already dead (dangling pointers).

use std::thread;

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

    // ANTI-PATTERN: the thread might outlive `data`
    // let handle = thread::spawn(|| {
    //     println!("{:?}", data);
    //     // error[E0373]: closure may outlive the current function, but it borrows `data`
    // });

    // CORRECT: use `move` to transfer ownership to the thread
    let handle = thread::spawn(move || {
        println!("{:?}", data); // `data` is now owned by this thread
    });

    // println!("{:?}", data); // ERROR: data was already moved to the thread

    handle.join().unwrap();
}

Thread Configuration

use std::thread;

let handle = thread::Builder::new()
    .name("worker-1".to_string())
    .stack_size(4 * 1024 * 1024) // 4 MB stack
    .spawn(|| {
        println!("Thread name: {}", thread::current().name().unwrap());
    })
    .expect("Failed to create thread");

handle.join().unwrap();

Send and Sync: The Thread Safety Guardian Traits

The two most important traits for concurrency in Rust are Send and Sync — both are marker traits with no methods, just marking a type’s properties:

Send: The type is safe to move to another thread. Almost all types in Rust are Send by default.
Sync: The type is safe to access from multiple threads via a shared reference. T is Sync if and only if &T is Send.

flowchart TD
    A[Type T] --> B{Implements Send?}
    B -- Yes --> C[Safe to move to another thread]
    B -- No --> D[Only on its creating thread<br/>example: Rc T, raw pointers]
    A --> E{Implements Sync?}
    E -- Yes --> F[Safe to access via shared ref<br/>from many threads]
    E -- No --> G[Needs a wrapper like<br/>Mutex or RwLock]

Why `Rc<T>` Isn’t Send

Rc<T> (Reference Counted) uses a non-atomic counter because it assumes no concurrency — faster than Arc<T>, but not thread-safe:

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

fn main() {
    let rc = Rc::new(42);

    thread::spawn(move || {
        println!("{}", rc);
        // error[E0277]: `Rc<i32>` cannot be sent between threads safely
        // the trait `Send` is not implemented for `Rc<i32>`
    });
}

The compiler immediately rejects this — no runtime crash, no undefined behavior. Just a clear error message.

Arc: Shared Ownership Across Threads

Arc<T> (Atomically Reference Counted) is the thread-safe version of Rc<T> that uses atomic operations to manage the reference count:

use std::sync::Arc;
use std::thread;

fn main() {
    let data = Arc::new(vec![1, 2, 3]);
    let mut handles = vec![];

    for i in 0..3 {
        let data_clone = Arc::clone(&data); // clone the pointer, not the data
        let handle = thread::spawn(move || {
            println!("Thread {}: {:?}", i, data_clone);
        });
        handles.push(handle);
    }

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

Arc<T> alone only provides shared immutable access. To write from multiple threads, Arc is combined with Mutex or RwLock.

Mutex and RwLock

Mutex<T>: RAII-Based Locking

Mutex<T> in Rust stores the data inside the Mutex. To access the data, you must acquire the lock — the result is a MutexGuard that provides access to the data. When the MutexGuard goes out of scope, the lock is automatically released via RAII. There’s no risk of forgetting to call unlock().

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

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

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            for _ in 0..1000 {
                let mut guard = counter.lock().unwrap(); // acquire the lock
                *guard += 1;
                // guard goes out of scope: lock automatically released (RAII)
            }
        });
        handles.push(handle);
    }

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

    println!("Counter: {}", *counter.lock().unwrap()); // always 10000
}

The Arc<Mutex<T>> pattern is the most common idiom in Rust for shared mutable state across threads.

Mutex Poisoning

If a thread holding the lock panics, the Mutex becomes poisoned to prevent other threads from accessing data that may be in an inconsistent state:

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

fn main() {
    let mutex = Arc::new(Mutex::new(0_i32));
    let m = Arc::clone(&mutex);

    let _ = thread::spawn(move || {
        let _guard = m.lock().unwrap();
        panic!("panic while holding the lock!"); // the mutex becomes poisoned
    }).join();

    match mutex.lock() {
        Ok(val) => println!("Normal: {}", *val),
        Err(poisoned) => {
            // you can force access with into_inner()
            println!("Poisoned, but data: {}", *poisoned.into_inner());
        }
    }
}

RwLock<T>: Multiple Readers or a Single Writer

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

fn main() {
    let cache = Arc::new(RwLock::new(HashMap::<String, String>::new()));

    // Writer
    {
        let mut map = cache.write().unwrap(); // exclusive
        map.insert("key".to_string(), "value".to_string());
    }

    // Many concurrent readers
    let mut handles = vec![];
    for i in 0..5 {
        let cache = Arc::clone(&cache);
        let h = thread::spawn(move || {
            let map = cache.read().unwrap(); // many readers OK simultaneously
            println!("Reader {}: {:?}", i, map.get("key"));
        });
        handles.push(h);
    }

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

	`Mutex<T>`	`RwLock<T>`
Concurrent reads	No	Yes
Write	One, exclusive	One, exclusive
Overhead	Low	Slightly higher
Good for	Write-heavy / mixed	Read-heavy
Poisoning	Yes	Yes

Channels: Communication Between Threads

Rust’s standard library provides std::sync::mpsc (multiple producer, single consumer):

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

fn main() {
    let (tx, rx) = mpsc::channel::<String>();
    let tx1 = tx.clone();
    let tx2 = tx.clone();

    thread::spawn(move || {
        tx1.send("message from thread 1".to_string()).unwrap();
        tx1.send("another message from thread 1".to_string()).unwrap();
    });

    thread::spawn(move || {
        tx2.send("message from thread 2".to_string()).unwrap();
    });

    drop(tx); // drop the original so rx knows all senders are done

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

Bounded Channels with sync_channel

use std::sync::mpsc;

// Channel with capacity 10: senders block when the buffer is full
let (tx, rx) = mpsc::sync_channel::<i32>(10);

thread::spawn(move || {
    for i in 0..20 {
        tx.send(i).unwrap(); // blocks when the buffer is full — natural backpressure
    }
});

for val in rx {
    println!("{}", val);
}

Atomic Types: Lock-Free Operations

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

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

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

    for handle in handles { handle.join().unwrap(); }
    println!("Counter: {}", counter.load(Ordering::SeqCst)); // always 10000
}

Memory Ordering

use std::sync::atomic::Ordering;

// Relaxed: only atomicity, no relative ordering guarantees
// Good for pure counters not used for synchronization
counter.fetch_add(1, Ordering::Relaxed);

// Acquire/Release: establishes a happens-before relationship
// Release: all prior operations can't be reordered after the store
// Acquire: all subsequent operations can't be reordered before the load
flag.store(true, Ordering::Release);       // sender: publish data, then set the flag
if flag.load(Ordering::Acquire) { ... }   // receiver: read the flag, then read the data

// SeqCst: the strongest guarantee, a global total order of all atomic operations
// Safest, easiest to understand, slightly slower
counter.load(Ordering::SeqCst);

Choosing the wrong Ordering doesn’t cause a compile error, but can cause subtle bugs on CPUs with weak memory models like ARM and PowerPC. For non-performance-critical code, use SeqCst as the safe default. Optimize to Relaxed or Acquire/Release only if you truly understand the memory ordering model and have profiled that this is a bottleneck.

Deadlocks in Rust: Still Possible

Rust prevents data races, but doesn’t prevent deadlocks. A deadlock is a logic problem, not a memory safety problem, so it’s outside the scope of what the compiler can statically verify.

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

fn main() {
    let lock_a = Arc::new(Mutex::new(0));
    let lock_b = Arc::new(Mutex::new(0));
    let (la1, lb1) = (Arc::clone(&lock_a), Arc::clone(&lock_b));
    let (la2, lb2) = (Arc::clone(&lock_a), Arc::clone(&lock_b));

    // ANTI-PATTERN: different lock ordering in two threads — DEADLOCK
    let t1 = thread::spawn(move || {
        let _a = la1.lock().unwrap();   // take A
        thread::sleep(Duration::from_millis(10));
        let _b = lb1.lock().unwrap();   // wait for B — DEADLOCK
    });

    let t2 = thread::spawn(move || {
        let _b = lb2.lock().unwrap();   // take B
        thread::sleep(Duration::from_millis(10));
        let _a = la2.lock().unwrap();   // wait for A — DEADLOCK
    });

    t1.join().unwrap(); // hangs forever
    t2.join().unwrap();
}

flowchart LR
    T1[Thread 1<br/>holds lock_a] -->|waiting for| LB[lock_b]
    T2[Thread 2<br/>holds lock_b] -->|waiting for| LA[lock_a]
    LA -->|held by| T1
    LB -->|held by| T2

// CORRECT: always acquire locks in the same order across all threads
let t1 = thread::spawn(move || {
    let _a = la1.lock().unwrap(); // A first
    let _b = lb1.lock().unwrap(); // B second
});

let t2 = thread::spawn(move || {
    let _a = la2.lock().unwrap(); // A first (consistent order!)
    let _b = lb2.lock().unwrap(); // B second
});

Async/Await in Rust: The Future-Based Model

OS-based threading works well for CPU-bound work, but for I/O-bound concurrency with thousands of concurrent connections, OS thread overhead becomes the bottleneck. The solution is async/await with an asynchronous runtime.

Future: The Basic Unit of Async Rust

Future is a trait representing a computation that isn’t finished yet. Unlike a Promise in JavaScript or a Coroutine in Kotlin, which start running when created, Future in Rust is lazy — it does nothing until polled by the runtime:

// async fn is syntactic sugar for a function returning impl Future
async fn fetch_data(url: &str) -> String {
    // `await` polls the Future until it completes
    // while waiting, the task is suspended, the thread can run other tasks
    reqwest::get(url).await.unwrap().text().await.unwrap()
}

// Equivalent without the sugar syntax:
fn fetch_data_manual(url: String) -> impl std::future::Future<Output = String> {
    async move {
        reqwest::get(&url).await.unwrap().text().await.unwrap()
    }
}

Tokio: The De-facto Async Runtime

Rust provides the Future trait and async/await syntax, but doesn’t provide a built-in runtime. The runtime is responsible for efficiently polling futures and managing the thread pool. Tokio is the most popular runtime in the Rust ecosystem:

# Cargo.toml
[dependencies]
tokio = { version = "1.52", features = ["full"] }

// #[tokio::main] sets up the Tokio runtime and runs an async main
#[tokio::main]
async fn main() {
    let result = fetch_data_local().await;
    println!("Result: {}", result);
}

async fn fetch_data_local() -> String {
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
    "data complete".to_string()
}

Single-Thread vs Multi-Thread Runtime

// Multi-thread (default): work-stealing scheduler, threads = number of CPU cores
#[tokio::main]
async fn main() { /* ... */ }

// Explicit multi-thread with 4 workers
#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() { /* ... */ }

// Single-thread: all tasks on one thread, no parallelism
#[tokio::main(flavor = "current_thread")]
async fn main() { /* ... */ }

Tokio Tasks: Rust’s Goroutines

tokio::task::spawn launches an async task that runs concurrently on the Tokio runtime:

use tokio::task;

#[tokio::main]
async fn main() {
    // Spawn a task — runs concurrently
    let handle = task::spawn(async {
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
        42_i32
    });

    let result = handle.await.unwrap();
    println!("Result: {}", result);

    // Spawn many parallel tasks and collect the results
    let handles: Vec<_> = (0..10)
        .map(|i| task::spawn(async move {
            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
            i * 2
        }))
        .collect();

    for handle in handles {
        println!("Result: {}", handle.await.unwrap());
    }
}

spawn_blocking: Blocking Code in an Async Context

Blocking code must not run directly in an async task — it will block the worker thread and prevent other tasks from running:

use tokio::task;

#[tokio::main]
async fn main() {
    // ANTI-PATTERN: blocking calls in an async task
    // std::thread::sleep(Duration::from_secs(1)); // freezes the worker thread!
    // std::fs::read_to_string("file.txt");        // blocking I/O!

    // CORRECT: move it to the blocking thread pool
    let result = task::spawn_blocking(|| {
        std::thread::sleep(std::time::Duration::from_millis(100));
        heavy_computation()
    }).await.unwrap();

    println!("Result: {}", result);

    // CORRECT: use tokio::fs for async file I/O
    let content = tokio::fs::read_to_string("data.txt").await;
}

fn heavy_computation() -> i64 {
    (1..=1_000_000_i64).sum()
}

flowchart TD
    subgraph "Tokio Runtime"
        subgraph "Async Worker Threads"
            W1[Worker 1]
            W2[Worker 2]
        end
        subgraph "Blocking Thread Pool"
            B1[Blocking Thread 1]
            B2[Blocking Thread 2]
        end
    end
    AT1[Async Task 1] --> W1
    AT2[Async Task 2] --> W2
    BT1[Blocking Task via spawn_blocking] --> B1
    W1 -.suspends at await.- W1

Race Conditions in Async Rust

Even though Rust prevents data races at the memory level, logical race conditions — especially check-then-act patterns — can still happen in async code:

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

// ANTI-PATTERN: check-then-act is not atomic
async fn get_or_insert(
    cache: Arc<Mutex<HashMap<String, String>>>,
    key: String,
) -> String {
    {
        let map = cache.lock().await;
        if let Some(val) = map.get(&key) {
            return val.clone();
        }
    } // lock released here

    // GAP: another task could insert the same key here!
    let new_val = fetch_expensive(&key).await;

    let mut map = cache.lock().await;
    map.insert(key, new_val.clone()); // may overwrite another task's result
    new_val
}

// CORRECT: use the atomic entry API, or hold the lock across the whole operation
async fn get_or_insert_safe(
    cache: Arc<Mutex<HashMap<String, String>>>,
    key: String,
) -> String {
    let mut map = cache.lock().await;
    if let Some(val) = map.get(&key) {
        return val.clone(); // already exists, return directly
    }
    // Lock still held — no other task can get in
    let new_val = format!("value-for-{}", key); // sync operation inside the lock
    map.insert(key, new_val.clone());
    new_val
}

async fn fetch_expensive(key: &str) -> String {
    tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
    format!("value-for-{}", key)
}

Tokio Sync Primitives

tokio::sync provides async-aware versions of all synchronization primitives. Important: don’t use std::sync::Mutex while holding the lock across an await point — this can cause deadlocks because the thread could switch to running another task while the lock is still held:

use tokio::sync::{Mutex, RwLock, Semaphore, Notify};
use std::sync::Arc;

// Async Mutex: safe to hold across await
let mutex = Arc::new(Mutex::new(0_i32));
{
    let mut guard = mutex.lock().await;
    *guard += 1;
    some_async_fn().await; // safe: tokio Mutex suspends the coroutine, doesn't block the thread
} // guard dropped: lock released

// Async RwLock
let rwlock = Arc::new(RwLock::new(vec![1, 2, 3]));
let read = rwlock.read().await;   // many concurrent readers
let write = rwlock.write().await; // exclusive writer

// Semaphore: limit concurrent operations
let semaphore = Arc::new(Semaphore::new(10));
let permit = semaphore.acquire().await.unwrap();
do_limited_work().await;
drop(permit); // return the slot

// Notify: signal between tasks
let notify = Arc::new(Notify::new());
let n = Arc::clone(&notify);
tokio::spawn(async move { n.notified().await; println!("Notified!"); });
notify.notify_one();

async fn some_async_fn() {}
async fn do_limited_work() {}

When to Use `std::sync::Mutex` vs `tokio::sync::Mutex`

// Use std::sync::Mutex if:
// the lock is held very briefly, with no await inside
use std::sync::Mutex;
let fast = Mutex::new(0_i32);
{ *fast.lock().unwrap() += 1; } // no await — safe and faster

// Use tokio::sync::Mutex if:
// you need to await while the lock is held
use tokio::sync::Mutex as AsyncMutex;
let slow = Arc::new(AsyncMutex::new(vec![]));
let mut g = slow.lock().await;
let data = fetch_async().await; // await inside the lock — must use a tokio Mutex
g.push(data);

async fn fetch_async() -> String { "data".to_string() }

Async Channels in Tokio

Tokio provides four types of channels for communication between async tasks:

use tokio::sync::{mpsc, oneshot, broadcast, watch};

// ── mpsc: Multiple Producer, Single Consumer ──────────────────────────────
let (tx, mut rx) = mpsc::channel::<String>(100); // bounded

for i in 0..5 {
    let tx = tx.clone();
    tokio::spawn(async move {
        tx.send(format!("Message {}", i)).await.unwrap();
    });
}
drop(tx);
while let Some(msg) = rx.recv().await { println!("{}", msg); }

// ── oneshot: One message, once ────────────────────────────────────────────
let (tx, rx) = oneshot::channel::<String>();
tokio::spawn(async move {
    tx.send("result".to_string()).unwrap();
});
let result = rx.await.unwrap();

// ── broadcast: One to many ────────────────────────────────────────────────
let (tx, _) = broadcast::channel::<String>(16);
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();
tx.send("event".to_string()).unwrap();
// rx1 and rx2 each receive "event"

// ── watch: State propagation ──────────────────────────────────────────────
let (tx, rx) = watch::channel("initial");
let mut rx1 = rx.clone();
tokio::spawn(async move {
    rx1.changed().await.unwrap();
    println!("New state: {}", *rx1.borrow());
});
tx.send("updated").unwrap();

Channel	Producer	Consumer	Use Case
`mpsc`	Many	One	Task workers, pipelines
`oneshot`	One	One	Request-response
`broadcast`	One	Many	Event broadcasting, pub-sub
`watch`	One	Many	State propagation, config

select!: Waiting on Several Futures at Once

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let (tx1, mut rx1) = mpsc::channel::<&str>(1);
    let (tx2, mut rx2) = mpsc::channel::<&str>(1);

    tokio::spawn(async move {
        sleep(Duration::from_millis(200)).await;
        tx1.send("from channel 1").await.unwrap();
    });
    tokio::spawn(async move {
        sleep(Duration::from_millis(100)).await;
        tx2.send("from channel 2").await.unwrap(); // this one first
    });

    loop {
        tokio::select! {
            Some(msg) = rx1.recv() => { println!("rx1: {}", msg); break; }
            Some(msg) = rx2.recv() => { println!("rx2: {}", msg); break; }
            _ = sleep(Duration::from_secs(5)) => { println!("Timeout!"); break; }
        }
    }
}

select! is very useful for timeouts, cancellation via done-channels, and racing several alternative data sources.

Cancellation in Tokio

Tokio uses drop-based cancellation: when a JoinHandle is dropped or abort() is called, the task is cancelled at the next await point:

#[tokio::main]
async fn main() {
    let handle = tokio::spawn(async {
        tokio::time::sleep(tokio::time::Duration::from_secs(10)).await; // cancelled here
        println!("will not print");
    });

    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
    handle.abort();

    match handle.await {
        Ok(_) => println!("Finished normally"),
        Err(e) if e.is_cancelled() => println!("Cancelled"),
        Err(e) => println!("Panic: {}", e),
    }
}

Cancellation Safety

Because cancellation happens at every await point, code that isn’t cancellation-safe can leave state half-finished:

// ANTI-PATTERN: state could be left half-finished if cancelled
async fn update(state: &mut Vec<String>, key: String) {
    let val = fetch(&key).await; // cancelled HERE: the insert never happens
    state.push(val);
}

// CORRECT: fetch first, then modify state (no await after the modification)
async fn update_safe(state: &mut Vec<String>, key: String) {
    let val = fetch(&key).await; // cancel here is OK: state hasn't changed yet
    state.push(val);             // no await: can't be cancelled here
}

async fn fetch(_key: &str) -> String { "value".to_string() }

Production Concurrency Patterns

Worker Pools with a Semaphore

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

#[tokio::main]
async fn main() {
    let semaphore = Arc::new(Semaphore::new(5)); // max 5 concurrent
    let mut handles = vec![];

    for i in 0..20 {
        let sem = Arc::clone(&semaphore);
        handles.push(tokio::spawn(async move {
            let _permit = sem.acquire().await.unwrap(); // wait for a slot
            println!("Task {} started", i);
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
            println!("Task {} finished", i);
            // permit dropped: slot available for the next task
        }));
    }

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

Async Pipelines

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx1, rx1) = mpsc::channel(100);
    let (tx2, rx2) = mpsc::channel(100);

    tokio::join!(
        produce(tx1, vec!["a", "b", "c"]),
        transform(rx1, tx2),
        consume(rx2),
    );
}

async fn produce(tx: mpsc::Sender<String>, items: Vec<&str>) {
    for item in items { tx.send(item.to_string()).await.unwrap(); }
}

async fn transform(mut rx: mpsc::Receiver<String>, tx: mpsc::Sender<String>) {
    while let Some(item) = rx.recv().await {
        tx.send(item.to_uppercase()).await.unwrap();
    }
}

async fn consume(mut rx: mpsc::Receiver<String>) {
    while let Some(item) = rx.recv().await { println!("Output: {}", item); }
}

Graceful Shutdown

use tokio::sync::broadcast;
use tokio::signal;

#[tokio::main]
async fn main() {
    let (shutdown_tx, _) = broadcast::channel::<()>(1);

    for i in 0..3 {
        let mut rx = shutdown_tx.subscribe();
        tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = rx.recv() => { println!("Worker {} shutting down", i); break; }
                    _ = tokio::time::sleep(tokio::time::Duration::from_millis(500)) => {
                        println!("Worker {} working...", i);
                    }
                }
            }
        });
    }

    signal::ctrl_c().await.unwrap();
    println!("Ctrl+C — graceful shutdown...");
    shutdown_tx.send(()).unwrap();
    tokio::time::sleep(tokio::time::Duration::from_millis(600)).await;
}

Common Anti-Patterns

Anti-Pattern 1: Blocking in an Async Context

// ANTI-PATTERN
async fn bad() {
    std::thread::sleep(std::time::Duration::from_secs(1)); // freezes the worker thread!
    let _ = std::fs::read_to_string("file.txt");           // blocking I/O!
}

// CORRECT
async fn good() {
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
    let _ = tokio::fs::read_to_string("file.txt").await;
    // or for CPU-heavy work:
    tokio::task::spawn_blocking(|| heavy()).await.unwrap();
}

fn heavy() -> i64 { (1..=1_000_000_i64).sum() }

Anti-Pattern 2: std Mutex Across an Await

use std::sync::Mutex;

// ANTI-PATTERN: std Mutex held across an await — potential deadlock
async fn bad(m: Arc<Mutex<Vec<String>>>) {
    let mut g = m.lock().unwrap();
    let data = fetch().await; // await while the lock is held — DANGER
    g.push(data);
}

// CORRECT option A: fetch first, then lock
async fn good_a(m: Arc<Mutex<Vec<String>>>) {
    let data = fetch().await;       // fetch without the lock
    m.lock().unwrap().push(data);   // brief lock, no await
}

// CORRECT option B: use tokio::sync::Mutex
async fn good_b(m: Arc<tokio::sync::Mutex<Vec<String>>>) {
    let mut g = m.lock().await;
    let data = fetch().await; // safe with a tokio Mutex
    g.push(data);
}

async fn fetch() -> String { "data".to_string() }

Anti-Pattern 3: Unlimited Tasks

// ANTI-PATTERN: unbounded spawning — OOM during load spikes
async fn bad(items: Vec<String>) {
    let handles: Vec<_> = items.into_iter()
        .map(|item| tokio::spawn(async move { process(item).await }))
        .collect(); // 100,000 items = 100,000 tasks at once!
    for h in handles { h.await.unwrap(); }
}

// CORRECT: limit with a Semaphore
async fn good(items: Vec<String>) {
    let sem = Arc::new(Semaphore::new(50));
    let mut set = tokio::task::JoinSet::new();
    for item in items {
        let sem = Arc::clone(&sem);
        set.spawn(async move {
            let _p = sem.acquire().await.unwrap();
            process(item).await;
        });
    }
    while let Some(r) = set.join_next().await { r.unwrap(); }
}

async fn process(_item: String) {}

Anti-Pattern 4: Swallowing Cancellation Errors

// ANTI-PATTERN: catching all errors including task cancellation
async fn bad_wrapper() {
    match tokio::spawn(async { some_work().await }).await {
        Ok(_) => {}
        Err(_) => {} // swallows JoinError — you can't tell panic from cancel!
    }
}

// CORRECT: check the error type
async fn good_wrapper() {
    match tokio::spawn(async { some_work().await }).await {
        Ok(_) => println!("Finished"),
        Err(e) if e.is_cancelled() => println!("Cancelled"),
        Err(e) => std::panic::resume_unwind(e.into_panic()), // re-panic
    }
}

async fn some_work() {}

Unsafe and Raw Concurrency

Everything discussed above is safe Rust. Rust also provides unsafe blocks for code needing the lowest-level control, where the programmer takes over full responsibility for safety:

use std::sync::atomic::{AtomicPtr, Ordering};

// Example of a simple lock-free stack using raw pointers
struct Node<T> {
    data: T,
    next: *mut Node<T>,
}

struct LockFreeStack<T> {
    head: AtomicPtr<Node<T>>,
}

impl<T> LockFreeStack<T> {
    fn new() -> Self {
        Self { head: AtomicPtr::new(std::ptr::null_mut()) }
    }

    fn push(&self, data: T) {
        let node = Box::into_raw(Box::new(Node {
            data,
            next: std::ptr::null_mut(),
        }));

        loop {
            let head = self.head.load(Ordering::Acquire);
            unsafe { (*node).next = head; }
            if self.head.compare_exchange(
                head, node,
                Ordering::Release,
                Ordering::Relaxed,
            ).is_ok() { break; }
        }
    }
}

unsafe is the escape hatch from all of Rust’s safety guarantees. Code inside it can cause data races, dangling pointers, and undefined behavior — exactly like C. Use it only when truly necessary, always document why the code is safe, and keep the unsafe area as small as possible.

Concurrent Rust Code Review Checklist

OWNERSHIP AND SEND/SYNC:
  □ Is data shared across threads wrapped in Arc?
  □ Do types sent to threads implement Send?
  □ Do types shared via references implement Sync?
  □ Are move closures used when a thread needs access to external data?

MUTEX AND RWLOCK:
  □ Is std::sync::Mutex never held across an await? (use tokio::sync::Mutex)
  □ Is the lock scope as narrow as possible — no heavy operations inside?
  □ Is lock acquisition ordering consistent across the codebase (prevent deadlocks)?
  □ Is Mutex poisoning handled correctly?
  □ Is RwLock used for read-heavy workloads?

ATOMICS:
  □ Is the Ordering chosen appropriately (SeqCst for safety, Relaxed/Acq-Rel for performance)?
  □ Do compound operations use CAS, not two separate operations?

ASYNC AND TOKIO:
  □ Are there no std::thread::sleep or blocking I/O calls in async tasks?
  □ Is blocking/CPU-heavy code run via spawn_blocking?
  □ Is the number of concurrent tasks limited (Semaphore, JoinSet)?
  □ Is the JoinError type from tasks checked (cancelled vs panic)?
  □ Are select! branches cancellation-safe?

CHANNELS:
  □ Is the right channel type used (mpsc/oneshot/broadcast/watch)?
  □ Are bounded channels used for natural backpressure?
  □ Are senders dropped when unused so receivers know the channel is done?

UNSAFE:
  □ Is unsafe truly necessary with no safe alternative?
  □ Are all safety invariants maintained manually?
  □ Is there a comment explaining why this unsafe code is genuinely safe?

Summary

Rust prevents data races at compile time through its ownership and borrowing system — if it compiles, there’s no race condition at the memory level.
The Send trait (safe to move to another thread) and Sync trait (safe to access via shared reference from many threads) are verified statically by the compiler.
Arc<T> for shared ownership across threads; Rc<T> for single-thread only — Rc isn’t Send, and the compiler rejects sending it to another thread.
Arc<Mutex<T>> is the most common idiom for shared mutable state — the lock is automatically released via RAII when the MutexGuard goes out of scope.
RwLock<T> for read-heavy workloads: many concurrent readers, one exclusive writer.
Atomic types for lock-free operations on primitives — choose the right Ordering; SeqCst for a safe default, Relaxed/Acquire/Release for optimization.
Rust prevents data races, but doesn’t prevent deadlocks — consistent lock acquisition ordering remains the programmer’s responsibility.
Future in Rust is lazy — it doesn’t run until polled by the runtime. Unlike a Promise or Coroutine, which start running when created.
Tokio (v1.52.x) is the de-facto async runtime — providing a multi-thread work-stealing scheduler, async I/O, and a complete set of sync primitives.
tokio::task::spawn for concurrent async tasks; spawn_blocking for blocking/CPU-heavy code so it doesn’t freeze the worker thread.
Don’t use std::sync::Mutex across an await point — use tokio::sync::Mutex, which suspends the task without blocking the thread.
The four Tokio channels: mpsc (many-to-one), oneshot (one message), broadcast (one-to-many), watch (current state always available).
tokio::select! for waiting on multiple futures/channels — useful for timeouts, cancellation, and racing alternatives.
Cancellation in Tokio is drop-based, happening at every await point — make code cancellation-safe so there’s no half-finished state.
unsafe is the escape hatch from all of Rust’s guarantees — use it only when absolutely necessary, document the invariants, and keep its scope as small as possible.