457 Fundamental

457: Condition Variable Pattern

Functional Programming

Tutorial

The Problem

A mutex protects shared state but provides no way for a thread to wait until the state changes. A consumer thread checking "is the queue empty?" must repeatedly check (busy-wait) or sleep — wasting CPU or adding latency. Condition variables solve this: a thread calls wait (which atomically releases the mutex and sleeps), and another thread calls notify_one or notify_all when the state changes. The waiting thread re-acquires the mutex and re-checks the condition. This is the foundation of thread synchronization: producer-consumer queues, barrier synchronization, and event notification.

Condition variables are used in thread::park, bounded channel internals, database row locking, async runtime wakers, and any producer-consumer pattern requiring efficient sleep-until-ready.

🎯 Learning Outcomes

• Understand how Condvar enables efficient waiting for state changes without busy-waiting

• Learn the Arc<(Mutex<State>, Condvar)> pair pattern for sharing condition variables

• See how wait_while(guard, |s| condition) atomically releases lock and blocks until condition is false

• Understand spurious wakeups and why condition variables always use a predicate loop

• Learn notify_one vs. notify_all semantics

Code Example

#![allow(clippy::all)]
// 457. Condvar for thread notification
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::Duration;

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn test_notify() {
        let p = Arc::new((Mutex::new(false), Condvar::new()));
        let pp = Arc::clone(&p);
        let w = thread::spawn(move || {
            thread::sleep(Duration::from_millis(5));
            *pp.0.lock().unwrap() = true;
            pp.1.notify_one();
        });
        let g = p.1.wait_while(p.0.lock().unwrap(), |&mut v| !v).unwrap();
        assert!(*g);
        w.join().unwrap();
    }
}

(* 457. Condvar – OCaml *)
let m=Mutex.create () let c=Condition.create () let items=Queue.create ()
let done_=ref false

let () =
  let prod = Thread.create (fun () ->
    for i=1 to 5 do
      Thread.delay 0.01;
      Mutex.lock m; Queue.push i items; Condition.signal c; Mutex.unlock m
    done;
    Mutex.lock m; done_:=true; Condition.broadcast c; Mutex.unlock m
  ) () in
  let cons = Thread.create (fun () ->
    let go = ref true in
    while !go do
      Mutex.lock m;
      while Queue.is_empty items && not !done_ do Condition.wait c m done;
      if not (Queue.is_empty items) then
        let v = Queue.pop items in (Mutex.unlock m; Printf.printf "got %d\n%!" v)
      else (Mutex.unlock m; go:=false)
    done
  ) () in
  Thread.join prod; Thread.join cons

Key Differences

Pair pattern: Rust idiomatically pairs (Mutex<T>, Condvar) as a tuple; OCaml typically uses separate mutex and condition values.

wait_while: Rust's Condvar::wait_while handles the spurious wakeup loop automatically; OCaml requires explicit while loop.

Guard reacquisition: Rust's wait returns the MutexGuard when woken; OCaml requires re-locking explicitly after Condition.wait.

Parking: Rust's thread::park + unpark is a simpler one-shot condition variable; OCaml has no equivalent.

OCaml Approach

OCaml's Condition.t provides the same primitive: Condition.wait cond mutex atomically releases the mutex and waits; Condition.signal cond wakes one waiter; Condition.broadcast cond wakes all. The pattern let (mutex, cond) = ... and while not predicate do Condition.wait cond mutex done handles spurious wakeups. OCaml 4.x's Thread.create + Mutex + Condition is the standard synchronization toolkit.

Full Source

#![allow(clippy::all)]
// 457. Condvar for thread notification
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::Duration;

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn test_notify() {
        let p = Arc::new((Mutex::new(false), Condvar::new()));
        let pp = Arc::clone(&p);
        let w = thread::spawn(move || {
            thread::sleep(Duration::from_millis(5));
            *pp.0.lock().unwrap() = true;
            pp.1.notify_one();
        });
        let g = p.1.wait_while(p.0.lock().unwrap(), |&mut v| !v).unwrap();
        assert!(*g);
        w.join().unwrap();
    }
}

(* 457. Condvar – OCaml *)
let m=Mutex.create () let c=Condition.create () let items=Queue.create ()
let done_=ref false

let () =
  let prod = Thread.create (fun () ->
    for i=1 to 5 do
      Thread.delay 0.01;
      Mutex.lock m; Queue.push i items; Condition.signal c; Mutex.unlock m
    done;
    Mutex.lock m; done_:=true; Condition.broadcast c; Mutex.unlock m
  ) () in
  let cons = Thread.create (fun () ->
    let go = ref true in
    while !go do
      Mutex.lock m;
      while Queue.is_empty items && not !done_ do Condition.wait c m done;
      if not (Queue.is_empty items) then
        let v = Queue.pop items in (Mutex.unlock m; Printf.printf "got %d\n%!" v)
      else (Mutex.unlock m; go:=false)
    done
  ) () in
  Thread.join prod; Thread.join cons

✓ Tests Rust test suite

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn test_notify() {
        let p = Arc::new((Mutex::new(false), Condvar::new()));
        let pp = Arc::clone(&p);
        let w = thread::spawn(move || {
            thread::sleep(Duration::from_millis(5));
            *pp.0.lock().unwrap() = true;
            pp.1.notify_one();
        });
        let g = p.1.wait_while(p.0.lock().unwrap(), |&mut v| !v).unwrap();
        assert!(*g);
        w.join().unwrap();
    }
}

Exercises

Bounded producer-consumer: Implement a BoundedQueue<T> using Arc<(Mutex<VecDeque<T>>, Condvar, Condvar)> — one Condvar for "not empty" (unblocks consumers) and one for "not full" (unblocks producers). Test with 4 producers and 4 consumers.

Event broadcast: Create an EventBroadcaster using Condvar::notify_all. When broadcast() is called, all waiting threads receive the event simultaneously. Verify with 10 waiting threads that all unblock on a single broadcast.

Timeout wait: Use Condvar::wait_timeout_while to implement a try_recv_timeout(timeout: Duration) -> Option<T> that waits for an item but gives up after the timeout. Verify it returns None when no producer sends within the timeout.

Open Source Repos

functional-rust

View the source for this example on GitHub — OCaml and Rust side by side in the repo.

Rust