Queues and bounded buffers

Chapter 16 split one job across threads. Many programs have a different shape. Some threads produce work: they read requests, parse files, or receive messages. Other threads consume it: they process each item. The two groups run at different speeds, so they need a place to leave work for each other.

That place is a queue. Producers add items at the back, and consumers take them from the front. Items come out in the order they went in, which is called FIFO, for first in, first out.

This chapter builds the queue six times. Each version fixes a limitation of the one before, and the last one removes the lock.

17.1 Why a queue needs a limit

Suppose producers add items faster than consumers can handle them. If the queue can grow without limit, it grows until the machine runs out of memory, and the program is killed (figure 17.1).

A bounded queue has a fixed capacity. When it is full, push waits until a consumer makes room. The producer slows down to the consumer’s speed. Slowing down the source of work when the rest of the system cannot keep up is called backpressure. A slow system is better than a crashed one.

A bounded queue with blocking push and pop is also called a bounded buffer.

17.2 A shared queue

The first version is unbounded. It shows the two tools every version uses: a Mutex around a VecDeque, and a Condvar for waiting.

VecDeque from chapter 3 adds at the back and removes from the front in O(1). The Mutex makes sure one thread at a time touches it.

The Condvar is for consumers that find the queue empty. Section 16.5 introduced condition variables. A thread waits on one while holding a mutex, and the wait releases the mutex and sleeps in one step. When another thread calls notify_one, one sleeping thread wakes up with the mutex locked again.

Listing 17.1 The queue (lines 7 to 29). src/bin/thread_safe_queue.rs

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

#[derive(Default)]
struct SyncQueue {
    items: Mutex<VecDeque<i32>>,
    not_empty: Condvar,
}

impl SyncQueue {
    fn push(&self, item: i32) {
        if let Ok(mut q) = self.items.lock() {
            q.push_back(item);
            self.not_empty.notify_one();
        }
    }

    fn pop(&self) -> Option<i32> {
        let q = self.items.lock().ok()?;
        let mut q = self
            .not_empty
            .wait_while(q, |q| q.is_empty())
            .ok()?;
        q.pop_front()
    }
}

push locks the queue, adds the item, and wakes one waiting consumer. if let Ok(mut q) skips the push if the lock is poisoned, so in that case the item is silently lost. Section 17.4 shows better ways to handle that.

pop uses wait_while. It takes the guard and a condition, and waits for as long as the condition is true:

let mut q = self.not_empty.wait_while(q, |q| q.is_empty()).ok()?;

wait_while contains the while loop that section 16.5 wrote by hand. It checks the condition, waits if it is true, and checks again after each wakeup. It returns the guard when the queue is not empty. The .ok()? turns a poisoned lock into None.

Animation 17.1 follows one item from a producer to a consumer that is asleep in wait_while. Then it takes out notify_one.

Listing 17.2 main (lines 31 to 56). src/bin/thread_safe_queue.rs

fn main() {
    let q = Arc::new(SyncQueue::default());

    // producer
    let producer = {
        let q = Arc::clone(&q);
        thread::spawn(move || {
            for i in 0..5 {
                q.push(i);
                println!("Producer pushed {i}");
            }
        })
    };

    let consumer = {
        let q = Arc::clone(&q);
        thread::spawn(move || {
            for _ in 0..5 {
                println!("Cosnumer popped: {:?}", q.pop());
            }
        })
    };

    producer.join().unwrap();
    consumer.join().unwrap();
}

main shares the queue through an Arc. Each thread is created inside a block that makes its own clone of the Arc first:

let producer = {
    let q = Arc::clone(&q);
    thread::spawn(move || { ... })
};

The block keeps the clone’s name short, and the outer q stays available for the next thread.

$ cargo run --bin thread_safe_queue
Producer pushed 0
Producer pushed 1
Producer pushed 2
Producer pushed 3
Producer pushed 4
Cosnumer popped: Some(0)
Cosnumer popped: Some(1)
Cosnumer popped: Some(2)
Cosnumer popped: Some(3)
Cosnumer popped: Some(4)

In this run the producer finished before the consumer started. On another run the lines can interleave. The values always come out in the order 0 to 4, because the queue is FIFO.

17.3 A bounded buffer

The bounded version adds a capacity, and a second reason to wait: a producer must wait while the queue is full. So it has two condition variables, one for each kind of waiting thread (figure 17.2).

• not_empty is where consumers wait. A producer notifies it after adding an item.
• not_full is where producers wait. A consumer notifies it after removing an item.

With a single condition variable and notify_one, a producer could wake another producer instead of a consumer. The woken producer would find the queue still full and sleep again, while the consumer that could make progress never woke. Every thread could end up asleep. notify_all on one condition variable avoids that, but wakes every thread for each change. Two condition variables wake the right kind of thread directly.

Listing 17.3 The type and new (lines 19 to 38). src/bin/bounded_buffer.rs

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

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

impl<T> BoundedQueue<T> {
    fn new(capacity: usize) -> Self {
        // Capacity 0 makes no sense: no producer could ever push,
        // and every pop would block forever.
        assert!(capacity > 0, "capacity must be > 0");

        Self {
            inner: Mutex::new(VecDeque::with_capacity(capacity)),
            capacity,
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }
    // ...
}

VecDeque::with_capacity(capacity) allocates room for every item at the start, so the deque never grows later. A capacity of 0 is rejected, because nothing could ever be pushed.

Listing 17.4 push (lines 42 to 76).

impl<T> BoundedQueue<T> {
    // ...
    fn push(&self, item: T) {
        // Acquire the shelf lock. `guard` is our exclusive handle to the deque.
        let mut guard = self.inner.lock().unwrap();

        // Wait while the shelf is full.
        //
        // Why `while` and not `if`:
        //   1. Condvars are allowed to wake up for no reason (spurious wakeup).
        //   2. Even on a real wakeup, another producer may have raced us and
        //      taken the free slot before we reacquired the lock.
        // So we must re-check the condition every time we wake.
        //
        // Why `guard = ...wait(guard)`:
        //   `wait` consumes the guard (it releases the lock on our behalf),
        //   puts us to sleep, then re-locks and hands us a NEW guard.
        //   The reassignment just means "here's my fresh key to the shelf."
        //   The VecDeque inside is the same one — we're not replacing the queue.
        while guard.len() == self.capacity {
            guard = self.not_full.wait(guard).unwrap();
        }

        // We have room. Add the item to the back of the deque.
        guard.push_back(item);

        // Release the lock BEFORE ringing the bell.
        // This is an optimization, not a correctness requirement.
        // If we notify while still holding the lock, the thread we wake will
        // immediately block on `lock()` until we release it — a wasted wakeup.
        drop(guard);

        // Ring the "not empty" bell: one consumer can now make progress.
        // `notify_one` (not `notify_all`) because exactly ONE item was added,
        // so only ONE consumer can succeed. Waking all of them = stampede.
        self.not_empty.notify_one();
    }
    // ...
}

The comments in push explain each decision. Figure 17.3 puts the steps of a blocked push in order.

The line guard = self.not_full.wait(guard).unwrap(); looks odd at first. wait takes the guard by value, because it must release the mutex. When the thread wakes, wait locks the mutex again and returns a new guard. The assignment stores that new guard in the same variable.

After the push, the code drops the guard before calling notify_one. If it notified first, the woken consumer would try to lock a mutex that is still held, and wait again at once. Unlocking first avoids that extra wait. The program would be correct either way.

Listing 17.5 pop (lines 78 to 102).

impl<T> BoundedQueue<T> {
    // ...
    fn pop(&self) -> T {
        // Acquire the shelf lock.
        let mut guard = self.inner.lock().unwrap();

        // Wait while the shelf is empty.
        // Same `while` reasoning as push: spurious wakeups + races with other
        // consumers who may have grabbed the item before we reacquired the lock.
        while guard.is_empty() {
            guard = self.not_empty.wait(guard).unwrap();
        }

        // Safe to unwrap: the `while` loop guarantees the deque is non-empty,
        // and we hold the lock so nobody can change that between the check
        // and this line.
        let item = guard.pop_front().unwrap();

        // Release before notifying (same optimization as in push).
        drop(guard);

        // Ring the "not full" bell: one producer can now make progress.
        // Exactly one slot opened, so wake exactly one producer.
        self.not_full.notify_one();

        item
    }
}

pop mirrors push. It waits on not_empty while the queue is empty, removes the front item, unlocks, and wakes one producer through not_full. The unwrap() on pop_front() cannot fail. The loop ended because the queue was not empty, and the thread has held the lock since that check.

Listing 17.6 main (lines 105 to 150). src/bin/bounded_buffer.rs

fn main() {
    // Small capacity (3) so that with two slow consumers you can actually
    // SEE producers block: the queue fills up and push() sleeps in `not_full`.
    let queue = Arc::new(BoundedQueue::new(3));

    let mut handles = vec![];

    // ---- Producers ----
    // Two producers, each pushes 5 items. `Arc::clone` bumps the refcount so
    // each thread shares the same queue; ownership of the clone moves in.
    for p in 0..2 {
        let q = Arc::clone(&queue);
        handles.push(thread::spawn(move || {
            for i in 0..5 {
                let item = format!("p{}:item{}", p, i);
                println!("  [producer {}] pushing {}", p, item);
                q.push(item);
                // Slow down so we don't blast through the whole buffer instantly.
                thread::sleep(Duration::from_millis(30));
            }
        }));
    }

    // ---- Consumers ----
    // Two consumers, each pops 5 items. Consumers are SLOWER than producers
    // here (60ms vs 30ms), so the queue fills up and you'll see producers
    // stall on `not_full.wait(...)`.
    for c in 0..2 {
        let q = Arc::clone(&queue);
        handles.push(thread::spawn(move || {
            for _ in 0..5 {
                let item = q.pop();
                println!("[consumer {}] got {}", c, item);
                thread::sleep(Duration::from_millis(60));
            }
        }));
    }

    // Join everything. If our synchronization is wrong, this is where you'd
    // hang (a thread sleeping forever) or panic (unwrap on empty / on a
    // poisoned mutex after another thread panicked while holding it).
    for h in handles {
        h.join().unwrap();
    }
    println!("done");
}

main starts two producers and two consumers on a queue with room for 3 items. Producers sleep 30 ms after each push, and consumers sleep 60 ms after each pop, so the producers are faster.

$ cargo run --bin bounded_buffer
  [producer 0] pushing p0:item0
  [producer 1] pushing p1:item0
[consumer 1] got p0:item0
[consumer 0] got p1:item0
  [producer 0] pushing p0:item1
  [producer 1] pushing p1:item1
  [producer 0] pushing p0:item2
  [producer 1] pushing p1:item2
[consumer 1] got p0:item1
[consumer 0] got p1:item1
  [producer 0] pushing p0:item3
  [producer 1] pushing p1:item3
[consumer 1] got p0:item2
[consumer 0] got p1:item2
  [producer 0] pushing p0:item4
  [producer 1] pushing p1:item4
[consumer 1] got p1:item3
[consumer 0] got p0:item3
[consumer 1] got p0:item4
[consumer 0] got p1:item4
done

The order changes from run to run. Items from one producer always come out in that producer’s order. The “pushing” line is printed before push is called, so the output does not show how long a push waited. Exercise 1 adds that.

17.4 When the lock is poisoned

bounded_buffer.rs calls .unwrap() on every lock() and wait(). If a thread panics while holding the lock, the mutex is poisoned, as section 16.6 showed. Every other thread’s unwrap() then panics too. Two more versions return an error instead (figure 17.4).

17.4.1 A custom error type

Listing 17.7 The error and an extension trait (lines 9 to 31). src/bin/bounded_buffer_with_error_handling.rs

use std::{
    collections::VecDeque,
    fmt,
    sync::{Arc, Condvar, Mutex, PoisonError},
    thread,
    time::Duration,
};

#[derive(Debug)]
pub struct QueuePoisonedError;

impl fmt::Display for QueuePoisonedError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "The queue's internal mutex was poisoned by a panicking thread"
        )
    }
}

impl std::error::Error for QueuePoisonedError {}

trait PoisonMap<T> {
    fn map_poison(self) -> Result<T, QueuePoisonedError>;
}

impl<T> PoisonMap<T> for Result<T, PoisonError<T>> {
    fn map_poison(self) -> Result<T, QueuePoisonedError> {
        self.map_err(|_| QueuePoisonedError)
    }
}

QueuePoisonedError is a unit struct: it carries no data. It implements Display for a readable message, and std::error::Error so it works with ? and Box<dyn Error>.

PoisonMap adds a method to a type the code does not own. A trait defined in your crate can be implemented for any type, including the standard library’s Result. A trait used this way is called an extension trait. impl<T> PoisonMap<T> for Result<T, PoisonError<T>> gives every Result<T, PoisonError<T>> a map_poison method, which replaces the PoisonError with QueuePoisonedError.

Listing 17.8 push and pop (lines 51 to 74).

impl<T> BoundedQueue<T> {
    // ...
    fn push(&self, item: T) -> Result<(), QueuePoisonedError> {
        let mut guard = self.inner.lock().map_poison()?;
        while guard.len() == self.capacity {
            guard = self.not_full.wait(guard).map_poison()?;
        }
        guard.push_back(item);
        drop(guard);
        self.not_empty.notify_one();
        Ok(())
    }

    fn pop(&self) -> Result<T, QueuePoisonedError> {
        let mut guard = self.inner.lock().map_poison()?;
        while guard.is_empty() {
            guard = self
                .not_empty
                .wait(guard)
                .map_poison()?;
        }
        let item = guard.pop_front().unwrap();
        drop(guard);
        self.not_full.notify_one();
        Ok(item)
    }
}

Each lock() and wait() call ends in .map_poison()?. The logic is the same as listing 17.4, and each method now returns a Result.

The complete program is listing 17.19 at the end of the chapter.

The threads in main handle an error by printing it with dbg! and leaving their loop with break.

The second test poisons the queue on purpose. It spawns a thread that locks the inner mutex and panics while holding it. A test in the same file can reach the private inner field. After that, both push and pop must return Err(QueuePoisonedError).

$ cargo test --bin bounded_buffer_with_error_handling
running 2 tests
test tests::test_zero_capacity_panics - should panic ... ok
test tests::test_queue_poisoning_on_thread_panic ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out

17.4.2 A boxed error

The third version returns a general error type:

use std::{
    collections::VecDeque,
    error::Error,
    sync::{Arc, Condvar, Mutex},
    thread,
    time::Duration,
};

type RuntimeError = Box<dyn Error + Send + Sync>;

Box<dyn Error + Send + Sync> can hold any error type. + Send + Sync is needed because the error travels between threads: each thread returns a Result, and join hands it to the main thread.

Listing 17.9 push (lines 29 to 46). src/bin/bounded_buffer_error_propagation.rs

impl<T> BoundedQueue<T> {
    // ...
    fn push(&self, item: T) -> Result<(), RuntimeError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|e| e.to_string())?;

        while guard.len() == self.capacity {
            guard = self
                .not_full
                .wait(guard)
                .map_err(|e| e.to_string())?;
        }

        guard.push_back(item);
        drop(guard);
        self.not_empty.notify_one();
        Ok(())
    }
    // ...
}

Why .map_err(|e| e.to_string()) instead of ? directly? A PoisonError contains the guard, and the guard borrows the mutex. A Box<dyn Error> must not borrow anything local, so the poison error cannot go into one. to_string() turns it into a String, which owns its text. Then ? converts the String into the boxed error, through a From implementation in the standard library.

Listing 17.10 main (lines 72 to 111). src/bin/bounded_buffer_error_propagation.rs

fn main() -> Result<(), RuntimeError> {
    let queue = Arc::new(BoundedQueue::new(3));

    let producers = (0..2).map(|p| {
        let q = Arc::clone(&queue);
        thread::spawn(move || -> Result<(), RuntimeError> {
            for i in 0..5 {
                let item = format!("p{}:item{}", p, i);
                println!("  [producer {}] pushing {}", p, item);
                q.push(item)?;
                thread::sleep(Duration::from_millis(30));
            }
            Ok(())
        })
    });

    let consumers = (0..2).map(|c| {
        let q = Arc::clone(&queue);
        thread::spawn(move || -> Result<(), RuntimeError> {
            for _ in 0..5 {
                let item = q.pop()?;
                println!("[consumer {}] got {}", c, item);
                thread::sleep(Duration::from_millis(60));
            }
            Ok(())
        })
    });

    let handles: Vec<_> = producers.chain(consumers).collect();

    for h in handles {
        match h.join() {
            Ok(r) => println!("Thread joined : {:#?}", r),
            Err(e) => eprintln!("Error occured : {:#?}", e),
        }
    }

    println!("done");
    Ok(())
}

main builds the threads with two iterator chains, producers and consumers. These are lazy: no thread starts until something consumes them. producers.chain(consumers).collect() starts all four threads while building the Vec, and only then does the loop join them.

Each thread’s closure is written move || -> Result<(), RuntimeError> { ... }, with its return type, so the ? inside it knows what error type to convert to. join() returns Ok(r), where r is the thread’s own Result.

$ cargo run --bin bounded_buffer_error_propagation
  [producer 1] pushing p1:item0
  [producer 0] pushing p0:item0
[consumer 1] got p1:item0
[consumer 0] got p0:item0
  [producer 0] pushing p0:item1
  [producer 1] pushing p1:item1
[consumer 1] got p0:item1
[consumer 0] got p1:item1
  [producer 0] pushing p0:item2
  [producer 1] pushing p1:item2
  [producer 0] pushing p0:item3
  [producer 1] pushing p1:item3
[consumer 0] got p1:item2
[consumer 1] got p0:item2
  [producer 0] pushing p0:item4
  [producer 1] pushing p1:item4
Thread joined : Ok(
    (),
)
[consumer 0] got p1:item3
[consumer 1] got p0:item3
Thread joined : Ok(
    (),
)
[consumer 1] got p0:item4
[consumer 0] got p1:item4
Thread joined : Ok(
    (),
)
Thread joined : Ok(
    (),
)
done

The join messages appear between the consumer lines, because main joins the producers first, while the consumers are still running.

17.5 Many producers and consumers with wait_while

The next version is the bounded buffer written more compactly, with wait_while for both waits. Its main uses scoped threads, so no Arc is needed.

The type is the same as in section 17.3. It holds a VecDeque behind a Mutex, a capacity, and one Condvar for each thing a thread can wait for:

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

pub struct BoundedBuffer<T> {
    queue: Mutex<VecDeque<T>>,
    capacity: usize,
    not_empty: Condvar,
    not_full: Condvar,
}

impl<T> BoundedBuffer<T> {
    pub fn new(capacity: usize) -> Self {
        assert!(capacity > 0, "capacity must be > 0");

        Self {
            queue: Mutex::new(VecDeque::with_capacity(capacity)),
            capacity,
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }
    // ...
}

push waits while the queue is full. wait_while takes the guard and a condition, and loops for you. It sleeps while the condition is true, and returns the guard once it is false. The while loop of section 17.3 is inside it, so a spurious wakeup is handled the same way.

impl<T> BoundedBuffer<T> {
    // ...
    /// Blocks while the buffer is full.
    pub fn push(&self, item: T) {
        let cap = self.capacity;

        let mut queue = self.queue.lock().unwrap();
        queue = self
            .not_full
            .wait_while(queue, |q| q.len() >= cap)
            .unwrap();

        queue.push_back(item);
        drop(queue);

        self.not_empty.notify_one();
    }
    // ...
}

push reads self.capacity into a local cap before the closure |q| q.len() >= cap. The closure then captures a usize instead of borrowing self. After pushing, drop(queue) releases the lock before notify_one, so the woken consumer does not wake only to wait for the lock.

pop is the mirror image: it waits while the queue is empty, takes the front item, and wakes one producer.

impl<T> BoundedBuffer<T> {
    // ...
    /// Blocks while the buffer is empty.
    pub fn pop(&self) -> T {
        let mut queue = self.queue.lock().unwrap();
        queue = self
            .not_empty
            .wait_while(queue, |q| q.is_empty())
            .unwrap();

        let item = queue.pop_front().unwrap();
        self.not_full.notify_one();
        item
    }
}

MPMC stands for multiple producers, multiple consumers. In main, four producers push 10 items each, and two consumers pop 20 each:

fn main() {
    let buffer = BoundedBuffer::<(usize, usize)>::new(8);
    let buffer = &buffer;

    const PRODUCERS: usize = 4;
    const CONSUMERS: usize = 2;
    const PER_PRODUCER: usize = 10;
    const PER_CONSUMER: usize = PRODUCERS * PER_PRODUCER / CONSUMERS;

    thread::scope(|s| {
        for p in 0..PRODUCERS {
            s.spawn(move || {
                for i in 0..PER_PRODUCER {
                    buffer.push((p, i));
                }
            });
        }

        for c in 0..CONSUMERS {
            s.spawn(move || {
                for _ in 0..PER_CONSUMER {
                    let item = buffer.pop();
                    println!("consumer {c}: {item:?}");
                }
            });
        }
    });
}

PER_CONSUMER is computed from the other constants, so every pushed item is popped and every thread finishes.

The complete program is listing 17.21 at the end of the chapter.

let buffer = &buffer; replaces the buffer with a reference to it. Each move closure then copies the reference into its thread. The scope guarantees that the buffer outlives all the threads.

$ cargo run --bin mpmc_bounded_buffer
consumer 1: (3, 0)
consumer 1: (2, 0)
consumer 0: (1, 0)
consumer 0: (3, 2)
consumer 0: (3, 3)
...
consumer 0: (3, 8)
consumer 0: (0, 9)
consumer 0: (2, 9)

The full output has 40 lines. Each item is a (producer, index) pair. Across the whole output, each producer’s indexes appear in increasing order, but the producers are mixed together. Which consumer prints an item depends on which one the operating system happened to wake.

17.6 Closing a queue

Every version so far has one problem. A consumer loop needs to know when to stop. The demos stop after a fixed count, but a real consumer does not know how many items will come. It calls pop in a loop, and when the producers finish, pop waits forever.

The library version adds a way to close the queue (figure 17.5).

Listing 17.11 The state (lines 16 to 27). src/problems/bounded_queue.rs

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

struct State<T> {
    items: VecDeque<T>,
    capacity: usize,
    closed: bool,
}

/// A bounded MPMC queue. Share it behind `Arc`; all methods take `&self`.
pub struct BoundedQueue<T> {
    state: Mutex<State<T>>,
    not_empty: Condvar,
    not_full: Condvar,
}

The mutex now protects a small struct: the items, the capacity, and a closed flag. The flag must be inside the mutex, because threads check it together with the items.

Listing 17.12 push and pop (lines 47 to 94).

impl<T> BoundedQueue<T> {
    // ...
    /// Push an item, blocking while the queue is full.
    ///
    /// Returns `Err(item)` if the queue is closed, so the caller keeps ownership
    /// instead of losing the value.
    pub fn push(&self, item: T) -> Result<(), T> {
        let mut state = self
            .state
            .lock()
            .expect("queue mutex poisoned");
        loop {
            if state.closed {
                return Err(item);
            }
            if state.items.len() < state.capacity {
                state.items.push_back(item);
                self.not_empty.notify_one();
                return Ok(());
            }
            state = self
                .not_full
                .wait(state)
                .expect("queue mutex poisoned");
        }
    }

    /// Pop an item, blocking while the queue is empty.
    ///
    /// Returns `None` once the queue is closed and drained, which is how a
    /// consumer loop terminates cleanly.
    pub fn pop(&self) -> Option<T> {
        let mut state = self
            .state
            .lock()
            .expect("queue mutex poisoned");
        loop {
            if let Some(item) = state.items.pop_front() {
                self.not_full.notify_one();
                return Some(item);
            }
            if state.closed {
                return None;
            }
            state = self
                .not_empty
                .wait(state)
                .expect("queue mutex poisoned");
        }
    }
    // ...
}

Both methods use a loop that checks every condition after each wakeup.

push returns Result<(), T>. If the queue is closed, it returns Err(item), handing the item back. The value is not lost, and the caller decides what to do with it.

pop returns Option<T>. It checks for an item before it checks closed, so a closed queue still gives up the items already in it. Only a closed queue that is empty returns None. A consumer can therefore loop with while let Some(item) = queue.pop(), and the loop ends after the last item.

Listing 17.13 close (lines 109 to 122).

impl<T> BoundedQueue<T> {
    // ...
    /// Close the queue.
    ///
    /// Pending and future `pop` calls return `None` after the queue drains, and
    /// future `push` calls fail. Idempotent.
    pub fn close(&self) {
        let mut state = self
            .state
            .lock()
            .expect("queue mutex poisoned");
        state.closed = true;
        drop(state);
        self.not_empty.notify_all();
        self.not_full.notify_all();
    }
    // ...
}

close sets the flag and wakes every waiting thread on both condition variables with notify_all. Each woken thread sees closed in its loop and returns. Calling close twice does no harm. An operation that gives the same result when repeated is called idempotent.

Animation 17.3 closes a full queue while a producer waits in push. Then it closes a queue without waking anyone.

The complete file, with its tests, is listing 17.22 at the end of the chapter.

The tests check the four promises:

• preserves_order_across_threads pushes 0 to 999 from one thread and pops them on another. They must arrive in order, and the consumer’s while let loop must end after close.
• never_exceeds_capacity records the largest length seen, with a slow consumer. It must never pass 4.
• close_wakes_a_blocked_consumer starts a consumer on an empty queue, waits 20 ms so it is asleep, and closes the queue. The consumer must return None instead of sleeping forever.
• push_after_close_returns_the_item checks that push hands the item back.

$ cargo test --lib problems::bounded_queue
running 4 tests
test problems::bounded_queue::tests::push_after_close_returns_the_item ... ok
test problems::bounded_queue::tests::preserves_order_across_threads ... ok
test problems::bounded_queue::tests::never_exceeds_capacity ... ok
test problems::bounded_queue::tests::close_wakes_a_blocked_consumer ... ok

test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 133 filtered out

17.7 A ring buffer without a lock

Every queue so far takes a lock for each operation. With exactly one producer thread and one consumer thread, the lock can go. This case is called SPSC, for single producer, single consumer.

17.7.1 The idea

A ring buffer is a fixed array used in a circle (figure 17.6). Two counters track it:

• head counts the items taken so far. The consumer reads the next item at head % N.
• tail counts the items added so far. The producer writes the next item at tail % N.

The number of items in the ring is tail - head. The ring is empty when the two are equal, and full when the difference is N.

No lock is needed because each counter has one writer. Only the producer changes tail, and only the consumer changes head. Each side only reads the other’s counter.

The ordering of each store makes the handoff safe. The producer writes the item into its slot first, and then stores the new tail with Release. The consumer loads tail with Acquire before it reads the slot. This is the release and acquire pairing from section 16.4. If the consumer sees the new tail, it also sees the item written before it.

17.7.2 The type

Listing 17.14 The ring (lines 25 to 38). src/problems/ring_buffer.rs

use std::{
    cell::UnsafeCell,
    mem::MaybeUninit,
    sync::atomic::{AtomicUsize, Ordering},
};

/// Ring buffer with capacity `N` (the maximum number of queued items).
pub struct SpscRing<T, const N: usize> {
    slots: [UnsafeCell<MaybeUninit<T>>; N],
    /// Next index to read. Written only by the consumer.
    head: AtomicUsize,
    /// Next index to write. Written only by the producer.
    tail: AtomicUsize,
}

// SAFETY: at most one thread mutates `head` and one mutates `tail`. A slot is
// initialized before `tail` is published with a release store, and the consumer
// performs an acquire load of `tail` before reading that slot, so there is no
// data race and no read of uninitialized memory under the SPSC discipline.
unsafe impl<T: Send, const N: usize> Sync for SpscRing<T, N> {}

Three new pieces of Rust appear here:

• const N: usize is a const generic: a type parameter that is a number instead of a type. SpscRing<u64, 256> is a ring of 256 slots, and the array has that length at compile time.
• MaybeUninit<T> is memory that may or may not hold a valid T. An empty slot holds no value, and Rust does not allow a variable of type T to hold garbage. MaybeUninit makes the uninitialized state explicit. Reading it as a T is unsafe, because the code must know that the slot holds a value.
• UnsafeCell lets the producer write a slot through &self, as in the spin lock of section 16.4.

new builds the array with std::array::from_fn, which calls a closure once per index.

17.7.3 Push and pop

Listing 17.15 push and pop (lines 71 to 107).

impl<T, const N: usize> SpscRing<T, N> {
    // ...
    /// Enqueue a value, returning `Err(value)` when the ring is full.
    pub fn push(&self, value: T) -> Result<(), T> {
        let tail = self.tail.load(Ordering::Relaxed);
        let head = self.head.load(Ordering::Acquire);
        if tail.wrapping_sub(head) >= N {
            return Err(value);
        }

        let index = tail % N;
        // SAFETY: the slot at `index` is free because the ring is not full, and
        // only the producer writes it.
        unsafe { (*self.slots[index].get()).write(value) };

        // Publish the slot to the consumer.
        self.tail
            .store(tail.wrapping_add(1), Ordering::Release);
        Ok(())
    }

    /// Dequeue a value, or `None` when the ring is empty.
    pub fn pop(&self) -> Option<T> {
        let head = self.head.load(Ordering::Relaxed);
        let tail = self.tail.load(Ordering::Acquire);
        if head == tail {
            return None;
        }

        let index = head % N;
        // SAFETY: `head != tail`, so the producer has published this slot and
        // only the consumer reads it.
        let value = unsafe { (*self.slots[index].get()).assume_init_read() };

        // Publish the freed slot to the producer.
        self.head
            .store(head.wrapping_add(1), Ordering::Release);
        Some(value)
    }
}

push loads its own counter, tail, with Relaxed, because only this thread writes it. It loads the other side’s counter, head, with Acquire. That load pairs with the consumer’s Release store of head, so the consumer has finished reading a slot before the producer writes it again.

If the ring is full, push returns Err(value), handing the value back. Otherwise it writes the value into the slot with MaybeUninit::write, and publishes it by storing tail + 1 with Release.

pop is the mirror image. It loads tail with Acquire, reads the value out of the slot with assume_init_read, and publishes the free slot by storing head + 1 with Release.

The counters use wrapping_add and wrapping_sub. After 2 to the power 64 operations a counter would wrap around to 0, and wrapping arithmetic keeps tail - head correct across that point.

Animation 17.4 runs the ring with 8 slots. The producer pushes and the consumer pops, the counters grow past 8, and the ring fills. The last part stores tail before writing the slot.

Listing 17.16 Drop (lines 116 to 121).

impl<T, const N: usize> Drop for SpscRing<T, N> {
    fn drop(&mut self) {
        // Drop any items that were never popped.
        while self.pop().is_some() {}
    }
}

MaybeUninit never drops its contents, because it cannot know whether a value is there. So when the ring is dropped, its Drop pops every remaining item. Each popped value is a normal T, and it is dropped at the end of the loop body. The test dropped_ring_drops_queued_items counts those drops.

The complete file, with its tests, is listing 17.23 at the end of the chapter.

The concurrent test sends 50,000 values through a ring of 256 slots. When the ring is full, the producer calls thread::yield_now() and tries again. When it is empty, the consumer does the same. The producer ends with a sentinel value, u64::MAX, which tells the consumer to stop. The consumer must receive every value, in order.

$ cargo test --lib problems::ring_buffer
running 4 tests
test problems::ring_buffer::tests::is_fifo_within_capacity ... ok
test problems::ring_buffer::tests::full_ring_returns_the_value ... ok
test problems::ring_buffer::tests::dropped_ring_drops_queued_items ... ok
test problems::ring_buffer::tests::producer_and_consumer_run_concurrently_in_order ... ok

test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 133 filtered out

17.8 The complete files

Each file below is shown whole, in the order the chapter used it.

Listing 17.17 The shared queue and its main. src/bin/thread_safe_queue.rs

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

#[derive(Default)]
struct SyncQueue {
    items: Mutex<VecDeque<i32>>,
    not_empty: Condvar,
}

impl SyncQueue {
    fn push(&self, item: i32) {
        if let Ok(mut q) = self.items.lock() {
            q.push_back(item);
            self.not_empty.notify_one();
        }
    }

    fn pop(&self) -> Option<i32> {
        let q = self.items.lock().ok()?;
        let mut q = self
            .not_empty
            .wait_while(q, |q| q.is_empty())
            .ok()?;
        q.pop_front()
    }
}

fn main() {
    let q = Arc::new(SyncQueue::default());

    // producer
    let producer = {
        let q = Arc::clone(&q);
        thread::spawn(move || {
            for i in 0..5 {
                q.push(i);
                println!("Producer pushed {i}");
            }
        })
    };

    let consumer = {
        let q = Arc::clone(&q);
        thread::spawn(move || {
            for _ in 0..5 {
                println!("Cosnumer popped: {:?}", q.pop());
            }
        })
    };

    producer.join().unwrap();
    consumer.join().unwrap();
}

Listing 17.18 The bounded buffer and its main. src/bin/bounded_buffer.rs

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

// A thread-safe bounded buffer (a "shelf" that holds at most `capacity` items).
//
// Three fields do all the work:
//   - `inner`     : the actual VecDeque, wrapped in a Mutex so only one thread
//                   touches it at a time.
//   - `not_empty` : waiting room for CONSUMERS. A consumer sleeps here when the
//                   shelf is empty. Producers ring this bell after pushing.
//   - `not_full`  : waiting room for PRODUCERS. A producer sleeps here when the
//                   shelf is full. Consumers ring this bell after popping.
//
// Two separate condvars so we always wake the RIGHT kind of thread.
// One shared condvar would cause livelock: consumers waking consumers,
// producers waking producers, everyone re-checking and going back to sleep.
struct BoundedQueue<T> {
    inner: Mutex<VecDeque<T>>,
    capacity: usize,
    not_empty: Condvar,
    not_full: Condvar,
}

impl<T> BoundedQueue<T> {
    fn new(capacity: usize) -> Self {
        // Capacity 0 makes no sense: no producer could ever push,
        // and every pop would block forever.
        assert!(capacity > 0, "capacity must be > 0");

        Self {
            inner: Mutex::new(VecDeque::with_capacity(capacity)),
            capacity,
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }

    fn push(&self, item: T) {
        // Acquire the shelf lock. `guard` is our exclusive handle to the deque.
        let mut guard = self.inner.lock().unwrap();

        // Wait while the shelf is full.
        //
        // Why `while` and not `if`:
        //   1. Condvars are allowed to wake up for no reason (spurious wakeup).
        //   2. Even on a real wakeup, another producer may have raced us and
        //      taken the free slot before we reacquired the lock.
        // So we must re-check the condition every time we wake.
        //
        // Why `guard = ...wait(guard)`:
        //   `wait` consumes the guard (it releases the lock on our behalf),
        //   puts us to sleep, then re-locks and hands us a NEW guard.
        //   The reassignment just means "here's my fresh key to the shelf."
        //   The VecDeque inside is the same one — we're not replacing the queue.
        while guard.len() == self.capacity {
            guard = self.not_full.wait(guard).unwrap();
        }

        // We have room. Add the item to the back of the deque.
        guard.push_back(item);

        // Release the lock BEFORE ringing the bell.
        // This is an optimization, not a correctness requirement.
        // If we notify while still holding the lock, the thread we wake will
        // immediately block on `lock()` until we release it — a wasted wakeup.
        drop(guard);

        // Ring the "not empty" bell: one consumer can now make progress.
        // `notify_one` (not `notify_all`) because exactly ONE item was added,
        // so only ONE consumer can succeed. Waking all of them = stampede.
        self.not_empty.notify_one();
    }

    fn pop(&self) -> T {
        // Acquire the shelf lock.
        let mut guard = self.inner.lock().unwrap();

        // Wait while the shelf is empty.
        // Same `while` reasoning as push: spurious wakeups + races with other
        // consumers who may have grabbed the item before we reacquired the lock.
        while guard.is_empty() {
            guard = self.not_empty.wait(guard).unwrap();
        }

        // Safe to unwrap: the `while` loop guarantees the deque is non-empty,
        // and we hold the lock so nobody can change that between the check
        // and this line.
        let item = guard.pop_front().unwrap();

        // Release before notifying (same optimization as in push).
        drop(guard);

        // Ring the "not full" bell: one producer can now make progress.
        // Exactly one slot opened, so wake exactly one producer.
        self.not_full.notify_one();

        item
    }
}

fn main() {
    // Small capacity (3) so that with two slow consumers you can actually
    // SEE producers block: the queue fills up and push() sleeps in `not_full`.
    let queue = Arc::new(BoundedQueue::new(3));

    let mut handles = vec![];

    // ---- Producers ----
    // Two producers, each pushes 5 items. `Arc::clone` bumps the refcount so
    // each thread shares the same queue; ownership of the clone moves in.
    for p in 0..2 {
        let q = Arc::clone(&queue);
        handles.push(thread::spawn(move || {
            for i in 0..5 {
                let item = format!("p{}:item{}", p, i);
                println!("  [producer {}] pushing {}", p, item);
                q.push(item);
                // Slow down so we don't blast through the whole buffer instantly.
                thread::sleep(Duration::from_millis(30));
            }
        }));
    }

    // ---- Consumers ----
    // Two consumers, each pops 5 items. Consumers are SLOWER than producers
    // here (60ms vs 30ms), so the queue fills up and you'll see producers
    // stall on `not_full.wait(...)`.
    for c in 0..2 {
        let q = Arc::clone(&queue);
        handles.push(thread::spawn(move || {
            for _ in 0..5 {
                let item = q.pop();
                println!("[consumer {}] got {}", c, item);
                thread::sleep(Duration::from_millis(60));
            }
        }));
    }

    // Join everything. If our synchronization is wrong, this is where you'd
    // hang (a thread sleeping forever) or panic (unwrap on empty / on a
    // poisoned mutex after another thread panicked while holding it).
    for h in handles {
        h.join().unwrap();
    }
    println!("done");
}

Listing 17.19 The bounded buffer with a custom error type, with tests. src/bin/bounded_buffer_with_error_handling.rs

use std::{
    collections::VecDeque,
    fmt,
    sync::{Arc, Condvar, Mutex, PoisonError},
    thread,
    time::Duration,
};

#[derive(Debug)]
pub struct QueuePoisonedError;

impl fmt::Display for QueuePoisonedError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "The queue's internal mutex was poisoned by a panicking thread"
        )
    }
}

impl std::error::Error for QueuePoisonedError {}

trait PoisonMap<T> {
    fn map_poison(self) -> Result<T, QueuePoisonedError>;
}

impl<T> PoisonMap<T> for Result<T, PoisonError<T>> {
    fn map_poison(self) -> Result<T, QueuePoisonedError> {
        self.map_err(|_| QueuePoisonedError)
    }
}

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

impl<T> BoundedQueue<T> {
    fn new(capacity: usize) -> Self {
        assert!(capacity > 0, "capacity must be > 0");
        Self {
            inner: Mutex::new(VecDeque::with_capacity(capacity)),
            capacity,
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }

    fn push(&self, item: T) -> Result<(), QueuePoisonedError> {
        let mut guard = self.inner.lock().map_poison()?;
        while guard.len() == self.capacity {
            guard = self.not_full.wait(guard).map_poison()?;
        }
        guard.push_back(item);
        drop(guard);
        self.not_empty.notify_one();
        Ok(())
    }

    fn pop(&self) -> Result<T, QueuePoisonedError> {
        let mut guard = self.inner.lock().map_poison()?;
        while guard.is_empty() {
            guard = self
                .not_empty
                .wait(guard)
                .map_poison()?;
        }
        let item = guard.pop_front().unwrap();
        drop(guard);
        self.not_full.notify_one();
        Ok(item)
    }
}

fn main() {
    let queue = Arc::new(BoundedQueue::new(3));
    let mut handles = vec![];

    // ---- Producers ----
    for p in 0..2 {
        let q = Arc::clone(&queue);
        handles.push(thread::spawn(move || {
            for i in 0..5 {
                let item = format!("p{}:item{}", p, i);
                println!("  [producer {}] pushing {}", p, item);

                // Uses match, avoids naming an error variable, uses dbg!
                match q.push(item) {
                    Ok(_) => {}
                    Err(err) => {
                        dbg!(err);
                        break;
                    }
                }
                thread::sleep(Duration::from_millis(30));
            }
        }));
    }

    // ---- Consumers ----
    for c in 0..2 {
        let q = Arc::clone(&queue);
        handles.push(thread::spawn(move || {
            for _ in 0..5 {
                // Uses match, avoids naming an error variable, uses dbg!
                match q.pop() {
                    Ok(item) => {
                        println!("[consumer {}] got {}", c, item);
                    }
                    Err(err) => {
                        dbg!(err);
                        break;
                    }
                }
                thread::sleep(Duration::from_millis(60));
            }
        }));
    }

    // Join everything.
    for h in handles {
        match h.join() {
            Ok(_) => {}
            Err(err) => {
                dbg!(err);
            }
        }
    }
    println!("done");
}

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

    // 1. Mark the test as one that MUST panic to pass
    // 2. Optional: use 'expected' to verify it panics with the correct error message
    #[test]
    #[should_panic(expected = "capacity must be > 0")]
    fn test_zero_capacity_panics() {
        // This line invokes the assert!(capacity > 0) inside new()
        let _queue: BoundedQueue<i32> = BoundedQueue::new(0);
    }

    #[test]
    fn test_queue_poisoning_on_thread_panic() {
        // 1. Create a queue shared via Arc so multiple threads can touch it
        let queue = Arc::new(BoundedQueue::new(5));
        let queue_clone = Arc::clone(&queue);

        // 2. Spawn a thread designed to crash while holding the internal lock
        let handle = thread::spawn(move || {
            // Manually lock the inner mutex so this thread owns the lock guard
            let _guard = queue_clone.inner.lock().unwrap();

            // Trigger a serious runtime panic while holding the lock guard!
            panic!("Intentional worker thread crash!");
        });

        // 3. Wait for the thread to die. It will return an Err because it panicked.
        let join_result = handle.join();
        assert!(join_result.is_err(), "The thread was supposed to panic");

        // 4. Now try to interact with the queue.
        // Because the thread panicked while holding the guard, the Mutex is poisoned.
        // Our push and pop methods should cleanly catch this and return our custom error.
        match queue.push("test_item") {
            Err(QueuePoisonedError) => {
                // Success! The queue correctly caught the poisoned state
                // instead of panicking the main thread.
            }
            Ok(_) => {
                panic!("Expected QueuePoisonedError, but push succeeded!");
            }
        }

        // Verify pop behaves the same way
        match queue.pop() {
            Err(QueuePoisonedError) => {}
            Ok(_) => panic!("Expected QueuePoisonedError, but pop succeeded!"),
        }
    }
}

Listing 17.20 The bounded buffer with a boxed error. src/bin/bounded_buffer_error_propagation.rs

use std::{
    collections::VecDeque,
    error::Error,
    sync::{Arc, Condvar, Mutex},
    thread,
    time::Duration,
};

type RuntimeError = Box<dyn Error + Send + Sync>;

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

impl<T> BoundedQueue<T> {
    fn new(capacity: usize) -> Self {
        assert!(capacity > 0, "capacity must be > 0");
        Self {
            inner: Mutex::new(VecDeque::with_capacity(capacity)),
            capacity,
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }

    fn push(&self, item: T) -> Result<(), RuntimeError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|e| e.to_string())?;

        while guard.len() == self.capacity {
            guard = self
                .not_full
                .wait(guard)
                .map_err(|e| e.to_string())?;
        }

        guard.push_back(item);
        drop(guard);
        self.not_empty.notify_one();
        Ok(())
    }

    fn pop(&self) -> Result<T, RuntimeError> {
        let mut guard = self
            .inner
            .lock()
            .map_err(|e| e.to_string())?;

        while guard.is_empty() {
            guard = self
                .not_empty
                .wait(guard)
                .map_err(|e| e.to_string())?;
        }

        let item = match guard.pop_front() {
            Some(item) => item,
            None => return Err("deque is non-empty by loop invariant".into()),
        };

        drop(guard);
        self.not_full.notify_one();
        Ok(item)
    }
}

fn main() -> Result<(), RuntimeError> {
    let queue = Arc::new(BoundedQueue::new(3));

    let producers = (0..2).map(|p| {
        let q = Arc::clone(&queue);
        thread::spawn(move || -> Result<(), RuntimeError> {
            for i in 0..5 {
                let item = format!("p{}:item{}", p, i);
                println!("  [producer {}] pushing {}", p, item);
                q.push(item)?;
                thread::sleep(Duration::from_millis(30));
            }
            Ok(())
        })
    });

    let consumers = (0..2).map(|c| {
        let q = Arc::clone(&queue);
        thread::spawn(move || -> Result<(), RuntimeError> {
            for _ in 0..5 {
                let item = q.pop()?;
                println!("[consumer {}] got {}", c, item);
                thread::sleep(Duration::from_millis(60));
            }
            Ok(())
        })
    });

    let handles: Vec<_> = producers.chain(consumers).collect();

    for h in handles {
        match h.join() {
            Ok(r) => println!("Thread joined : {:#?}", r),
            Err(e) => eprintln!("Error occured : {:#?}", e),
        }
    }

    println!("done");
    Ok(())
}

Listing 17.21 Many producers and consumers with wait_while. src/bin/mpmc_bounded_buffer.rs

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

pub struct BoundedBuffer<T> {
    queue: Mutex<VecDeque<T>>,
    capacity: usize,
    not_empty: Condvar,
    not_full: Condvar,
}

impl<T> BoundedBuffer<T> {
    pub fn new(capacity: usize) -> Self {
        assert!(capacity > 0, "capacity must be > 0");

        Self {
            queue: Mutex::new(VecDeque::with_capacity(capacity)),
            capacity,
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }

    /// Blocks while the buffer is full.
    pub fn push(&self, item: T) {
        let cap = self.capacity;

        let mut queue = self.queue.lock().unwrap();
        queue = self
            .not_full
            .wait_while(queue, |q| q.len() >= cap)
            .unwrap();

        queue.push_back(item);
        drop(queue);

        self.not_empty.notify_one();
    }

    /// Blocks while the buffer is empty.
    pub fn pop(&self) -> T {
        let mut queue = self.queue.lock().unwrap();
        queue = self
            .not_empty
            .wait_while(queue, |q| q.is_empty())
            .unwrap();

        let item = queue.pop_front().unwrap();
        self.not_full.notify_one();
        item
    }
}

fn main() {
    let buffer = BoundedBuffer::<(usize, usize)>::new(8);
    let buffer = &buffer;

    const PRODUCERS: usize = 4;
    const CONSUMERS: usize = 2;
    const PER_PRODUCER: usize = 10;
    const PER_CONSUMER: usize = PRODUCERS * PER_PRODUCER / CONSUMERS;

    thread::scope(|s| {
        for p in 0..PRODUCERS {
            s.spawn(move || {
                for i in 0..PER_PRODUCER {
                    buffer.push((p, i));
                }
            });
        }

        for c in 0..CONSUMERS {
            s.spawn(move || {
                for _ in 0..PER_CONSUMER {
                    let item = buffer.pop();
                    println!("consumer {c}: {item:?}");
                }
            });
        }
    });
}

Listing 17.22 The closable bounded queue, with tests. src/problems/bounded_queue.rs

//! Bounded blocking queue with backpressure.
//!
//! Unbounded queues are a classic production failure: a fast producer and a slow
//! consumer turn into an OOM kill instead of a slow response. A bounded queue makes
//! the producer wait, which is backpressure.
//!
//! This is an MPMC queue (many producers, many consumers) built from one `Mutex`
//! and two condition variables: `not_empty` for waiting consumers and `not_full`
//! for waiting producers. `close` unblocks everyone so shutdown cannot hang.

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

struct State<T> {
    items: VecDeque<T>,
    capacity: usize,
    closed: bool,
}

/// A bounded MPMC queue. Share it behind `Arc`; all methods take `&self`.
pub struct BoundedQueue<T> {
    state: Mutex<State<T>>,
    not_empty: Condvar,
    not_full: Condvar,
}

impl<T> BoundedQueue<T> {
    /// Create a queue that holds at most `capacity` items.
    ///
    /// Panics if `capacity` is zero, because a zero-capacity queue can never make
    /// progress.
    pub fn new(capacity: usize) -> Self {
        assert!(capacity > 0, "capacity must be > 0");
        Self {
            state: Mutex::new(State {
                items: VecDeque::with_capacity(capacity),
                capacity,
                closed: false,
            }),
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }

    /// Push an item, blocking while the queue is full.
    ///
    /// Returns `Err(item)` if the queue is closed, so the caller keeps ownership
    /// instead of losing the value.
    pub fn push(&self, item: T) -> Result<(), T> {
        let mut state = self
            .state
            .lock()
            .expect("queue mutex poisoned");
        loop {
            if state.closed {
                return Err(item);
            }
            if state.items.len() < state.capacity {
                state.items.push_back(item);
                self.not_empty.notify_one();
                return Ok(());
            }
            state = self
                .not_full
                .wait(state)
                .expect("queue mutex poisoned");
        }
    }

    /// Pop an item, blocking while the queue is empty.
    ///
    /// Returns `None` once the queue is closed and drained, which is how a
    /// consumer loop terminates cleanly.
    pub fn pop(&self) -> Option<T> {
        let mut state = self
            .state
            .lock()
            .expect("queue mutex poisoned");
        loop {
            if let Some(item) = state.items.pop_front() {
                self.not_full.notify_one();
                return Some(item);
            }
            if state.closed {
                return None;
            }
            state = self
                .not_empty
                .wait(state)
                .expect("queue mutex poisoned");
        }
    }

    /// Pop without blocking.
    pub fn try_pop(&self) -> Option<T> {
        let mut state = self
            .state
            .lock()
            .expect("queue mutex poisoned");
        let item = state.items.pop_front();
        if item.is_some() {
            self.not_full.notify_one();
        }
        item
    }

    /// Close the queue.
    ///
    /// Pending and future `pop` calls return `None` after the queue drains, and
    /// future `push` calls fail. Idempotent.
    pub fn close(&self) {
        let mut state = self
            .state
            .lock()
            .expect("queue mutex poisoned");
        state.closed = true;
        drop(state);
        self.not_empty.notify_all();
        self.not_full.notify_all();
    }

    pub fn is_closed(&self) -> bool {
        self.state
            .lock()
            .expect("queue mutex poisoned")
            .closed
    }

    pub fn capacity(&self) -> usize {
        self.state
            .lock()
            .expect("queue mutex poisoned")
            .capacity
    }

    pub fn len(&self) -> usize {
        self.state
            .lock()
            .expect("queue mutex poisoned")
            .items
            .len()
    }

    pub fn is_empty(&self) -> bool {
        self.state
            .lock()
            .expect("queue mutex poisoned")
            .items
            .is_empty()
    }
}

#[cfg(test)]
mod tests {
    use std::{
        sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
        },
        thread,
        time::Duration,
    };

    use super::*;

    #[test]
    fn preserves_order_across_threads() {
        let queue = Arc::new(BoundedQueue::new(8));
        let producer = {
            let queue = Arc::clone(&queue);
            thread::spawn(move || {
                for value in 0..1_000 {
                    queue
                        .push(value)
                        .expect("queue stays open");
                }
                queue.close();
            })
        };

        let mut received = Vec::new();
        while let Some(value) = queue.pop() {
            received.push(value);
        }
        producer
            .join()
            .expect("producer panicked");

        assert_eq!(received, (0..1_000).collect::<Vec<_>>());
    }

    #[test]
    fn never_exceeds_capacity() {
        let queue = Arc::new(BoundedQueue::new(4));
        let observed_peak = Arc::new(AtomicUsize::new(0));
        let producer = {
            let queue = Arc::clone(&queue);
            let observed_peak = Arc::clone(&observed_peak);
            thread::spawn(move || {
                for value in 0..200u32 {
                    queue
                        .push(value)
                        .expect("queue stays open");
                    observed_peak.fetch_max(queue.len(), Ordering::SeqCst);
                }
                queue.close();
            })
        };

        while queue.pop().is_some() {
            thread::sleep(Duration::from_micros(50));
        }
        producer
            .join()
            .expect("producer panicked");

        assert!(observed_peak.load(Ordering::SeqCst) <= 4);
    }

    #[test]
    fn close_wakes_a_blocked_consumer() {
        let queue = Arc::new(BoundedQueue::<u32>::new(1));
        let consumer = {
            let queue = Arc::clone(&queue);
            thread::spawn(move || queue.pop())
        };

        thread::sleep(Duration::from_millis(20));
        queue.close();

        assert_eq!(
            consumer
                .join()
                .expect("consumer panicked"),
            None
        );
    }

    #[test]
    fn push_after_close_returns_the_item() {
        let queue = BoundedQueue::new(1);
        queue.close();
        assert_eq!(
            queue.push("workload".to_string()),
            Err("workload".to_string())
        );
    }
}

Listing 17.23 The SPSC ring buffer, with tests. src/problems/ring_buffer.rs

//! Lock-free single-producer / single-consumer (SPSC) ring buffer.
//!
//! The classic producer/consumer exercise, but with no mutex: the producer owns
//! the `tail` index and the consumer owns the `head` index, and each publishes its
//! write with a release store so the other side sees the slot only after it is
//! initialized. This is the shape inside channel implementations, log shippers,
//! and kernel ring buffers.
//!
//! Two rules keep this sound:
//!
//! - exactly one thread calls [`SpscRing::push`] and exactly one calls
//!   [`SpscRing::pop`];
//! - a slot is written before `tail` is published, and read only after `head`
//!   reaches it.
//!
//! If you need many producers or consumers, use a mutex-backed queue such as
//! [`crate::problems::bounded_queue::BoundedQueue`].

use std::{
    cell::UnsafeCell,
    mem::MaybeUninit,
    sync::atomic::{AtomicUsize, Ordering},
};

/// Ring buffer with capacity `N` (the maximum number of queued items).
pub struct SpscRing<T, const N: usize> {
    slots: [UnsafeCell<MaybeUninit<T>>; N],
    /// Next index to read. Written only by the consumer.
    head: AtomicUsize,
    /// Next index to write. Written only by the producer.
    tail: AtomicUsize,
}

// SAFETY: at most one thread mutates `head` and one mutates `tail`. A slot is
// initialized before `tail` is published with a release store, and the consumer
// performs an acquire load of `tail` before reading that slot, so there is no
// data race and no read of uninitialized memory under the SPSC discipline.
unsafe impl<T: Send, const N: usize> Sync for SpscRing<T, N> {}

impl<T, const N: usize> SpscRing<T, N> {
    /// Create an empty ring. Panics if `N` is zero.
    pub fn new() -> Self {
        assert!(N > 0, "ring capacity must be > 0");
        Self {
            slots: std::array::from_fn(|_| UnsafeCell::new(MaybeUninit::uninit())),
            head: AtomicUsize::new(0),
            tail: AtomicUsize::new(0),
        }
    }

    /// Maximum number of items the ring can hold.
    pub const fn capacity(&self) -> usize {
        N
    }

    /// Number of queued items. Only meaningful to the owning side in a race.
    pub fn len(&self) -> usize {
        self.tail
            .load(Ordering::Acquire)
            .wrapping_sub(self.head.load(Ordering::Acquire))
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    pub fn is_full(&self) -> bool {
        self.len() >= N
    }

    /// Enqueue a value, returning `Err(value)` when the ring is full.
    pub fn push(&self, value: T) -> Result<(), T> {
        let tail = self.tail.load(Ordering::Relaxed);
        let head = self.head.load(Ordering::Acquire);
        if tail.wrapping_sub(head) >= N {
            return Err(value);
        }

        let index = tail % N;
        // SAFETY: the slot at `index` is free because the ring is not full, and
        // only the producer writes it.
        unsafe { (*self.slots[index].get()).write(value) };

        // Publish the slot to the consumer.
        self.tail
            .store(tail.wrapping_add(1), Ordering::Release);
        Ok(())
    }

    /// Dequeue a value, or `None` when the ring is empty.
    pub fn pop(&self) -> Option<T> {
        let head = self.head.load(Ordering::Relaxed);
        let tail = self.tail.load(Ordering::Acquire);
        if head == tail {
            return None;
        }

        let index = head % N;
        // SAFETY: `head != tail`, so the producer has published this slot and
        // only the consumer reads it.
        let value = unsafe { (*self.slots[index].get()).assume_init_read() };

        // Publish the freed slot to the producer.
        self.head
            .store(head.wrapping_add(1), Ordering::Release);
        Some(value)
    }
}

impl<T, const N: usize> Default for SpscRing<T, N> {
    fn default() -> Self {
        Self::new()
    }
}

impl<T, const N: usize> Drop for SpscRing<T, N> {
    fn drop(&mut self) {
        // Drop any items that were never popped.
        while self.pop().is_some() {}
    }
}

#[cfg(test)]
mod tests {
    use std::{
        sync::{Arc, atomic::AtomicUsize},
        thread,
    };

    use super::*;

    #[test]
    fn is_fifo_within_capacity() {
        let ring = SpscRing::<u32, 4>::new();
        assert!(ring.push(1).is_ok());
        assert!(ring.push(2).is_ok());
        assert!(ring.push(3).is_ok());
        assert_eq!(ring.pop(), Some(1));
        assert_eq!(ring.pop(), Some(2));
        assert_eq!(ring.pop(), Some(3));
        assert_eq!(ring.pop(), None);
    }

    #[test]
    fn full_ring_returns_the_value() {
        let ring = SpscRing::<u32, 2>::new();
        assert!(ring.push(1).is_ok());
        assert!(ring.push(2).is_ok());
        assert!(ring.is_full());
        assert_eq!(ring.push(3), Err(3));
    }

    #[test]
    fn producer_and_consumer_run_concurrently_in_order() {
        const ITEMS: u64 = 50_000;
        const SENTINEL: u64 = u64::MAX;

        let ring = Arc::new(SpscRing::<u64, 256>::new());

        let producer = {
            let ring = Arc::clone(&ring);
            thread::spawn(move || {
                for value in 0..ITEMS {
                    let mut pending = value;
                    loop {
                        match ring.push(pending) {
                            Ok(()) => break,
                            Err(returned) => {
                                pending = returned;
                                thread::yield_now();
                            }
                        }
                    }
                }
                let mut pending = SENTINEL;
                loop {
                    match ring.push(pending) {
                        Ok(()) => break,
                        Err(returned) => {
                            pending = returned;
                            thread::yield_now();
                        }
                    }
                }
            })
        };

        let consumer = {
            let ring = Arc::clone(&ring);
            thread::spawn(move || {
                let mut received = Vec::with_capacity(ITEMS as usize);
                loop {
                    match ring.pop() {
                        Some(SENTINEL) => break,
                        Some(value) => received.push(value),
                        None => thread::yield_now(),
                    }
                }
                received
            })
        };

        producer
            .join()
            .expect("producer panicked");
        let received = consumer
            .join()
            .expect("consumer panicked");

        assert_eq!(received, (0..ITEMS).collect::<Vec<_>>());
    }

    #[test]
    fn dropped_ring_drops_queued_items() {
        struct Tracker(Arc<AtomicUsize>);
        impl Drop for Tracker {
            fn drop(&mut self) {
                self.0.fetch_add(1, Ordering::SeqCst);
            }
        }

        let dropped = Arc::new(AtomicUsize::new(0));
        {
            let ring = SpscRing::<Tracker, 4>::new();
            assert!(
                ring.push(Tracker(Arc::clone(&dropped)))
                    .is_ok()
            );
            assert!(
                ring.push(Tracker(Arc::clone(&dropped)))
                    .is_ok()
            );
            assert!(
                ring.push(Tracker(Arc::clone(&dropped)))
                    .is_ok()
            );
        }
        assert_eq!(dropped.load(Ordering::SeqCst), 3);
    }
}

Chapter 18 puts a queue at the center of a thread pool: a fixed set of threads that take jobs from a shared queue.

Exercises

1. In bounded_buffer.rs, measure how long each push waits and print it. Change the sleep times until the producers visibly block.
2. Add try_push(&self, item: T) -> Result<(), T> to the library BoundedQueue. It should return at once when the queue is full.
3. Add pop_timeout(&self, timeout: Duration) -> Option<T> using Condvar::wait_timeout_while.
4. Replace the two condition variables in bounded_buffer.rs with one, using notify_all. Check that it still works, and count the wakeups with an atomic counter.
5. Split SpscRing into Producer<T, N> and Consumer<T, N> handles that share the ring through an Arc. Give push and pop a &mut self receiver, so that each handle can be used by only one thread at a time.