Synchronization Primitives

Condvar

std::sync::Condvar (Condition Variable) is a synchronization primitive that allows a thread to sleep (consuming zero CPU cycles) while atomically releasing a MutexGuard<'_, T> until another thread mutates the state inside the Mutex<T> and notifies the Condvar.

How Condvar::wait Atomically Unlocks and Re-Locks a MutexGuard

A Condvar is always paired with a Mutex<T> ((Mutex<State>, Condvar)). Calling guard = cvar.wait(guard).unwrap() performs three steps:

  1. Atomically unlocks guard and puts the calling thread to sleep in the kernel wait queue (preventing any "lost wakeup" window between checking the condition and going to sleep!).
  2. Waits until another thread calls cvar.notify_one() or cvar.notify_all().
  3. Re-acquires the Mutex lock before returning the fresh MutexGuard<'_, T> to the caller!
Waiting Thread (`cvar.wait_while`)          Notifying Thread (`cvar.notify_one`)
----------------------------------          ------------------------------------
1. Locks `Mutex` -> gets `guard`
2. Checks predicate (`queue.is_empty()`)
3. Calls `cvar.wait(guard)`:
     - Atomically unlocks `Mutex`
     - Sleeps in OS kernel  ..............  4. Locks `Mutex` -> pushes item -> unlocks `Mutex`
                            <.............  5. Calls `cvar.notify_one()`
6. Wakes up & re-acquires `Mutex` `guard`
7. Re-checks predicate -> Proceeds!

Core Condvar Methods

Method Signature Description
Condvar::new() const fn new() -> Condvar Creates a new condition variable (const-constructible)
.wait(guard) fn wait<'a, T>(&self, guard: MutexGuard<'a, T>) -> LockResult<MutexGuard<'a, T>> Unlocks guard, sleeps until notified (or spuriously woken), and re-locks guard
.wait_while(guard, cond) fn wait_while<'a, T, F>(&self, guard: MutexGuard<'a, T>, condition: F) -> LockResult<MutexGuard<'a, T>> Loops calling .wait(guard) for as long as condition(&mut *guard) returns true (automatically handles spurious wakeups!)
.wait_timeout(guard, dur) fn wait_timeout<'a, T>(&self, guard: MutexGuard<'a, T>, dur: Duration) -> LockResult<(MutexGuard<'a, T>, WaitTimeoutResult)> Waits with a timeout; .timed_out() indicates whether dur elapsed
.notify_one() fn notify_one(&self) Wakes up one waiting thread (if any are currently waiting)
.notify_all() fn notify_all(&self) Wakes up all waiting threads

Why wait_while (or a while Loop) Is Mandatory Around .wait()

Operating-system condition variables are subject to spurious wakeups (a waiting thread can wake up even if no thread called notify_one(), or a third thread can steal the queued item before the woken thread re-acquires the mutex). Using cvar.wait_while(guard, |state| ...) automatically re-evaluates the condition in a loop until the desired state holds:

use std::collections::VecDeque;
use std::sync::{Condvar, Mutex};
use std::thread;

struct BoundedQueue<T> {
    buffer: Mutex<VecDeque<T>>,
    not_empty: Condvar,
    not_full: Condvar,
    capacity: usize,
}

impl<T> BoundedQueue<T> {
    fn new(capacity: usize) -> Self {
        Self {
            buffer: Mutex::new(VecDeque::with_capacity(capacity)),
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
            capacity,
        }
    }

    fn push(&self, item: T) {
        let guard = self.buffer.lock().unwrap();
        // Wait while the buffer is full (`len == capacity`):
        let mut guard = self
            .not_full
            .wait_while(guard, |q| q.len() == self.capacity)
            .unwrap();
        guard.push_back(item);
        drop(guard); // Unlock before notifying so the woken consumer doesn't immediately block!
        self.not_empty.notify_one();
    }

    fn pop(&self) -> T {
        let guard = self.buffer.lock().unwrap();
        // Wait while the buffer is empty (`q.is_empty()`):
        let mut guard = self
            .not_empty
            .wait_while(guard, |q| q.is_empty())
            .unwrap();
        let item = guard.pop_front().unwrap();
        drop(guard);
        self.not_full.notify_one();
        item
    }
}

fn main() {
    let queue = BoundedQueue::new(2);

    thread::scope(|s| {
        s.spawn(|| {
            for n in 1..=4 {
                queue.push(n * 10);
            }
        });

        for _ in 1..=4 {
            println!("consumer popped: {}", queue.pop());
        }
    });
}

consumer popped: 10
consumer popped: 20
consumer popped: 30
consumer popped: 40