Producer, consumer and queues

The pieces of the chapter combine into a few shapes that appear in real programs: a queue between a producer and a consumer, a value that many threads read and one writes, and a worker that stops when a flag is set. Each one is a lock, a condition and a rule about who waits for whom.

This lesson builds a bounded queue with back pressure, a reader-writer lock, a stopping flag, and ends with the compiler refusing a write through a read guard.

A bounded queue

A bounded queue has two conditions: not empty for the consumer, not full for the producer. Two condition variables on the same mutex keep the two waiting groups apart, so a consumer that makes room wakes a producer and a producer that adds an item wakes a consumer.

The queue holds at most two items and the producer sends five. The consumer drains them in order, and the producer waits whenever it is ahead.

use std::collections::VecDeque; use std::sync::{Arc, Condvar, Mutex}; use std::thread; struct Queue { items: Mutex<VecDeque<i32>>, space: Condvar, ready: Condvar, capacity: usize, } fn main() { let queue = Arc::new(Queue { items: Mutex::new(VecDeque::new()), space: Condvar::new(), ready: Condvar::new(), capacity: 2, }); let producer = { let queue = Arc::clone(&queue); thread::spawn(move || { for value in 1..=5 { let mut items = queue.items.lock().unwrap(); while items.len() == queue.capacity { items = queue.space.wait(items).unwrap(); } items.push_back(value); queue.ready.notify_one(); } }) }; let mut values = Vec::new(); let mut items = queue.items.lock().unwrap(); while values.len() < 5 { while items.is_empty() { items = queue.ready.wait(items).unwrap(); } if let Some(value) = items.pop_front() { values.push(value); queue.space.notify_one(); } } producer.join().unwrap(); println!("{values:?}"); }

rustc --edition 2024 bounded.rs
./bounded

[1, 2, 3, 4, 5]

The order is the queue order, which is why the output is the same on every run even though the two threads interleave. Back pressure is the wait on a full queue: a producer that could fill memory instead waits, and the consumer sets the pace. The lesson on condition variables is the pair behind it.

A channel with a bound is the same queue without the code: sync_channel(n) blocks the sender when n items are waiting, and the standard library implements the two conditions inside. The hand-written version is worth reading once because it shows what the channel does, and worth replacing with the channel in a program that only needs the behaviour.

Readers and a writer

RwLock lets any number of readers in at once and a writer only when nobody reads. It is the right type for a value that is read often and written rarely, such as a configuration that is reloaded.

The program reads a vector, appends to it under the write lock and reads it again. The two reads are separate, so the lock is free between them.

use std::sync::RwLock; fn main() { let data = RwLock::new(vec![1, 2, 3]); { let reader = data.read().unwrap(); println!("{}", reader.len()); } { let mut writer = data.write().unwrap(); writer.push(4); } let reader = data.read().unwrap(); println!("{reader:?}"); }

rustc --edition 2024 rwlock.rs
./rwlock

3
[1, 2, 3, 4]

The read guard and the write guard are different types, and only the write guard implements DerefMut. That is what the last section turns into a compile error. The RefCell lesson is the single-threaded version of the same distinction.

With RwLock the reader count is the interesting number: a writer waits until every reader is gone, so a stream of readers can starve a writer on some implementations. The standard library does not promise fairness, and a configuration that is reloaded rarely is a better fit than a pair of hot loops.

A consumer that must not block uses try_lock on the mutex and try_recv on a channel: both return at once and let the thread do something else. The waits in this lesson block because a worker with nothing else to do should sleep, and a program that polls a queue from the main thread should not.

A worker that stops on a flag

A long-running worker is stopped with an AtomicBool that the main thread sets. The worker reads it with a relaxed load, because it only needs the value and not the data around it.

The worker counts until the flag is set. The program sleeps, stops the worker and prints whether it did any work, which is stable even though the count itself depends on the machine.

use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::thread; use std::time::Duration; fn main() { let stop = Arc::new(AtomicBool::new(false)); let worker = { let stop = Arc::clone(&stop); thread::spawn(move || { let mut count = 0; while !stop.load(Ordering::Relaxed) { count += 1; thread::sleep(Duration::from_millis(5)); } count }) }; thread::sleep(Duration::from_millis(50)); stop.store(true, Ordering::Relaxed); let count = worker.join().unwrap(); println!("{}", count > 0); }

rustc --edition 2024 flag.rs
./flag

true

The sleep in the loop is what keeps the flag check from spinning a core, and the sleep in main is what gives the worker something to do. A real worker checks the flag between units of work and passes a deadline to the calls that can block, which the lesson on time covers.

A write through the read lock

A read guard dereferences to a shared reference, so it cannot be written. The compiler reports E0594 and names the missing DerefMut, which is the same message a RefCell borrow produces at run time.

The program takes a read lock and adds one. The fix is write instead of read.

use std::sync::RwLock; fn main() { let data = RwLock::new(1); let guard = data.read().unwrap(); *guard += 1; println!("{guard}"); }

rustc --edition 2024 read_err.rs

error[E0594]: cannot assign to data in dereference of `std::sync::RwLockReadGuard<'_, i32>`
--> read_err.rs:6:5
|
6 | *guard += 1;
| ^^^^^^^^^^^ cannot assign
|
= help: trait `DerefMut` is required to modify through a dereference, but it is not implemented for `std::sync::RwLockReadGuard<'_, i32>`

error: aborting due to 1 previous error

For more information about this error, try `rustc --explain E0594`.

The help line is the whole design: several readers may hold the lock at once, so a reader cannot change what the others are reading. The next lesson leaves the thread-per-task model and looks at futures.

The three shapes of this lesson are independent: a queue between two threads, a value many threads read and one writes, and a worker that ends on a flag. A program usually needs two of them at once, which is how the pool of the previous lesson and the server of the next one are built.

Exercises

1. Add a close method to the bounded queue that wakes every waiter and ends both loops.
2. Use an RwLock for a cache that threads read and one thread refreshes.
3. Make the stopping flag an AtomicUsize that counts units of work and stops after a hundred.

The code on this page was built with rustc 1.96.1 and run on Alpine Linux 3.24.3 (x86_64) in a Docker container.

Article author: Arthur Isaev