Shared State: Mutex and RwLock

Tasks that run at the same time often need to read and change the same data. A mutex (mutual exclusion lock) lets one task at a time use the data. A read-write lock lets many tasks read at once, or one task write. This lesson shows which lock type fits async code, how to keep locks from blocking the runtime, and how deadlocks and contention look when you make them deterministic.

API	Purpose
`Arc<std::sync::Mutex<T>>`	short, non-async critical sections
`tokio::sync::Mutex<T>`	a lock whose guard may be held across `.await`
`lock`, `try_lock`, `lock_owned`, `try_lock_owned`, `blocking_lock`	ways to acquire a `Mutex`
`tokio::sync::RwLock<T>`	many readers or one writer, `downgrade`, `with_max_readers`
`TryLockError`	the error when a lock is busy
sharding	several locks instead of one, to cut contention

Several examples use #[tokio::main(flavor = "current_thread", start_paused = true)]. It runs everything on one thread and replaces the real clock with a virtual clock that jumps forward whenever every task is waiting on a timer. It needs the test-util feature of Tokio. Time: sleep, timeout, and interval explains it. With this clock, sleep(10ms) finishes at once in real time, and the elapsed virtual time is exact.

Example 1: A shared counter with Arc<std::sync::Mutex>

Wrap the data in Arc<Mutex<T>> so that every task can own a handle to the same lock. Arc gives shared ownership. Mutex gives exclusive access to the value inside.

Rust

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

#[tokio::main]
async fn main() {
    let counter = Arc::new(Mutex::new(0u32));
    let mut handles = Vec::new();

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        handles.push(tokio::spawn(async move {
            for _ in 0..100 {
                *counter.lock().unwrap() += 1;
            }
        }));
    }

    for handle in handles {
        handle.await.unwrap();
    }
    println!("total = {}", *counter.lock().unwrap());
}

Output

total = 1000

counter.lock() returns a guard. The guard is a temporary, so it is dropped at the end of the statement and the lock is released. Ten tasks each add 100, and no update is lost.

The standard library mutex blocks the calling thread when the lock is taken. That is acceptable here because the lock is held for one addition. The Tokio documentation for Mutex states that using the standard mutex in async code is "ok and often preferred" when the critical section does not contain an .await.

Example 2: Wrap the lock in a type with non-async methods

A critical section is the code that runs while a lock is held. Keep it short and free of .await. One way to guarantee this is to hide the lock behind a struct whose methods are not async. A non-async method cannot hold a guard across an .await, because it has none.

Rust

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

#[derive(Clone, Default)]
struct Db {
    inner: Arc<Mutex<HashMap<String, u32>>>,
}

impl Db {
    fn incr(&self, key: &str) -> u32 {
        let mut map = self.inner.lock().unwrap();
        let value = map.entry(key.to_string()).or_insert(0);
        *value += 1;
        *value
    }

    fn get(&self, key: &str) -> Option<u32> {
        self.inner.lock().unwrap().get(key).copied()
    }
}

#[tokio::main]
async fn main() {
    let db = Db::default();

    let mut handles = Vec::new();
    for i in 0..6 {
        let db = db.clone();
        handles.push(tokio::spawn(async move {
            let key = if i % 2 == 0 { "even" } else { "odd" };
            db.incr(key);
        }));
    }
    for handle in handles {
        handle.await.unwrap();
    }

    println!("even = {:?}", db.get("even"));
    println!("odd = {:?}", db.get("odd"));
    println!("none = {:?}", db.get("none"));
}

Output

even = Some(3)
odd = Some(3)
none = None

Db derives Clone, and cloning it clones the Arc, so every task gets a handle to the same map. The Tokio Mutex documentation describes this wrapper pattern and points to the mini-redis Db as an example.

Example 3: A std guard held across .await does not compile

A std::sync::MutexGuard is not Send: it must be dropped on the thread that locked it. tokio::spawn needs the future to be Send, because the multi-thread runtime can move a task to another worker thread at each .await. A guard that lives across an .await makes the future not Send.

Rust✗ does not compile

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

#[tokio::main]
async fn main() {
    let data = Arc::new(Mutex::new(0));
    let handle = tokio::spawn(async move {
        let mut guard = data.lock().unwrap();
        tokio::time::sleep(Duration::from_millis(10)).await;
        *guard += 1;
    });
    handle.await.unwrap();
}

Output

error: future cannot be sent between threads safely
   --> src/main.rs:7:18
    |
  7 |       let handle = tokio::spawn(async move {
    |  __________________^
  8 | |         let mut guard = data.lock().unwrap();
  9 | |         tokio::time::sleep(Duration::from_millis(10)).await;
 10 | |         *guard += 1;
 11 | |     });
    | |______^ future created by async block is not `Send`
    |
    = help: within `{async block@src/main.rs:7:31: 7:41}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, i32>`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:9:55
    |
  8 |         let mut guard = data.lock().unwrap();
    |             --------- has type `std::sync::MutexGuard<'_, i32>` which is not `Send`
  9 |         tokio::time::sleep(Duration::from_millis(10)).await;
    |                                                       ^^^^^ await occurs here, with `mut guard` maybe used later
note: required by a bound in `tokio::spawn`
   --> ~/.cargo/registry/src/.../tokio-1.53.1/src/task/spawn.rs:176:21
    |
174 |     pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
    |            ----- required by a bound in this function
175 |     where
176 |         F: Future + Send + 'static,
    |                     ^^^^ required by this bound in `spawn`

The compiler rejects the program before it runs. The reason to keep guards away from .await goes beyond this error: the Tokio documentation says that even when the compiler accepts such code (for example when the task cannot move between threads), it "virtually never leads to correct concurrent code in practice as it can easily lead to deadlocks". Example 5 shows the deadlock.

Example 4: Release the guard before the await

Two changes fix Example 3. Either end the guard's scope before the .await, or copy the value out and let the guard drop.

Rust

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

#[tokio::main]
async fn main() {
    let data = Arc::new(Mutex::new(0));

    let scoped = {
        let data = Arc::clone(&data);
        tokio::spawn(async move {
            {
                let mut guard = data.lock().unwrap();
                *guard += 1;
            } // guard dropped here
            tokio::time::sleep(Duration::from_millis(10)).await;
        })
    };

    scoped.await.unwrap();

    let copied = {
        let data = Arc::clone(&data);
        tokio::spawn(async move {
            let current = *data.lock().unwrap(); // guard is a temporary
            tokio::time::sleep(Duration::from_millis(10)).await;
            println!("copied value was {current}");
            *data.lock().unwrap() += 10;
        })
    };

    copied.await.unwrap();
    println!("final = {}", *data.lock().unwrap());
}

Output

copied value was 1
final = 11

The copy in the second task is a value read at one moment. Another task may change the data during the sleep. When the decision must stay valid across the .await, the lock has to be held across it, which needs the Tokio mutex (Example 6).

Example 5: The deadlock a std guard causes when it crosses an await

A future that is not spawned does not need to be Send, so the compiler accepts a guard across an .await inside block_on, join!, or a current_thread runtime. The program then has a hazard: while one future waits with the lock held, another future on the same thread calls lock(), which blocks the whole thread, and the first future never gets to run again.

To show the hazard without hanging, this example calls try_lock in the contender. try_lock on a standard mutex returns an error instead of blocking. tokio::join! runs both futures inside one task on one thread.

Rust

use std::sync::{Arc, Mutex, TryLockError};
use std::time::Duration;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let data = Arc::new(Mutex::new(0));

    let holder = async {
        let mut guard = data.lock().unwrap();
        sleep(Duration::from_millis(10)).await; // guard held across await
        *guard += 1;
        println!("holder finished");
    };

    let contender = async {
        sleep(Duration::from_millis(1)).await;
        match data.try_lock() {
            Ok(_) => println!("contender got the lock"),
            Err(TryLockError::WouldBlock) => {
                println!("contender: lock is held, lock() here would block the thread");
            }
            Err(TryLockError::Poisoned(_)) => println!("poisoned"),
        }
    };

    tokio::join!(holder, contender);
}

Output

contender: lock is held, lock() here would block the thread
holder finished

At 1 ms the holder is suspended at its sleep and still owns the lock. If the contender called data.lock() at this point, the only thread would block inside the contender. The timer that would wake the holder could never fire, so neither future would finish. try_lock reports WouldBlock and the program continues.

Example 6: tokio::sync::Mutex holds across await

tokio::sync::Mutex has an async lock. A task that finds the lock taken is suspended, and the thread runs other tasks. Its guard is Send, so it can be held across .await.

The Tokio documentation names the main use for this type: "shared mutable access to IO resources such as a database connection". The example models a connection whose write takes 10 ms of virtual time. Three tasks share it, and the lock makes the writes happen one after another.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::{sleep, Instant};

struct Connection {
    log: Vec<String>,
}

impl Connection {
    async fn write_line(&mut self, line: String) {
        sleep(Duration::from_millis(10)).await; // stands in for a network write
        self.log.push(line);
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let conn = Arc::new(Mutex::new(Connection { log: Vec::new() }));
    let start = Instant::now();

    let mut handles = Vec::new();
    for i in 0..3 {
        let conn = Arc::clone(&conn);
        handles.push(tokio::spawn(async move {
            let mut guard = conn.lock().await;
            guard.write_line(format!("message {i}")).await;
        }));
    }
    for handle in handles {
        handle.await.unwrap();
    }

    let guard = conn.lock().await;
    println!("log = {:?}", guard.log);
    println!("virtual elapsed = {} ms", start.elapsed().as_millis());
}

Output

log = ["message 0", "message 1", "message 2"]
virtual elapsed = 30 ms

The three writes take 30 ms of virtual time in total, not 10 ms, because only one task at a time can hold the guard. The messages arrive in spawn order because of the queue described in Example 9.

The Tokio documentation also notes that the async mutex "is more expensive than the blocking mutex". It suggests a second design for shared I/O resources: spawn one task that owns the resource and send it messages. mpsc Channels covers channels.

Situation	Lock
Data only, critical section has no `.await`	`Arc<std::sync::Mutex<T>>`
The guard must stay alive across an `.await`	`tokio::sync::Mutex<T>`
Many readers, rare writers, no `.await` in sections	`std::sync::RwLock<T>`
Many readers, rare writers, guard held across `.await`	`tokio::sync::RwLock<T>`
An I/O resource shared by many tasks	one owner task plus a channel, or `tokio::sync::Mutex`

Example 7: try_lock and TryLockError

try_lock returns immediately. It gives Ok(guard) when the lock is free and Err(TryLockError) when someone holds it. The error message is fixed text.

Rust

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

#[tokio::main]
async fn main() {
    let mutex = Arc::new(Mutex::new(5));

    let first = mutex.try_lock();
    println!("first try_lock ok: {}", first.is_ok());

    match mutex.try_lock() {
        Ok(_) => println!("second try_lock ok"),
        Err(e) => println!("second try_lock error: {e}"),
    }

    drop(first);
    println!("after drop, try_lock ok: {}", mutex.try_lock().is_ok());
}

Output

first try_lock ok: true
second try_lock error: operation would block
after drop, try_lock ok: true

try_lock never waits, so it cannot take part in a deadlock. It suits code that has an alternative when the lock is busy: skip the work, retry later, or report "busy". It does not join the wait queue that lock uses.

Example 8: OwnedMutexGuard and lock_owned

lock returns a guard that borrows the mutex, so the guard cannot move into a spawned task, which needs 'static data. lock_owned takes an Arc<Mutex<T>> by value and returns an OwnedMutexGuard<T> that owns a clone of the Arc. The owned guard is 'static and can be moved anywhere.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mutex = Arc::new(Mutex::new(Vec::<u32>::new()));

    // Lock in the parent task, then hand the guard to a child task.
    let guard = Arc::clone(&mutex).lock_owned().await;

    let child = tokio::spawn(async move {
        let mut guard = guard;
        sleep(Duration::from_millis(5)).await;
        guard.push(1);
        println!("child pushed and drops the guard");
    });

    println!("parent try_lock ok: {}", mutex.try_lock().is_ok());
    child.await.unwrap();
    println!("after child, try_lock ok: {}", mutex.try_lock().is_ok());
    println!("contents = {:?}", *mutex.lock().await);

    let owned = Arc::clone(&mutex).try_lock_owned();
    println!("try_lock_owned ok: {}", owned.is_ok());
}

Output

parent try_lock ok: false
child pushed and drops the guard
after child, try_lock ok: true
contents = [1]
try_lock_owned ok: true

The parent locks before it spawns, so the lock is certainly held when the child starts. The guard moves into the child, and the lock is released when the child drops it. try_lock_owned is the non-waiting version of lock_owned.

Example 9: Fairness: the mutex is a FIFO queue

The Tokio documentation states: "Tokio's Mutex operates on a guaranteed FIFO basis. This means that the order in which tasks call the lock method is the exact order in which they will acquire the lock." FIFO means first in, first out.

The example holds the lock, starts four tasks at staggered virtual times, and releases the lock. The tasks acquire it in the order in which they called lock.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let order = Arc::new(Mutex::new(Vec::new()));
    let gate = Arc::new(Mutex::new(()));

    let held = gate.lock().await;
    let mut handles = Vec::new();
    for i in 0..4u64 {
        let gate = Arc::clone(&gate);
        let order = Arc::clone(&order);
        handles.push(tokio::spawn(async move {
            sleep(Duration::from_millis(i)).await; // task i calls lock at i ms
            let _turn = gate.lock().await;
            order.lock().await.push(i);
        }));
    }

    sleep(Duration::from_millis(20)).await; // all four are queued
    drop(held);
    for handle in handles {
        handle.await.unwrap();
    }
    println!("acquired in order: {:?}", order.lock().await);
}

Output

acquired in order: [0, 1, 2, 3]

Task 0 called lock first and got the lock first. The mutex does not let a later caller jump ahead, and it does not starve an earlier caller.

Example 10: Cancelling lock loses the place in the queue

The lock documentation says: "Cancelling a call to lock makes you lose your place in the queue." A timeout around lock cancels the wait when the time is up. A retry then joins the back of the queue.

Task A calls lock first but gives up after 5 ms and retries. Task B calls lock at 2 ms and never gives up. B ends up ahead of A.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::{sleep, timeout};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let gate = Arc::new(Mutex::new(()));
    let order = Arc::new(std::sync::Mutex::new(Vec::new()));

    let held = gate.lock().await;

    let a = {
        let (gate, order) = (Arc::clone(&gate), Arc::clone(&order));
        tokio::spawn(async move {
            sleep(Duration::from_millis(1)).await;
            // First attempt: first in the queue, but gives up after 5 ms.
            let first = timeout(Duration::from_millis(5), gate.lock()).await;
            println!("A first attempt timed out: {}", first.is_err());
            drop(first);
            // Retry: goes to the back of the queue.
            let _turn = gate.lock().await;
            order.lock().unwrap().push("A");
        })
    };
    let b = {
        let (gate, order) = (Arc::clone(&gate), Arc::clone(&order));
        tokio::spawn(async move {
            sleep(Duration::from_millis(2)).await;
            let _turn = gate.lock().await;
            order.lock().unwrap().push("B");
        })
    };

    sleep(Duration::from_millis(20)).await;
    drop(held);
    a.await.unwrap();
    b.await.unwrap();
    println!("order = {:?}", order.lock().unwrap());
}

Output

A first attempt timed out: true
order = ["B", "A"]

A entered the queue at 1 ms and B at 2 ms. At 6 ms A's timeout drops its lock future, which removes A from the queue. A's retry enters behind B. This is why a loop of timeout(.., lock()) retries is not fair to the retrying task.

Example 11: Deadlock 1: two locks taken in opposite orders

A deadlock is a state where tasks wait for each other and none can continue. With two locks, task 1 takes a then wants b, while task 2 takes b then wants a. Each holds what the other needs.

A real deadlock would hang the program. Wrapping the second lock in timeout turns the hang into a result that can be printed.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::{sleep, timeout};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let a = Arc::new(Mutex::new(0));
    let b = Arc::new(Mutex::new(0));

    let task1 = async {
        let _ga = a.lock().await;
        sleep(Duration::from_millis(10)).await;
        let result = match timeout(Duration::from_millis(50), b.lock()).await {
            Ok(_) => "task1 got both locks",
            Err(_) => "task1 timed out waiting for b",
        };
        sleep(Duration::from_millis(100)).await; // keeps holding `a` after the timeout
        result
    };
    let task2 = async {
        let _gb = b.lock().await;
        sleep(Duration::from_millis(10)).await;
        let result = match timeout(Duration::from_millis(50), a.lock()).await {
            Ok(_) => "task2 got both locks",
            Err(_) => "task2 timed out waiting for a",
        };
        sleep(Duration::from_millis(100)).await; // keeps holding `b` after the timeout
        result
    };

    let (r1, r2) = tokio::join!(task1, task2);
    println!("{r1}");
    println!("{r2}");
}

Output

task1 timed out waiting for b
task2 timed out waiting for a

Each task keeps its first lock for another 100 ms after the timeout, so both waits fail. Without the timeouts, both futures would wait forever. The fix is a global lock order: every task that needs both locks takes a first.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::{sleep, timeout};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let a = Arc::new(Mutex::new(0));
    let b = Arc::new(Mutex::new(0));

    let task1 = async {
        let _ga = a.lock().await;
        sleep(Duration::from_millis(10)).await;
        let got_b = timeout(Duration::from_millis(50), b.lock()).await.is_ok();
        format!("task1 got both locks: {got_b}")
    };
    let task2 = async {
        // Same order as task1: a first, then b.
        let got_a = timeout(Duration::from_millis(50), a.lock()).await;
        match got_a {
            Ok(_ga) => {
                let got_b = timeout(Duration::from_millis(50), b.lock()).await.is_ok();
                format!("task2 got both locks: {got_b}")
            }
            Err(_) => "task2 timed out waiting for a".to_string(),
        }
    };

    let (r1, r2) = tokio::join!(task1, task2);
    println!("{r1}");
    println!("{r2}");
}

Output

task1 got both locks: true
task2 got both locks: true

Task 1 holds a and finishes with b; task 2 waits for a in the same order, so it gets both locks after task 1 releases them. The rule that prevents this deadlock is consistent lock order. When the order cannot be fixed, hold only one lock at a time, or replace the two locks with one lock around a struct that holds both values.

Example 12: Deadlock 2: locking twice in the same task

Neither std::sync::Mutex nor tokio::sync::Mutex is reentrant: a task that holds the lock and calls lock again waits for itself. The state is the same as any deadlock, with one task and one lock.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::timeout;

async fn add_one(counter: &Mutex<u32>) {
    *counter.lock().await += 1;
}

async fn add_two(counter: &Mutex<u32>) {
    let mut guard = counter.lock().await;
    *guard += 1;
    // BUG: add_one locks again while `guard` is alive.
    if timeout(Duration::from_millis(50), add_one(counter)).await.is_err() {
        println!("add_two: nested lock timed out");
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let counter = Arc::new(Mutex::new(0));
    add_two(&counter).await;
    println!("try_lock after add_two: {}", counter.try_lock().is_ok());
    println!("value = {}", *counter.lock().await);
}

Output

add_two: nested lock timed out
try_lock after add_two: true
value = 1

The nested add_one waited until the timeout dropped it. The outer guard was alive the whole time. The fix is to release the guard before calling a function that locks, or to pass the guard (or &mut T) into the helper so that only one function locks.

Example 13: RwLock: many readers or one writer

tokio::sync::RwLock<T> allows any number of read guards, or one write guard. It is useful for data that is read often and changed rarely. Like the mutex, its guards can be held across .await. try_read and try_write do not wait.

Rust

use tokio::sync::RwLock;

#[tokio::main]
async fn main() {
    let lock = RwLock::new(String::from("v1"));

    let r1 = lock.read().await;
    let r2 = lock.read().await;
    println!("two readers: {} {}", *r1, *r2);
    println!("try_write while readers exist: {}", lock.try_write().is_ok());
    println!("try_read while readers exist: {}", lock.try_read().is_ok());
    drop(r1);
    drop(r2);

    let mut w = lock.write().await;
    w.push_str("-updated");
    println!("try_read while writer exists: {}", lock.try_read().is_ok());
    drop(w);

    println!("value = {}", *lock.read().await);
}

Output

two readers: v1 v1
try_write while readers exist: false
try_read while readers exist: true
try_read while writer exists: false
value = v1-updated

Readers do not exclude each other, but any reader excludes a writer, and a writer excludes everyone. Sharing reads helps only when several tasks read at the same time. If writes are frequent, or reads are very short, a plain mutex gives the same result with a simpler policy.

Example 14: Fairness of RwLock: a waiting writer blocks new readers

The RwLock documentation states: "The priority policy of Tokio's read-write lock is fair (or write-preferring), in order to ensure that readers cannot starve writers. Fairness is ensured using a first-in, first-out queue for the tasks awaiting the lock; a read lock will not be given out until all write lock requests that were queued before it have been acquired and released."

The consequence: while a reader holds the lock and a writer waits, a new reader queues behind the writer even though only readers hold the lock now.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let lock = Arc::new(RwLock::new(0));

    let r1 = lock.read().await;
    println!("try_read with only a reader: {}", lock.try_read().is_ok());

    let writer = {
        let lock = Arc::clone(&lock);
        tokio::spawn(async move {
            *lock.write().await += 1;
        })
    };
    sleep(Duration::from_millis(1)).await; // the writer is now queued

    println!("try_read with a writer queued: {}", lock.try_read().is_ok());

    drop(r1);
    writer.await.unwrap();
    println!("try_read after the writer: {}", lock.try_read().is_ok());
}

Output

try_read with only a reader: true
try_read with a writer queued: false
try_read after the writer: true

Before the writer arrives, try_read succeeds. Once the writer is queued, try_read fails, although the current holder is a reader. A read-preferring lock would let the new reader in and could delay the writer forever if readers keep arriving. Tokio's policy prevents that starvation.

Example 15: Deadlock 3: a second read while a writer waits

The write-preferring policy has a consequence for nested reads. A task holds a read guard and takes a second read guard while a writer is queued. The second read waits behind the writer, and the writer waits for the first read guard, which the same task holds.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use tokio::time::{sleep, timeout};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let lock = Arc::new(RwLock::new(0));

    let first_read = lock.read().await;

    let writer = {
        let lock = Arc::clone(&lock);
        tokio::spawn(async move {
            *lock.write().await += 1;
        })
    };
    sleep(Duration::from_millis(1)).await; // writer is queued

    // Recursive read: waits behind the writer, which waits for `first_read`.
    let second = timeout(Duration::from_millis(50), lock.read()).await;
    println!("second read timed out: {}", second.is_err());

    drop(first_read);
    writer.await.unwrap();
    println!("value = {}", *lock.read().await);
}

Output

second read timed out: true
value = 1

Without the timeout, the program would wait forever. The rule: never acquire a read lock while the same task already holds one on the same RwLock. Pass the guard to the callee instead.

Example 16: Downgrade a write guard to a read guard

RwLockWriteGuard::downgrade turns a write guard into a read guard without releasing the lock in between. No other writer can enter between the write and the read, so the task can write a value and then read it back knowing it has not changed. Other readers can join once the guard is downgraded.

Rust

use tokio::sync::RwLock;

#[tokio::main]
async fn main() {
    let lock = RwLock::new(1);

    let mut write = lock.write().await;
    *write += 1;
    let read = write.downgrade();

    println!("value after downgrade = {}", *read);
    println!("try_write while downgraded guard exists: {}", lock.try_write().is_ok());
    println!("try_read while downgraded guard exists: {}", lock.try_read().is_ok());
    drop(read);
    println!("try_write after drop: {}", lock.try_write().is_ok());
}

Output

value after downgrade = 2
try_write while downgraded guard exists: false
try_read while downgraded guard exists: true
try_write after drop: true

RwLock::with_max_readers(value, n) limits the number of concurrent read guards to n. It panics if n is 0 or larger than u32::MAX >> 3, a limit read from the Tokio source.

Example 17: A panic does not poison a Tokio mutex

std::sync::Mutex is poisoned when a thread panics while holding the guard: later lock calls return Err. The Tokio documentation states that its mutex "does not poison the mutex when a thread holding the MutexGuard panics. In such a case, the mutex will be unlocked. If the panic is caught, this might leave the data protected by the mutex in an inconsistent state."

Rust

use std::sync::Arc;

#[tokio::main]
async fn main() {
    let tokio_mutex = Arc::new(tokio::sync::Mutex::new(0));
    let std_mutex = Arc::new(std::sync::Mutex::new(0));

    let t = Arc::clone(&tokio_mutex);
    let s = Arc::clone(&std_mutex);
    let handle = tokio::spawn(async move {
        let mut a = t.lock().await;
        let mut b = s.lock().unwrap();
        *a += 1;
        *b += 1;
        panic!("failed while holding both locks");
    });

    let result = handle.await;
    println!("task panicked: {}", result.unwrap_err().is_panic());
    println!("tokio mutex usable: {}", tokio_mutex.try_lock().is_ok());
    println!("std mutex poisoned: {}", std_mutex.lock().is_err());
    println!("tokio value after panic = {}", *tokio_mutex.lock().await);
}

Output

task panicked: true
tokio mutex usable: true
std mutex poisoned: true
tokio value after panic = 1

The Tokio mutex is usable again, and its value is the half-finished 1. Code that must keep an invariant across a panic needs to restore it itself, because nothing marks the data as suspect.

Example 18: blocking_lock from a blocking thread

blocking_lock acquires a Tokio Mutex from synchronous code. It blocks the thread, so it must run on a thread that is allowed to block, such as a spawn_blocking thread (Blocking and CPU-Bound Work). Calling it from async code panics.

Rust

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

#[tokio::main]
async fn main() {
    let mutex = Arc::new(Mutex::new(Vec::<u32>::new()));

    let worker = Arc::clone(&mutex);
    tokio::task::spawn_blocking(move || {
        for i in 0..3 {
            worker.blocking_lock().push(i);
        }
    })
    .await
    .unwrap();

    println!("contents = {:?}", *mutex.lock().await);
}

Output

contents = [0, 1, 2]

The same Mutex value serves both sync and async callers. The async side uses lock().await, and the blocking side uses blocking_lock().

Calling blocking_lock from async code panics, because blocking a runtime thread is not allowed there:

Rustpanics at runtime

use tokio::sync::Mutex;

#[tokio::main]
async fn main() {
    let mutex = Mutex::new(0);
    let _guard = mutex.blocking_lock(); // called inside an async context
}

Output

thread 'main' panicked at src/main.rs:6:24:
Cannot block the current thread from within a runtime. This happens because a function attempted to block the current thread while the thread is being used to drive asynchronous tasks.

Example 19: Contention and sharding

Contention means that tasks spend time waiting for a lock instead of working. A single lock around a whole map makes every operation wait for every other one, even when they touch different keys.

Sharding splits the data across several locks. A key picks its shard, for example by key % shard_count. Operations on different shards do not wait for each other. The example makes each critical section last 10 ms of virtual time, so the effect shows in the virtual clock.

Rust

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::{sleep, Instant};

struct Sharded {
    shards: Vec<Mutex<HashMap<u64, u64>>>,
}

impl Sharded {
    fn new(count: usize) -> Self {
        Sharded {
            shards: (0..count).map(|_| Mutex::new(HashMap::new())).collect(),
        }
    }

    async fn update(&self, key: u64) {
        let shard = &self.shards[(key as usize) % self.shards.len()];
        let mut map = shard.lock().await;
        sleep(Duration::from_millis(10)).await; // the critical section takes 10 ms
        *map.entry(key).or_insert(0) += 1;
    }
}

async fn run(shard_count: usize, keys: &[u64]) -> u128 {
    let db = Arc::new(Sharded::new(shard_count));
    let start = Instant::now();
    let mut handles = Vec::new();
    for &key in keys {
        let db = Arc::clone(&db);
        handles.push(tokio::spawn(async move { db.update(key).await }));
    }
    for handle in handles {
        handle.await.unwrap();
    }
    start.elapsed().as_millis()
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    println!("1 lock,  keys 0..4:      {} ms", run(1, &[0, 1, 2, 3]).await);
    println!("4 shards, keys 0..4:     {} ms", run(4, &[0, 1, 2, 3]).await);
    println!("4 shards, keys 0,4,8,12: {} ms", run(4, &[0, 4, 8, 12]).await);
}

Output

1 lock,  keys 0..4:      40 ms
4 shards, keys 0..4:     10 ms
4 shards, keys 0,4,8,12: 40 ms

One lock serializes four updates: 40 ms. Four shards with keys that land in different shards run the four updates in parallel: 10 ms. Keys 0, 4, 8, and 12 all map to shard 0, so sharding does not help and the time is 40 ms again. Sharding cuts contention only when the keys spread across shards. The number of shards is a trade-off: more shards mean less waiting and more memory, and an operation that needs the whole map has to lock every shard.

Example 20: Probing which shard is busy with try_lock

try_lock shows which parts of a sharded structure are busy, without waiting. The example holds shard 1 and checks each shard.

Rust

use tokio::sync::Mutex;

#[tokio::main]
async fn main() {
    let shards: Vec<Mutex<u32>> = (0..4).map(|_| Mutex::new(0)).collect();

    let held = shards[1].lock().await;

    for (index, shard) in shards.iter().enumerate() {
        println!("shard {index} free: {}", shard.try_lock().is_ok());
    }
    drop(held);
    println!("after release, shard 1 free: {}", shards[1].try_lock().is_ok());
}

Output

shard 0 free: true
shard 1 free: false
shard 2 free: true
shard 3 free: true
after release, shard 1 free: true

Only shard 1 reports busy. A task that works on key 5 (5 % 4 = 1) would wait; a task that works on key 6 (6 % 4 = 2) would not.

Interview points

Arc<std::sync::Mutex<T>> is correct in async code when the critical section has no .await. The Tokio documentation prefers it to the async mutex for plain data.
A std::sync::MutexGuard is not Send. A guard held across .await in a spawned task fails to compile. In non-spawned code it compiles and can block the only thread that could release the lock.
tokio::sync::Mutex exists so that the guard can live across .await. It costs more than the standard mutex.
Tokio's Mutex is FIFO. The order of lock calls is the order of acquisition. Cancelling lock gives up the queue position.
tokio::sync::RwLock is write-preferring: a queued writer blocks readers that arrive later. A second read by the same task while a writer waits deadlocks.
Neither mutex is reentrant. Locking twice in one task deadlocks.
Two locks taken in opposite orders deadlock. Use one global order, hold one lock at a time, or merge the data behind one lock.
The Tokio mutex is not poisoned by a panic. The std mutex is.
try_lock, try_read, and try_write never wait. They return TryLockError (message: "operation would block") when the lock is busy.
lock_owned and try_lock_owned return OwnedMutexGuard, which owns an Arc and can move into a spawned task.
blocking_lock is for synchronous threads, and it panics in async context.
Sharding cuts contention only if keys spread across shards. Whole-map operations must lock every shard.

Summary

Choose the lock by what the critical section does: no .await means the std mutex, an .await inside means the Tokio mutex.
Keep critical sections short, and hide locks behind non-async methods to make a guard across .await impossible.
Tokio Mutex and RwLock queue waiters in FIFO order, so they are fair. RwLock gives writers priority over readers that arrive later.
Deadlocks come from lock-order inversion, re-locking in one task, and recursive reads with a queued writer. try_lock and timeout make them observable.
OwnedMutexGuard and lock_owned move a held lock into another task.
RwLockWriteGuard::downgrade changes a write guard into a read guard without a gap.
Sharding splits one lock into several. Its benefit depends on how keys spread.
Channels are an alternative to locks for shared I/O resources. See mpsc Channels, and the Rust course lessons Shared State and Async Channels and Locks for further reading.