Safe Concurrency in Rust with Mutex and Channel

1. Overview

Rust calls its capability “fearless concurrency”. The compiler blocks data races at build time, before the program ever runs. This article covers two main tools from the standard library: Mutex (with Arc) for sharing state, and mpsc channels for passing messages between threads. After reading, you’ll know how to use each type, write tests for them, and avoid common pitfalls like deadlock or holding locks too long.

All examples use only the standard library, no external dependencies needed. You can create a project with cargo new orders-demo and paste the code into src/main.rs.

2. The Problem: Sharing Data Between Threads

Suppose you have multiple threads processing orders and want to count the total. A naive approach looks like this:

use std::thread;

fn main() {
    let mut total_orders = 0;

    let handles: Vec<_> = (0..4)
        .map(|_| thread::spawn(|| total_orders += 1))
        .collect();

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

This code doesn’t compile:

error[E0373]: closure may outlive the current function, but it borrows `total_orders`, which is owned by the current function

The compiler gives two reasons:

• thread::spawn requires the closure to be 'static. Child threads may outlive main, so they can’t borrow local variables.
• Even with move, each thread gets its own copy of total_orders. No one is actually accumulating into a shared variable.

To solve this, you need two things: shared ownership via Arc and synchronized access via Mutex.

3. Sharing State with Arc<Mutex<T>>

3.1. Counting Orders from Multiple Threads

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

fn main() {
    let total_orders = Arc::new(Mutex::new(0u32));
    let mut handles = Vec::new();

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

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

    println!("Total orders: {}", *total_orders.lock().unwrap());
}

Total orders: 4000

Here’s a breakdown of each part:

• Arc::new(Mutex::new(0u32)): Arc (Atomic Reference Counted) allows multiple threads to own the same value. Mutex ensures that only one thread accesses the value inside at any given time.
• Arc::clone(&total_orders) only increments the reference count, it doesn’t copy the data. Each thread gets its own Arc via move.
• counter.lock() blocks the current thread until it acquires the lock. It returns Result<MutexGuard<u32>, PoisonError<…>>.
• MutexGuard dereferences to &mut u32, so *... += 1 modifies the value directly. The guard is dropped at the end of the statement, and the lock is released automatically.

You can’t “forget to unlock” in Rust. You also can’t read the value without the lock, because the data lives inside the Mutex.

Note: You can’t replace Arc with Rc. Rc doesn’t implement Send, and the compiler will error as soon as you pass it to thread::spawn. Similarly, you can’t replace Mutex with RefCell because RefCell isn’t Sync.

3.2. The Scope of MutexGuard

The lock is held until the guard is dropped. If you assign the guard to a variable, the lock lives until the end of that variable’s scope:

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

fn reserve(inventory: &Mutex<HashMap<String, u32>>, sku: &str) -> bool {
    let reserved = {
        let mut stock = inventory.lock().unwrap();
        match stock.get_mut(sku) {
            Some(qty) if *qty > 0 => {
                *qty -= 1;
                true
            }
            _ => false,
        }
    }; // guard is dropped here, lock is released

    if reserved {
        // Simulating a slow PaymentService call, NOT holding the lock
        thread::sleep(Duration::from_millis(50));
    }
    reserved
}

fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("AO-THUN".to_string(), 2)])));

    let handles: Vec<_> = (0..3)
        .map(|i| {
            let inv = Arc::clone(&inventory);
            thread::spawn(move || println!("Customer {i}: {}", reserve(&inv, "AO-THUN")))
        })
        .collect();

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

Customer 0: true
Customer 1: true
Customer 2: false

The order of lines may vary between runs. The result is always exactly two true and one false.

The { ... } block limits the lock-holding time to a minimum. The slow part (payment call) runs after the lock is released, so other threads don’t have to wait. If you need to release it earlier in the middle, you can explicitly call drop(stock).

3.3. Mutex Poisoning

If a thread panics while holding a lock, Rust marks the Mutex as poisoned. The reason is that the data might be in an inconsistent state:

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

fn main() {
    let balance = Arc::new(Mutex::new(100i64));

    let b = Arc::clone(&balance);
    let _ = thread::spawn(move || {
        let mut guard = b.lock().unwrap();
        *guard -= 30;
        panic!("PaymentService error midway");
    })
    .join();

    match balance.lock() {
        Ok(v) => println!("Balance: {}", *v),
        Err(poisoned) => {
            let v = poisoned.into_inner();
            println!("Mutex is poisoned, current balance: {}", *v);
        }
    }
}

thread '<unnamed>' panicked at src/main.rs:10:9:
PaymentService error midway
Mutex is poisoned, current balance: 70

So .unwrap() after lock() essentially means “if another thread panicked while holding the lock, panic too”. For most applications, that’s a reasonable choice. If you know how to recover the data, into_inner() returns the guard so you can inspect and fix it.

4. Passing Messages with Channels

Mutex answers the question “multiple threads modifying the same data”. Channels go the opposite direction: each piece of data has only one owner at a time, and ownership is transferred through the channel. In many scenarios, this approach is simpler and less error-prone.

4.1. Multiple Producers, One Consumer

std::sync::mpsc stands for multiple producer, single consumer:

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

#[derive(Debug)]
struct Order {
    id: u32,
    customer: String,
    amount: u64,
}

fn main() {
    let (tx, rx) = mpsc::channel::<Order>();

    for shop in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let order = Order {
                id: shop * 100,
                customer: format!("customer-{shop}"),
                amount: 150_000 * shop as u64,
            };
            tx.send(order).unwrap();
        });
    }
    drop(tx); // important: drop the original sender

    let mut revenue = 0;
    for order in rx {
        println!("Received order #{} from {}", order.id, order.customer);
        revenue += order.amount;
    }
    println!("Revenue: {revenue}");
}

Received order #100 from customer-1
Received order #300 from customer-3
Received order #200 from customer-2
Revenue: 900000

Here’s what’s happening:

• mpsc::channel() returns a pair (Sender, Receiver). Each producer gets a clone via tx.clone().
• tx.send(order) moves order into the channel. After this line, the sending thread can’t use order anymore, and the compiler will error if you try. This ensures two threads never modify the same Order.
• for order in rx iterates until all Senders are dropped. If you omit drop(tx), the original sender still lives in main, and the loop will wait forever.

The order of reception depends on thread scheduling. The total revenue is always 900000.

4.2. Bounded Channels and Backpressure

mpsc::channel() has an unlimited buffer. If the producer is faster than the consumer, memory will grow. sync_channel(n) caps the buffer at n items, and send blocks when full:

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u32>(2);

    let producer = thread::spawn(move || {
        for id in 1..=5 {
            tx.send(id).unwrap();
            println!("Sent order {id}");
        }
    });

    thread::sleep(Duration::from_millis(100));
    for id in rx {
        println!("Processing order {id}");
        thread::sleep(Duration::from_millis(20));
    }
    producer.join().unwrap();
}

Sent order 1
Sent order 2
Processing order 1
Sent order 3
Processing order 2
...

The producer sends two orders then must wait for the consumer to pull them. This is natural backpressure, very useful for data processing pipelines. With sync_channel(0), each send must wait until someone recvs (rendezvous).

5. When You Don’t Need Arc: Scoped Threads

If threads are guaranteed to finish before the function returns, thread::scope lets you borrow data directly without Arc:

use std::sync::Mutex;
use std::thread;

fn main() {
    let processed = Mutex::new(Vec::new());
    let orders = vec![101, 102, 103, 104];

    thread::scope(|s| {
        for chunk in orders.chunks(2) {
            s.spawn(|| processed.lock().unwrap().extend_from_slice(chunk));
        }
    }); // all threads in the scope have joined here

    let mut result = processed.into_inner().unwrap();
    result.sort();
    println!("{result:?}");
}

[101, 102, 103, 104]

With scope, the compiler knows threads don’t outlive processed and orders. So borrowing with & is valid. After the scope, into_inner() extracts data from the Mutex without acquiring the lock.

6. Testing Concurrent Code

You should extract logic into pure functions for easier testing. Put the following in src/lib.rs:

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

pub fn count_orders(workers: u32, per_worker: u32) -> u32 {
    let total = Arc::new(Mutex::new(0));
    let handles: Vec<_> = (0..workers)
        .map(|_| {
            let total = Arc::clone(&total);
            thread::spawn(move || {
                for _ in 0..per_worker {
                    *total.lock().unwrap() += 1;
                }
            })
        })
        .collect();
    handles.into_iter().for_each(|h| h.join().unwrap());
    let result = *total.lock().unwrap();
    result
}

pub fn sum_payments(payments: Vec<u64>) -> u64 {
    let (tx, rx) = mpsc::channel();
    for amount in payments {
        let tx = tx.clone();
        thread::spawn(move || tx.send(amount).unwrap());
    }
    drop(tx);
    rx.iter().sum()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn counts_every_order_exactly_once() {
        assert_eq!(count_orders(8, 10_000), 80_000);
    }

    #[test]
    fn sums_all_payments_from_channel() {
        assert_eq!(sum_payments(vec![50_000, 120_000, 30_000]), 200_000);
    }

    #[test]
    fn empty_payments_returns_zero() {
        assert_eq!(sum_payments(vec![]), 0);
    }
}

cargo test

running 3 tests
test tests::empty_payments_returns_zero ... ok
test tests::sums_all_payments_from_channel ... ok
test tests::counts_every_order_exactly_once ... ok

test result: ok. 3 passed; 0 failed

The empty_payments_returns_zero test checks the edge case: if you forget drop(tx), this test will hang forever instead of returning 0. That’s a clear signal to catch the bug.

There’s a subtle point in count_orders: the value is assigned to a result variable before returning. The reason is that a temporary guard in the final expression could outlive total, and the compiler would report a borrow error.

7. Common Mistakes

Deadlock from reversed lock order. Thread A locks orders then customers, while thread B locks customers then orders. Both will wait forever. Rust only prevents data races, not deadlocks. You should establish a fixed lock order, or combine both pieces of data into a single Mutex<(A, B)>.

Temporaries live longer than you think. Consider this code:

while let Some(order) = queue.lock().unwrap().pop() {
    process(order); // the lock is still being held here!
}

Temporaries in while let (and match) expressions live until the end of the loop body. Any other thread wanting to push to queue will be blocked the entire time process runs. The fix is to separate it: let next = queue.lock().unwrap().pop(); then process afterward.

Calling lock() twice on the same thread. std::sync::Mutex is not reentrant. Calling lock() twice on the same thread will deadlock or panic.

Holding a lock while waiting for I/O or recv(). Any blocking operation while holding a guard turns Mutex into a bottleneck. You should copy or clone the needed data out, release the lock, then do the slow work.

Forgetting to drop Sender. As seen above, the consumer will wait forever.

Note: In async code (Tokio), don’t hold std::sync::MutexGuard across .await. Use tokio::sync::Mutex or Tokio’s channels instead. Also, for simple integers like counters, AtomicU32 is much faster than Mutex<u32>.

When to choose which?

• Use Mutex when multiple threads need to read/modify the same long-lived state, like a cache or inventory. If reads vastly outnumber writes, consider RwLock.
• Use channels when data flows from one place to another, like order queues, pipelines, or worker pools. Channels ensure each value has one owner, making the logic easier to reason about.

8. Conclusion

Rust turns common concurrency errors into compile errors. You can’t share data without Arc, can’t modify data without Mutex, and can’t reuse a value that’s been sent through a channel. Arc<Mutex<T>> suits shared state, while mpsc channels suit transferring ownership between threads. sync_channel adds backpressure, and thread::scope lets you omit Arc when thread lifetimes are clear. What the compiler can’t protect against is deadlock, guards living too long, and forgotten dropped senders. Those are spots where you need careful review and thorough tests.