Thread pools

Starting a thread costs time and memory. The operating system must create it, give it a stack, and schedule it. For a program that runs thousands of small jobs, starting a thread per job wastes most of its time on starting threads.

A thread pool starts a fixed number of threads once, and reuses them. The threads are called workers. Each worker waits for a job, runs it, and waits for the next. Jobs wait in a queue until a worker is free. This is the queue of chapter 17, with the consumers built in.

This chapter builds the pool several times. Each version fixes a problem found in the one before, so read them in order.

18.1 The parts of a pool

Every pool here has the same parts (figure 18.1).

The job type. A job is any closure that the pool can run once on another thread:

type Job = Box<dyn FnOnce() + Send + 'static>;

Each closure has its own unnamed type, so a queue cannot hold them directly. Box<dyn FnOnce()> puts each closure on the heap behind a pointer of one common type. dyn FnOnce() means “some type that can be called once with no arguments”. Send lets the job move to a worker thread. 'static means the job borrows nothing that could be dropped before the job runs.

The channel. std::sync::mpsc::channel() returns a Sender and a Receiver. Values sent through the sender come out of the receiver in the same order. mpsc stands for multiple producer, single consumer: the Sender can be cloned, but there is only one Receiver.

The shared receiver. Every worker needs to take jobs from the one receiver. So the pool wraps it in a Mutex, and shares the mutex through an Arc. A worker locks the mutex, calls recv() to take the next job, and runs it.

Shutdown. When every Sender is dropped, the channel closes. recv() then returns Err once the remaining jobs are taken. A worker that sees the Err leaves its loop, and the thread ends.

18.2 Version 1, and a bug you can measure

Version 1 writes the pool of figure 18.1 directly. new starts the workers, execute sends a job, and join closes the channel and waits for every worker. Each worker locks the shared receiver to take a job. How long it keeps that lock decides whether the workers run at the same time.

A Job is a boxed closure that runs once and can move to another thread. The pool holds the sending end of the job channel and the handles of its workers:

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

type Job = Box<dyn FnOnce() + Send + 'static>;

pub struct ThreadPool {
    sender: mpsc::Sender<Job>,
    workers: Vec<thread::JoinHandle<()>>,
}

new builds the parts of figure 18.1 and starts the workers:

impl ThreadPool {
    pub fn new(size: usize) -> Self {
        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));

        let workers = (0..size)
            .map(|_| {
                let rx = Arc::clone(&receiver);
                thread::spawn(move || {
                    while let Ok(job) = rx.lock().unwrap().recv() {
                        job();
                    }
                })
            })
            .collect();

        Self { sender, workers }
    }
    // ...
}

mpsc::channel::<Job>() creates the channel. The receiver goes into Arc::new(Mutex::new(...)), so every worker can share it and only one at a time can use it. For each worker, Arc::clone(&receiver) makes another handle to the same mutex, and thread::spawn(move || ...) starts a thread that owns that handle. The move closure is the worker’s whole life: take a job, run it, and repeat until recv fails. collect() gathers the JoinHandles that spawn returned into the workers vector.

Submitting a job and shutting down are two short methods:

impl ThreadPool {
    // ...
    pub fn execute(&self, job: Job) {
        self.sender.send(job).unwrap();
    }

    pub fn join(self) {
        drop(self.sender);
        for w in self.workers {
            let _ = w.join();
        }
    }
}

execute sends the boxed job into the channel. It does not wait for a worker: the job sits in the channel until a worker takes it. join takes self by value, so the pool cannot be used after it. drop(self.sender) closes the channel. Each worker finishes the jobs still queued, gets Err from recv, and ends. w.join() waits for each worker thread to finish.

main queues 10,000 jobs that each print a line, then joins:

fn main() {
    let pool = ThreadPool::new(4);
    for i in 0..10_000 {
        pool.execute(Box::new(move || println!("task {i}")));
    }

    pool.join();
}

Listing 18.1 The complete program. src/bin/thread_pool.rs

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

type Job = Box<dyn FnOnce() + Send + 'static>;

pub struct ThreadPool {
    sender: mpsc::Sender<Job>,
    workers: Vec<thread::JoinHandle<()>>,
}

impl ThreadPool {
    pub fn new(size: usize) -> Self {
        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));

        let workers = (0..size)
            .map(|_| {
                let rx = Arc::clone(&receiver);
                thread::spawn(move || {
                    while let Ok(job) = rx.lock().unwrap().recv() {
                        job();
                    }
                })
            })
            .collect();

        Self { sender, workers }
    }

    pub fn execute(&self, job: Job) {
        self.sender.send(job).unwrap();
    }

    pub fn join(self) {
        drop(self.sender);
        for w in self.workers {
            let _ = w.join();
        }
    }
}

fn main() {
    let pool = ThreadPool::new(4);
    for i in 0..10_000 {
        pool.execute(Box::new(move || println!("task {i}")));
    }

    pool.join();
}

The program prints 10,000 lines and looks correct. It has one bug, in the worker loop:

while let Ok(job) = rx.lock().unwrap().recv() {
    job();
}

rx.lock().unwrap() creates a MutexGuard. The guard is a temporary value inside the while let condition. A temporary in that position lives until the end of the loop body, so the guard is still alive while job() runs. The worker holds the lock for the whole job, and no other worker can take a job in the meantime.

To see the effect, I ran four jobs that each sleep for 200 ms on four workers. I ran them once with this loop and once with the loop of version 2. Figure 18.2 shows the result.

With while let, the four jobs ran one after another and took 814 ms. With the lock released before the job, they ran at the same time and took 205 ms. The first pool had four threads but used one at a time.

Animation 18.1 replays both runs. Watch the lock. In version 2 it is free again before each job starts. In version 1 it stays with the worker that is running.

The same bug has a second effect. If a job panics, the panic happens while the guard is alive, so it poisons the mutex. When I piped this program’s output into head, which closes the pipe after five lines, println! panicked inside a job. The other three workers then panicked on .unwrap() of the poisoned lock, and main panicked when send found no worker left.

18.3 Version 2: release the lock before the job

Listing 18.2 The pool type, the worker loop, and execute (lines 10 to 64). src/problems/thread_pool_v2.rs

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

type Job = Box<dyn FnOnce() + Send + 'static>;

pub struct ThreadPool {
    sender: mpsc::Sender<Job>,
    workers: Vec<thread::JoinHandle<()>>,
}

impl ThreadPool {
    /// Start `size` workers. Panics if `size` is zero.
    pub fn new(size: usize) -> Self {
        assert!(size > 0, "a thread pool needs at least one worker");

        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));

        let workers = (0..size)
            .map(|_| {
                let receiver = Arc::clone(&receiver);
                thread::spawn(move || {
                    loop {
                        // The guard is a temporary of this `let` statement, so the
                        // lock is released here, before the job runs.
                        let job = receiver.lock().unwrap().recv();
                        match job {
                            Ok(job) => job(),
                            Err(_) => break, // every sender is gone: shut down
                        }
                    }
                })
            })
            .collect();

        Self { sender, workers }
    }

    /// Queue a closure to run on the next free worker.
    pub fn execute<F>(&self, job: F)
    where
        F: FnOnce() + Send + 'static,
    {
        self.sender.send(Box::new(job)).unwrap();
    }

    /// Stop accepting jobs, let the workers drain the queue, and wait for them.
    pub fn join(self) {
        drop(self.sender);
        for worker in self.workers {
            let _ = worker.join();
        }
    }
}

Job is a type alias for the work a pool runs, a boxed closure. The closure runs once (FnOnce), can move to another thread (Send), and borrows nothing short-lived ('static). ThreadPool holds the sending end of the channel that carries jobs, and the handles of its worker threads.

The fix is one line. let job = receiver.lock().unwrap().recv(); is a separate statement, and the temporary guard is dropped at the end of that statement. The lock is released before match runs the job.

Version 2 also changes execute. It is generic over F: FnOnce() + Send + 'static, and does the boxing itself. Callers write pool.execute(|| ...) instead of pool.execute(Box::new(|| ...)).

Listing 18.3 The complete file, with tests. src/problems/thread_pool_v2.rs

//! Thread pool, version 2: workers really run in parallel.
//!
//! Version 1 (`src/bin/thread_pool.rs`) holds the receiver's lock while a job runs,
//! because the `MutexGuard` in `while let Ok(job) = rx.lock().unwrap().recv()` lives
//! until the end of the loop body. This version takes the job in its own `let`
//! statement, so the guard is dropped before the job starts.
//!
//! It also accepts any closure in `execute`, so callers no longer write `Box::new`.

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

type Job = Box<dyn FnOnce() + Send + 'static>;

pub struct ThreadPool {
    sender: mpsc::Sender<Job>,
    workers: Vec<thread::JoinHandle<()>>,
}

impl ThreadPool {
    /// Start `size` workers. Panics if `size` is zero.
    pub fn new(size: usize) -> Self {
        assert!(size > 0, "a thread pool needs at least one worker");

        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));

        let workers = (0..size)
            .map(|_| {
                let receiver = Arc::clone(&receiver);
                thread::spawn(move || {
                    loop {
                        // The guard is a temporary of this `let` statement, so the
                        // lock is released here, before the job runs.
                        let job = receiver.lock().unwrap().recv();
                        match job {
                            Ok(job) => job(),
                            Err(_) => break, // every sender is gone: shut down
                        }
                    }
                })
            })
            .collect();

        Self { sender, workers }
    }

    /// Queue a closure to run on the next free worker.
    pub fn execute<F>(&self, job: F)
    where
        F: FnOnce() + Send + 'static,
    {
        self.sender.send(Box::new(job)).unwrap();
    }

    /// Stop accepting jobs, let the workers drain the queue, and wait for them.
    pub fn join(self) {
        drop(self.sender);
        for worker in self.workers {
            let _ = worker.join();
        }
    }
}

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

    use super::*;

    #[test]
    fn runs_every_job() {
        let pool = ThreadPool::new(4);
        let counter = Arc::new(AtomicUsize::new(0));
        for _ in 0..100 {
            let counter = Arc::clone(&counter);
            pool.execute(move || {
                counter.fetch_add(1, Ordering::Relaxed);
            });
        }
        pool.join();
        assert_eq!(counter.load(Ordering::Relaxed), 100);
    }

    #[test]
    fn jobs_run_at_the_same_time() {
        // Four jobs wait on a barrier for four parties. They can only all get
        // through if four workers run them at once; version 1 would never finish.
        let pool = ThreadPool::new(4);
        let barrier = Arc::new(Barrier::new(4));
        let (done_tx, done_rx) = mpsc::channel();

        for _ in 0..4 {
            let barrier = Arc::clone(&barrier);
            let done_tx = done_tx.clone();
            pool.execute(move || {
                barrier.wait();
                done_tx.send(()).unwrap();
            });
        }

        for _ in 0..4 {
            done_rx
                .recv_timeout(Duration::from_secs(5))
                .expect("jobs did not run in parallel");
        }
        pool.join();
    }

    #[test]
    #[should_panic(expected = "at least one worker")]
    fn zero_workers_is_rejected() {
        let _ = ThreadPool::new(0);
    }
}

The test jobs_run_at_the_same_time checks the fix without measuring time. It uses a Barrier for four threads. A barrier makes each thread that calls wait() sleep until four threads have called it, then releases them all. Four jobs wait on the barrier. They can only get through if four workers run them at the same moment. With version 1’s loop, the first job would hold the lock and wait on the barrier forever.

The test waits for the results with recv_timeout(Duration::from_secs(5)), so a broken pool fails the test after five seconds instead of hanging it.

18.4 Version 3: errors, and shutdown on drop

Version 2 still panics in three places: for a size of zero, when send fails, and when the lock is poisoned. And if the caller forgets to call join, the pool is dropped without waiting for its jobs. Version 3 fixes both.

Listing 18.4 The job type and the error type (lines 8 to 32). src/problems/thread_pool_v3.rs

use std::{
    fmt,
    sync::{Arc, Mutex, mpsc},
    thread,
};

type Job = Box<dyn FnOnce() + Send + 'static>;

/// Why the pool refused a request.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PoolError {
    NoWorkers,
    ShutDown,
}

impl fmt::Display for PoolError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NoWorkers => write!(f, "a thread pool needs at least one worker"),
            Self::ShutDown => write!(f, "the thread pool has shut down"),
        }
    }
}

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

PoolError names the two ways a request can fail. It implements Display for a message, and Error so callers can use ? and Box<dyn Error> with it.

Listing 18.5 The pool, execute, shutdown, and Drop.

pub struct ThreadPool {
    // `None` after shutdown: dropping the sender is what tells workers to stop.
    sender: Option<mpsc::Sender<Job>>,
    workers: Vec<thread::JoinHandle<()>>,
}

impl ThreadPool {
    // ...
    /// Queue a closure to run on the next free worker.
    pub fn execute<F>(&self, job: F) -> Result<(), PoolError>
    where
        F: FnOnce() + Send + 'static,
    {
        self.sender
            .as_ref()
            .ok_or(PoolError::ShutDown)?
            .send(Box::new(job))
            .map_err(|_| PoolError::ShutDown)
    }

    /// Stop accepting jobs, drain the queue, and wait for every worker.
    /// Calling it again does nothing.
    pub fn shutdown(&mut self) {
        drop(self.sender.take());
        for worker in self.workers.drain(..) {
            let _ = worker.join();
        }
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        self.shutdown();
    }
}

The sender is now an Option<Sender<Job>>. shutdown takes it out with self.sender.take(), which leaves None behind, and drops it. execute checks for None with .ok_or(PoolError::ShutDown)?.

shutdown takes &mut self, not self, so that Drop can call it. self.workers.drain(..) removes the handles from the vector one at a time and gives each to the loop. A second call finds no sender and no workers, so it does nothing.

impl Drop for ThreadPool calls shutdown. When a pool goes out of scope, it closes its channel and waits for every queued job to finish.

Listing 18.6 The worker loop (lines 91 to 104).

fn worker_loop(receiver: &Mutex<mpsc::Receiver<Job>>) {
    loop {
        // A poisoned lock means another worker panicked while holding it;
        // stop this worker instead of panicking a second time.
        let Ok(guard) = receiver.lock() else { return };
        let job = guard.recv();
        drop(guard);

        match job {
            Ok(job) => job(),
            Err(_) => return,
        }
    }
}

The worker loop is now a named function. It uses let ... else to leave quietly if the lock is poisoned, and drops the guard explicitly before running the job.

Listing 18.7 The complete file, with tests. src/problems/thread_pool_v3.rs

//! Thread pool, version 3: errors instead of panics, and shutdown on drop.
//!
//! - `new` returns `Err(PoolError::NoWorkers)` for a size of zero.
//! - `execute` returns `Err(PoolError::ShutDown)` once the pool no longer accepts work.
//! - Dropping the pool closes the queue and joins every worker, so a pool that goes
//!   out of scope still finishes the jobs it was given.

use std::{
    fmt,
    sync::{Arc, Mutex, mpsc},
    thread,
};

type Job = Box<dyn FnOnce() + Send + 'static>;

/// Why the pool refused a request.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PoolError {
    NoWorkers,
    ShutDown,
}

impl fmt::Display for PoolError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NoWorkers => write!(f, "a thread pool needs at least one worker"),
            Self::ShutDown => write!(f, "the thread pool has shut down"),
        }
    }
}

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

pub struct ThreadPool {
    // `None` after shutdown: dropping the sender is what tells workers to stop.
    sender: Option<mpsc::Sender<Job>>,
    workers: Vec<thread::JoinHandle<()>>,
}

impl ThreadPool {
    /// Start `size` workers.
    pub fn new(size: usize) -> Result<Self, PoolError> {
        if size == 0 {
            return Err(PoolError::NoWorkers);
        }

        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));

        let workers = (0..size)
            .map(|_| {
                let receiver = Arc::clone(&receiver);
                thread::spawn(move || worker_loop(&receiver))
            })
            .collect();

        Ok(Self {
            sender: Some(sender),
            workers,
        })
    }

    /// Queue a closure to run on the next free worker.
    pub fn execute<F>(&self, job: F) -> Result<(), PoolError>
    where
        F: FnOnce() + Send + 'static,
    {
        self.sender
            .as_ref()
            .ok_or(PoolError::ShutDown)?
            .send(Box::new(job))
            .map_err(|_| PoolError::ShutDown)
    }

    /// Stop accepting jobs, drain the queue, and wait for every worker.
    /// Calling it again does nothing.
    pub fn shutdown(&mut self) {
        drop(self.sender.take());
        for worker in self.workers.drain(..) {
            let _ = worker.join();
        }
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        self.shutdown();
    }
}

fn worker_loop(receiver: &Mutex<mpsc::Receiver<Job>>) {
    loop {
        // A poisoned lock means another worker panicked while holding it;
        // stop this worker instead of panicking a second time.
        let Ok(guard) = receiver.lock() else { return };
        let job = guard.recv();
        drop(guard);

        match job {
            Ok(job) => job(),
            Err(_) => return,
        }
    }
}

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

    use super::*;

    #[test]
    fn zero_workers_is_an_error() {
        assert_eq!(ThreadPool::new(0).err(), Some(PoolError::NoWorkers));
    }

    #[test]
    fn dropping_the_pool_finishes_queued_jobs() {
        let counter = Arc::new(AtomicUsize::new(0));
        {
            let pool = ThreadPool::new(2).unwrap();
            for _ in 0..50 {
                let counter = Arc::clone(&counter);
                pool.execute(move || {
                    counter.fetch_add(1, Ordering::Relaxed);
                })
                .unwrap();
            }
        } // drop: close the queue and join
        assert_eq!(counter.load(Ordering::Relaxed), 50);
    }

    #[test]
    fn execute_after_shutdown_is_an_error() {
        let mut pool = ThreadPool::new(1).unwrap();
        pool.shutdown();
        assert_eq!(pool.execute(|| {}), Err(PoolError::ShutDown));
        pool.shutdown(); // idempotent
    }

    #[test]
    fn errors_render_a_message() {
        assert_eq!(
            PoolError::NoWorkers.to_string(),
            "a thread pool needs at least one worker"
        );
        assert_eq!(
            PoolError::ShutDown.to_string(),
            "the thread pool has shut down"
        );
    }
}

The test dropping_the_pool_finishes_queued_jobs creates a pool inside a block, queues 50 jobs, and lets the pool go out of scope at the closing brace. After the block, all 50 jobs must have run.

18.5 Version 4: surviving a panicking job

Version 3 has one remaining weakness. If a job panics, the panic unwinds out of worker_loop and ends that worker’s thread. The pool now has one worker fewer, and nothing reports it. After enough panics, no worker is left, and jobs queue up forever.

Version 4 catches the panic (figure 18.3).

Animation 18.2 shows the difference over many jobs. Version 4 counts the panic and keeps every worker. Version 3 loses one worker per panic, until no one is left to run the queue.

Listing 18.8 The types and the worker loop. src/problems/thread_pool_v4.rs

use std::{
    fmt,
    io,
    panic::{self, AssertUnwindSafe},
    sync::{
        Arc,
        Mutex,
        atomic::{AtomicUsize, Ordering},
        mpsc,
    },
    thread,
};

type Job = Box<dyn FnOnce() + Send + 'static>;

/// Why the pool could not be created or refused a job.
#[derive(Debug)]
pub enum PoolError {
    NoWorkers,
    Spawn(io::Error),
    ShutDown,
}

impl fmt::Display for PoolError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NoWorkers => write!(f, "a thread pool needs at least one worker"),
            Self::Spawn(error) => write!(f, "failed to spawn a worker thread: {error}"),
            Self::ShutDown => write!(f, "the thread pool has shut down"),
        }
    }
}

impl std::error::Error for PoolError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Spawn(error) => Some(error),
            _ => None,
        }
    }
}

/// What the pool did over its lifetime.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Report {
    pub completed: usize,
    pub panicked: usize,
}

#[derive(Default)]
struct Counters {
    completed: AtomicUsize,
    panicked: AtomicUsize,
}

pub struct ThreadPool {
    sender: Option<mpsc::Sender<Job>>,
    workers: Vec<thread::JoinHandle<()>>,
    counters: Arc<Counters>,
}

fn worker_loop(receiver: &Mutex<mpsc::Receiver<Job>>, counters: &Counters) {
    loop {
        let Ok(guard) = receiver.lock() else { return };
        let job = guard.recv();
        drop(guard);

        let Ok(job) = job else { return };

        // `AssertUnwindSafe` is sound here: the job owns everything it touches,
        // and a panicking job's partial state is never observed by this loop.
        match panic::catch_unwind(AssertUnwindSafe(job)) {
            Ok(()) => counters
                .completed
                .fetch_add(1, Ordering::Relaxed),
            Err(_) => counters
                .panicked
                .fetch_add(1, Ordering::Relaxed),
        };
    }
}

Version 4 adds three things to the types of version 3:

• PoolError::Spawn(io::Error), for a worker thread the operating system refused to start. Its source method returns the io::Error inside, so a caller can see the cause.
• Counters, two atomic counts that every worker updates. The pool and the workers share them through an Arc.
• Report, a plain copy of the two counts, which shutdown returns.

panic::catch_unwind(f) runs f. If f panics, the unwinding stops at catch_unwind, which returns Err. Otherwise it returns Ok with f’s result.

catch_unwind requires the closure to be UnwindSafe. The trait marks types that cannot leave broken data visible after a panic. A boxed dyn FnOnce does not have it, because the compiler cannot see what the closure captured. AssertUnwindSafe(job) is a wrapper that declares the closure safe anyway. The comment gives the reason: the job owns everything it touches, and this loop never looks at its partial state.

Each outcome increments one of two atomic counters, completed or panicked.

Listing 18.9 Named workers (lines 73 to 105).

impl ThreadPool {
    /// Start `size` named workers.
    pub fn new(size: usize) -> Result<Self, PoolError> {
        if size == 0 {
            return Err(PoolError::NoWorkers);
        }

        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));
        let counters = Arc::new(Counters::default());

        let mut pool = Self {
            sender: Some(sender),
            workers: Vec::with_capacity(size),
            counters,
        };

        for id in 0..size {
            let receiver = Arc::clone(&receiver);
            let counters = Arc::clone(&pool.counters);
            let spawned = thread::Builder::new()
                .name(format!("pool-worker-{id}"))
                .spawn(move || worker_loop(&receiver, &counters));

            match spawned {
                Ok(handle) => pool.workers.push(handle),
                // Dropping `pool` here closes the queue and joins the workers
                // that did start, so none of them is leaked.
                Err(error) => return Err(PoolError::Spawn(error)),
            }
        }

        Ok(pool)
    }
    // ...
}

thread::Builder configures a thread before starting it. Here it gives each worker a name, pool-worker-0 and so on, which appears in panic messages and debuggers. Builder::spawn returns a Result, because the operating system can refuse to create a thread. thread::spawn panics in that case.

The pool is built before the workers are started. If starting the third worker fails, return Err(...) drops the half-built pool. Its Drop closes the channel and joins the two workers that did start, so no thread is left running.

Listing 18.10 shutdown and the report (lines 113 to 136).

impl ThreadPool {
    // ...
    /// Close the queue, wait for every queued job, and report the outcome.
    pub fn shutdown(mut self) -> Report {
        self.close_and_join();
        Report {
            completed: self
                .counters
                .completed
                .load(Ordering::Relaxed),
            panicked: self
                .counters
                .panicked
                .load(Ordering::Relaxed),
        }
    }

    fn close_and_join(&mut self) {
        drop(self.sender.take());
        for worker in self.workers.drain(..) {
            // Jobs cannot unwind out of `worker_loop`, so a join error would mean
            // a bug in the loop itself; there is nothing further to clean up.
            let _ = worker.join();
        }
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        self.close_and_join();
    }
}

shutdown takes self by value and returns a Report with the two counts. The work is in close_and_join, which Drop also calls. When shutdown returns, self is dropped, and Drop runs close_and_join a second time. The second call finds nothing left to do.

Listing 18.11 The complete file, with tests. src/problems/thread_pool_v4.rs

//! Thread pool, version 4: survives panicking jobs and reports what happened.
//!
//! - Workers are named threads created with `thread::Builder`, and a failure to
//!   spawn one is returned as `PoolError::Spawn` after the already started workers
//!   are shut down.
//! - Each job runs inside `catch_unwind`, so a panicking job no longer kills its
//!   worker and silently shrinks the pool.
//! - `shutdown` consumes the pool and returns a `Report` with the number of jobs
//!   that completed and the number that panicked.

use std::{
    fmt,
    io,
    panic::{self, AssertUnwindSafe},
    sync::{
        Arc,
        Mutex,
        atomic::{AtomicUsize, Ordering},
        mpsc,
    },
    thread,
};

type Job = Box<dyn FnOnce() + Send + 'static>;

/// Why the pool could not be created or refused a job.
#[derive(Debug)]
pub enum PoolError {
    NoWorkers,
    Spawn(io::Error),
    ShutDown,
}

impl fmt::Display for PoolError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NoWorkers => write!(f, "a thread pool needs at least one worker"),
            Self::Spawn(error) => write!(f, "failed to spawn a worker thread: {error}"),
            Self::ShutDown => write!(f, "the thread pool has shut down"),
        }
    }
}

impl std::error::Error for PoolError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Spawn(error) => Some(error),
            _ => None,
        }
    }
}

/// What the pool did over its lifetime.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Report {
    pub completed: usize,
    pub panicked: usize,
}

#[derive(Default)]
struct Counters {
    completed: AtomicUsize,
    panicked: AtomicUsize,
}

pub struct ThreadPool {
    sender: Option<mpsc::Sender<Job>>,
    workers: Vec<thread::JoinHandle<()>>,
    counters: Arc<Counters>,
}

impl ThreadPool {
    /// Start `size` named workers.
    pub fn new(size: usize) -> Result<Self, PoolError> {
        if size == 0 {
            return Err(PoolError::NoWorkers);
        }

        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));
        let counters = Arc::new(Counters::default());

        let mut pool = Self {
            sender: Some(sender),
            workers: Vec::with_capacity(size),
            counters,
        };

        for id in 0..size {
            let receiver = Arc::clone(&receiver);
            let counters = Arc::clone(&pool.counters);
            let spawned = thread::Builder::new()
                .name(format!("pool-worker-{id}"))
                .spawn(move || worker_loop(&receiver, &counters));

            match spawned {
                Ok(handle) => pool.workers.push(handle),
                // Dropping `pool` here closes the queue and joins the workers
                // that did start, so none of them is leaked.
                Err(error) => return Err(PoolError::Spawn(error)),
            }
        }

        Ok(pool)
    }

    /// Queue a closure to run on the next free worker.
    pub fn execute<F>(&self, job: F) -> Result<(), PoolError>
    where
        F: FnOnce() + Send + 'static,
    {
        self.sender
            .as_ref()
            .ok_or(PoolError::ShutDown)?
            .send(Box::new(job))
            .map_err(|_| PoolError::ShutDown)
    }

    /// Close the queue, wait for every queued job, and report the outcome.
    pub fn shutdown(mut self) -> Report {
        self.close_and_join();
        Report {
            completed: self
                .counters
                .completed
                .load(Ordering::Relaxed),
            panicked: self
                .counters
                .panicked
                .load(Ordering::Relaxed),
        }
    }

    fn close_and_join(&mut self) {
        drop(self.sender.take());
        for worker in self.workers.drain(..) {
            // Jobs cannot unwind out of `worker_loop`, so a join error would mean
            // a bug in the loop itself; there is nothing further to clean up.
            let _ = worker.join();
        }
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        self.close_and_join();
    }
}

fn worker_loop(receiver: &Mutex<mpsc::Receiver<Job>>, counters: &Counters) {
    loop {
        let Ok(guard) = receiver.lock() else { return };
        let job = guard.recv();
        drop(guard);

        let Ok(job) = job else { return };

        // `AssertUnwindSafe` is sound here: the job owns everything it touches,
        // and a panicking job's partial state is never observed by this loop.
        match panic::catch_unwind(AssertUnwindSafe(job)) {
            Ok(()) => counters
                .completed
                .fetch_add(1, Ordering::Relaxed),
            Err(_) => counters
                .panicked
                .fetch_add(1, Ordering::Relaxed),
        };
    }
}

#[cfg(test)]
mod tests {
    use std::{error::Error, sync::mpsc::channel, time::Duration};

    use super::*;

    #[test]
    fn reports_completed_jobs() {
        let pool = ThreadPool::new(4).unwrap();
        for _ in 0..20 {
            pool.execute(|| {}).unwrap();
        }
        assert_eq!(
            pool.shutdown(),
            Report {
                completed: 20,
                panicked: 0
            }
        );
    }

    #[test]
    fn a_panicking_job_does_not_kill_its_worker() {
        // One worker: if the panic killed it, the second job would never run.
        let pool = ThreadPool::new(1).unwrap();
        let (done_tx, done_rx) = channel();

        pool.execute(|| panic!("job failed on purpose"))
            .unwrap();
        pool.execute(move || done_tx.send(()).unwrap())
            .unwrap();

        done_rx
            .recv_timeout(Duration::from_secs(5))
            .expect("the worker died with the panicking job");
        assert_eq!(
            pool.shutdown(),
            Report {
                completed: 1,
                panicked: 1
            }
        );
    }

    #[test]
    fn workers_are_named() {
        let pool = ThreadPool::new(1).unwrap();
        let (name_tx, name_rx) = channel();
        pool.execute(move || {
            let name = thread::current()
                .name()
                .map(str::to_string);
            name_tx.send(name).unwrap();
        })
        .unwrap();
        assert_eq!(name_rx.recv().unwrap().as_deref(), Some("pool-worker-0"));
    }

    #[test]
    fn zero_workers_is_an_error() {
        assert!(matches!(ThreadPool::new(0), Err(PoolError::NoWorkers)));
    }

    #[test]
    fn spawn_errors_expose_their_source() {
        let error = PoolError::Spawn(io::Error::other("no threads left"));
        assert!(
            error
                .to_string()
                .contains("no threads left")
        );
        assert!(error.source().is_some());
        assert!(PoolError::ShutDown.source().is_none());
    }
}

The test a_panicking_job_does_not_kill_its_worker uses a pool with one worker. The first job panics. If that killed the worker, the second job would never run, and recv_timeout would fail after five seconds.

The test spawn_errors_expose_their_source checks the source method. Error::source returns the lower-level error that caused this one, so a caller can print the whole chain.

$ cargo test --lib problems::thread_pool
running 12 tests
test problems::thread_pool_v3::tests::zero_workers_is_an_error ... ok
test problems::thread_pool_v3::tests::errors_render_a_message ... ok
test problems::thread_pool_v2::tests::zero_workers_is_rejected - should panic ... ok
test problems::thread_pool_v3::tests::execute_after_shutdown_is_an_error ... ok
test problems::thread_pool_v3::tests::dropping_the_pool_finishes_queued_jobs ... ok
test problems::thread_pool_v2::tests::runs_every_job ... ok
test problems::thread_pool_v4::tests::spawn_errors_expose_their_source ... ok
test problems::thread_pool_v2::tests::jobs_run_at_the_same_time ... ok
test problems::thread_pool_v4::tests::zero_workers_is_an_error ... ok
test problems::thread_pool_v4::tests::reports_completed_jobs ... ok
test problems::thread_pool_v4::tests::a_panicking_job_does_not_kill_its_worker ... ok
test problems::thread_pool_v4::tests::workers_are_named ... ok

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

The filter problems::thread_pool matches all three versions. cargo test hides what passing tests print, including the message of the job that panics on purpose. Add -- --nocapture to see it. The message names the thread, pool-worker-0, which shows the thread names at work.

18.6 Shorter pools

Three more pools take the lessons of versions 2 and 3 in a more compact form, and each adds one feature.

18.6.1 Shutdown in Drop, compactly

This pool shuts itself down when it goes out of scope. Its Drop closes the channel and joins the workers, so the caller never calls a shutdown method.

The sender is wrapped in an Option, as in version 3, so Drop can close the channel by setting it to None:

type Job = Box<dyn FnOnce() + Send + 'static>;

pub struct ThreadPool {
    workers: Vec<JoinHandle<()>>,
    job_sender: Option<mpsc::Sender<Job>>,
}

Listing 18.12 The complete program. src/bin/thread_pool_bruce_lee.rs

use std::{
    sync::{Arc, Mutex, mpsc},
    thread::{self, JoinHandle},
};

type Job = Box<dyn FnOnce() + Send + 'static>;

pub struct ThreadPool {
    workers: Vec<JoinHandle<()>>,
    job_sender: Option<mpsc::Sender<Job>>,
}

impl ThreadPool {
    pub fn new(size: usize) -> Self {
        assert!(size > 0, "pool size must be > 0");
        let (job_sender, job_receiver) = mpsc::channel();
        let job_receiver = Arc::new(Mutex::new(job_receiver));

        let workers = (0..size)
            .map(|_| Self::spawn_worker(Arc::clone(&job_receiver)))
            .collect();

        Self {
            workers,
            job_sender: Some(job_sender),
        }
    }

    fn spawn_worker(job_receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> JoinHandle<()> {
        thread::spawn(move || {
            loop {
                let message = job_receiver.lock().unwrap().recv();
                match message {
                    Ok(job) => job(),
                    Err(_) => break,
                }
            }
        })
    }

    pub fn execute(&self, task: impl FnOnce() + Send + 'static) {
        if let Some(job_sender) = &self.job_sender {
            job_sender.send(Box::new(task)).unwrap();
        }
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        self.job_sender = None; // close channel so workers stop
        for worker in self.workers.drain(..) {
            worker.join().unwrap(); // Wait for workers to finish... This will ensure gracefull shutdown..
        }
    }
}

fn main() {
    let pool = ThreadPool::new(4);
    for i in 0..8 {
        pool.execute(move || println!("task {i} on {:?}", thread::current().id()));
    }
}

This pool keeps version 2’s worker loop and version 3’s Option sender and Drop. self.job_sender = None drops the sender by overwriting it. thread::spawn(move || loop { ... }) passes a closure whose whole body is the loop.

main has no explicit shutdown. The pool is dropped at the end of main, and Drop waits for all eight jobs.

$ cargo run --bin thread_pool_bruce_lee
task 0 on ThreadId(4)
task 4 on ThreadId(4)
task 5 on ThreadId(4)
task 6 on ThreadId(4)
task 7 on ThreadId(4)
task 1 on ThreadId(3)
task 2 on ThreadId(2)
task 3 on ThreadId(5)

In this run, one worker took five of the eight jobs. Printing a line is so quick that the first worker to wake could take the next job before the others were scheduled. The pool does not promise an even split.

Drop calls worker.join().unwrap(). If a job panicked, its worker’s join returns Err, and the unwrap panics inside Drop. The later versions use let _ = worker.join(); instead.

18.6.2 A bounded queue and results

This pool adds two features. The queue has a fixed size, so a caller that submits jobs faster than the workers run them has to wait. And each job’s return value comes back to the caller.

The sender is now a SyncSender, the sending end of a channel with a fixed capacity:

type Job = Box<dyn FnOnce() + Send + 'static>;

pub struct ThreadPool {
    workers: Vec<JoinHandle<()>>,
    job_sender: Option<mpsc::SyncSender<Job>>, // changed: bounded sender
}

Listing 18.13 The complete program. src/bin/thread_pool_with_back_pressure.rs

use std::{
    sync::{Arc, Mutex, mpsc},
    thread::{self, JoinHandle},
};

type Job = Box<dyn FnOnce() + Send + 'static>;

pub struct ThreadPool {
    workers: Vec<JoinHandle<()>>,
    job_sender: Option<mpsc::SyncSender<Job>>, // changed: bounded sender
}

impl ThreadPool {
    pub fn new(size: usize, queue_size: usize) -> Self {
        // changed: queue_size
        assert!(size > 0, "pool size must be > 0");
        let (job_sender, job_receiver) = mpsc::sync_channel(queue_size); // changed: bounded queue
        let job_receiver = Arc::new(Mutex::new(job_receiver));

        let workers = (0..size)
            .map(|_| Self::spawn_worker(Arc::clone(&job_receiver)))
            .collect();

        Self {
            workers,
            job_sender: Some(job_sender),
        }
    }

    fn spawn_worker(job_receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> JoinHandle<()> {
        thread::spawn(move || {
            loop {
                let message = job_receiver.lock().unwrap().recv();
                match message {
                    Ok(job) => job(),
                    Err(_) => break,
                }
            }
        })
    }

    pub fn execute(&self, task: impl FnOnce() + Send + 'static) {
        if let Some(job_sender) = &self.job_sender {
            job_sender.send(Box::new(task)).unwrap(); // blocks when queue is full
        }
    }

    // new: run a task and get its result back
    pub fn submit<T: Send + 'static>(
        &self,
        task: impl FnOnce() -> T + Send + 'static,
    ) -> mpsc::Receiver<T> {
        let (result_sender, result_receiver) = mpsc::channel();
        self.execute(move || {
            let _ = result_sender.send(task());
        });
        result_receiver
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        self.job_sender = None;
        for worker in self.workers.drain(..) {
            let _ = worker.join();
        }
    }
}

fn main() {
    let pool = ThreadPool::new(4, 10); // 4 workers, at most 10 jobs waiting

    let results: Vec<_> = (0..8)
        .map(|i| pool.submit(move || i * i))
        .collect();

    for r in results {
        println!("{}", r.recv().unwrap());
    }
}

The comments marked changed and new show the two additions.

mpsc::sync_channel(queue_size) creates a channel that holds at most queue_size messages. SyncSender::send blocks while the channel is full. This is the backpressure of section 17.1, built into the standard library’s channel. A caller that queues jobs faster than the workers run them has to wait.

Figure 18.4 shows both additions: a queue that can fill up, and a private channel per job for its result.

submit returns a result. It creates a new channel for each job, and the job sends its return value into it. The caller gets the Receiver back at once, and calls recv() when it needs the value. A value that will be ready later is often called a future, and chapter 22 builds on that idea.

$ cargo run --bin thread_pool_with_back_pressure
0
1
4
9
16
25
36
49

The results print in order because main calls recv() on the receivers in order, even though the jobs may finish in any order.

18.6.3 Many producers

In this pool, several threads submit jobs at the same time, each through its own clone of the Sender. The worker loop also handles a poisoned lock and a closed channel separately.

SharedLockedChannel names the type every worker shares: one receiver, behind a Mutex, in an Arc. The pool holds a bounded sender in an Option, and the worker handles:

type Job = Box<dyn FnOnce() + Send + 'static>;
type SharedLockedChannel = Arc<Mutex<Receiver<Job>>>;

pub struct ThreadPool {
    tx: Option<SyncSender<Job>>,
    workers: Vec<JoinHandle<()>>,
}

Listing 18.14 The complete program. src/bin/mpmc_thread_pool_engine.rs

use std::{
    sync::{
        Arc,
        Mutex,
        mpsc::{self, Receiver, SendError, SyncSender},
    },
    thread::{self, JoinHandle},
};

type Job = Box<dyn FnOnce() + Send + 'static>;
type SharedLockedChannel = Arc<Mutex<Receiver<Job>>>;

pub struct ThreadPool {
    tx: Option<SyncSender<Job>>,
    workers: Vec<JoinHandle<()>>,
}

impl ThreadPool {
    pub fn new(num_workers: usize, queue_size: usize) -> Self {
        let (tx, rx) = mpsc::sync_channel::<Job>(queue_size);
        let rx = SharedLockedChannel::new(Mutex::new(rx));

        let workers = (0..num_workers)
            .map(|_| Self::spawn_worker(Arc::clone(&rx)))
            .collect();

        Self {
            tx: Some(tx),
            workers,
        }
    }

    fn spawn_worker(rx: SharedLockedChannel) -> JoinHandle<()> {
        thread::spawn(move || {
            loop {
                let Ok(receiver) = rx.lock() else { break }; // mutex poisoned
                let Ok(job) = receiver.recv() else { break }; // channel closed
                drop(receiver); // release lock before running the job
                job();
            }
        })
    }

    pub fn execute<F>(&self, f: F) -> Result<(), SendError<Job>>
    where
        F: FnOnce() + Send + 'static,
    {
        let job: Job = Box::new(f);
        match &self.tx {
            Some(tx) => tx.send(job), // blocks when queue is full
            None => Err(SendError(job)),
        }
    }

    pub fn join_all<I>(&self, handles: I)
    where
        I: IntoIterator<Item = JoinHandle<()>>,
    {
        for handle in handles {
            let _ = handle.join();
        }
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.tx.take()); // close channel so workers exit
        for worker in self.workers.drain(..) {
            let _ = worker.join();
        }
    }
}

fn main() {
    let pool = Arc::new(ThreadPool::new(4, 16));

    let producers: Vec<_> = (0..3)
        .map(|p| {
            let pool = Arc::clone(&pool);
            thread::spawn(move || {
                for i in 0..5 {
                    let _ = pool.execute(move || println!("producer {p}, job {i}"));
                }
            })
        })
        .collect();

    pool.join_all(producers);
} // last Arc dropped here: Drop waits for all queued jobs

This pool’s worker loop is the clearest of the chapter:

let Ok(receiver) = rx.lock() else { break }; // mutex poisoned
let Ok(job) = receiver.recv() else { break }; // channel closed
drop(receiver); // release lock before running the job
job();

Each let ... else line handles one failure and says which. The lock is released explicitly before the job runs.

execute returns Result<(), SendError<Job>>. When the pool is shut down, it returns the job inside the error, so the caller can still run it or report it.

main shares the pool between three producer threads with an Arc. Each producer queues five jobs. When the last Arc is dropped at the end of main, Drop runs and waits for every queued job.

$ cargo run --bin mpmc_thread_pool_engine
producer 0, job 0
producer 0, job 3
producer 0, job 4
producer 1, job 1
producer 1, job 2
producer 0, job 1
producer 1, job 4
producer 2, job 0
producer 2, job 1
producer 2, job 2
producer 0, job 2
producer 2, job 4
producer 2, job 3
producer 1, job 3
producer 1, job 0

All 15 jobs ran. The order is mixed, even within one producer, because four workers run the jobs at the same time.

18.7 A parallel map that keeps order

The last pool is a single function. map_order applies a function to every input on several threads, and returns the results in the same order as the inputs.

The difficulty is that jobs finish in any order. The fix is to tag each input with its position, and use the tag to put each result back in place (figure 18.5).

The function’s signature and its two channels come first:

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

pub fn map_order<T, R, F>(worker_count: usize, inputs: Vec<T>, mapper: F) -> Vec<R>
where
    T: Send + 'static,
    R: Send + 'static,
    F: Fn(T) -> R + Send + Sync + 'static,
{
    assert!(worker_count > 0, "worker_count must be > 0");
    assert!(!inputs.is_empty(), "inputs must not be empty");

    let expected = inputs.len();
    let mapper = Arc::new(mapper);
    let (job_tx, job_rx) = mpsc::channel::<(usize, T)>();
    let job_rx = Arc::new(Mutex::new(job_rx));
    let (result_tx, result_rx) = mpsc::channel::<(usize, R)>();
    // ...
}

T is the input type, R the result type, and F the function applied to each input. Inputs and results cross threads, so both must be Send. mapper is shared by all workers, so it goes in an Arc, and it must be Sync as well as Send. It has type F: Fn(T) -> R, not FnOnce, because each worker calls it many times.

Jobs go out as (usize, T) pairs, and results come back as (usize, R) pairs. The usize is the input’s position, the tag of figure 18.5.

Each worker gets its own handle to the mapper, the job receiver, and the result sender:

pub fn map_order<T, R, F>(worker_count: usize, inputs: Vec<T>, mapper: F) -> Vec<R>
where
    T: Send + 'static,
    R: Send + 'static,
    F: Fn(T) -> R + Send + Sync + 'static,
{
    // ...
    let mut handles = Vec::with_capacity(worker_count);
    for _worker_id in 0..worker_count {
        let mapper = Arc::clone(&mapper);
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        handles.push(thread::spawn(move || {
            loop {
                let job = {
                    let guard = job_rx.lock().expect("job_rx poisoned");
                    guard.recv()
                };
                match job {
                    Ok((idx, input)) => {
                        let output = mapper(input);
                        if result_tx.send((idx, output)).is_err() {
                            break;
                        }
                    }
                    Err(_) => break, // all senders dropped
                }
            }
        }));
    }
    // ...
}

The worker takes the job inside a block: let job = { let guard = ...; guard.recv() };. The guard is dropped at the end of the block, so the lock is released before mapper runs. This is the third way in this chapter to write the fix from section 18.3. If the result channel is closed, the caller has gone, and the worker stops.

The caller sends every job, then gives up its own senders:

pub fn map_order<T, R, F>(worker_count: usize, inputs: Vec<T>, mapper: F) -> Vec<R>
where
    T: Send + 'static,
    R: Send + 'static,
    F: Fn(T) -> R + Send + Sync + 'static,
{
    // ...
    // Send all jobs, then close the job channel so workers finish after draining it.
    for (idx, input) in inputs.into_iter().enumerate() {
        job_tx
            .send((idx, input))
            .expect("bounded by memory; channel only blocks until a worker receives");
    }
    drop(job_tx);

    // Drop our result sender. Only the worker clones remain, so `iter()` ends
    // after every worker exits.
    drop(result_tx);
    // ...
}

enumerate() attaches each input’s position. Dropping job_tx closes the job channel, so the workers stop when the queue is empty. Dropping the caller’s result_tx leaves only the workers’ clones. result_rx.iter() then ends when the last worker exits and drops its clone.

Last, the results are put back in input order:

pub fn map_order<T, R, F>(worker_count: usize, inputs: Vec<T>, mapper: F) -> Vec<R>
where
    T: Send + 'static,
    R: Send + 'static,
    F: Fn(T) -> R + Send + Sync + 'static,
{
    // ...
    // Reassemble results in input order.
    let mut ordered: Vec<Option<R>> = Vec::new();
    for (idx, value) in result_rx.iter() {
        if ordered.len() <= idx {
            ordered.resize_with(idx + 1, || None);
        }
        ordered[idx] = Some(value);
    }

    for handle in handles {
        let _ = handle.join();
    }

    assert_eq!(
        ordered.len(),
        expected,
        "every job must produce exactly one result"
    );
    ordered
        .into_iter()
        .map(|value| value.expect("all results present"))
        .collect()
}

Each result is stored at its index in ordered, a Vec<Option<R>>, which grows as needed with resize_with. At the end, every slot must hold Some, and the options are unwrapped into the final Vec<R>.

Listing 18.15 The complete file, with tests. src/problems/worker_pool.rs

//! Bounded worker pool with ordered results.
//!
//! Programs often fan out network calls across a bounded number of workers.
//! This example maps `Vec<T>` to `Vec<R>` with N threads and preserves input
//! order in the output. It demonstrates `Arc<Mutex<_>>`, channels, scoped
//! ownership, and panic-free dispatch.

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

pub fn map_order<T, R, F>(worker_count: usize, inputs: Vec<T>, mapper: F) -> Vec<R>
where
    T: Send + 'static,
    R: Send + 'static,
    F: Fn(T) -> R + Send + Sync + 'static,
{
    assert!(worker_count > 0, "worker_count must be > 0");
    assert!(!inputs.is_empty(), "inputs must not be empty");

    let expected = inputs.len();
    let mapper = Arc::new(mapper);
    let (job_tx, job_rx) = mpsc::channel::<(usize, T)>();
    let job_rx = Arc::new(Mutex::new(job_rx));
    let (result_tx, result_rx) = mpsc::channel::<(usize, R)>();

    let mut handles = Vec::with_capacity(worker_count);
    for _worker_id in 0..worker_count {
        let mapper = Arc::clone(&mapper);
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        handles.push(thread::spawn(move || {
            loop {
                let job = {
                    let guard = job_rx.lock().expect("job_rx poisoned");
                    guard.recv()
                };
                match job {
                    Ok((idx, input)) => {
                        let output = mapper(input);
                        if result_tx.send((idx, output)).is_err() {
                            break;
                        }
                    }
                    Err(_) => break, // all senders dropped
                }
            }
        }));
    }

    // Send all jobs, then close the job channel so workers finish after draining it.
    for (idx, input) in inputs.into_iter().enumerate() {
        job_tx
            .send((idx, input))
            .expect("bounded by memory; channel only blocks until a worker receives");
    }
    drop(job_tx);

    // Drop our result sender. Only the worker clones remain, so `iter()` ends
    // after every worker exits.
    drop(result_tx);

    // Reassemble results in input order.
    let mut ordered: Vec<Option<R>> = Vec::new();
    for (idx, value) in result_rx.iter() {
        if ordered.len() <= idx {
            ordered.resize_with(idx + 1, || None);
        }
        ordered[idx] = Some(value);
    }

    for handle in handles {
        let _ = handle.join();
    }

    assert_eq!(
        ordered.len(),
        expected,
        "every job must produce exactly one result"
    );
    ordered
        .into_iter()
        .map(|value| value.expect("all results present"))
        .collect()
}

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

    #[test]
    fn maps_all_inputs_and_preserves_order() {
        let inputs: Vec<i32> = (0..100).collect();
        let results = map_order(4, inputs, |x| x * x);
        assert_eq!(results.len(), 100);
        for (i, value) in results.iter().enumerate() {
            assert_eq!(*value, (i as i32) * (i as i32));
        }
    }

    #[test]
    fn one_worker_works() {
        let results = map_order(1, vec![1, 2, 3], |x| x + 10);
        assert_eq!(results, vec![11, 12, 13]);
    }
}

$ cargo test --lib problems::worker_pool
running 2 tests
test problems::worker_pool::tests::one_worker_works ... ok
test problems::worker_pool::tests::maps_all_inputs_and_preserves_order ... ok

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

18.8 The versions compared

Table 18.1 What each pool handles

Chapter 19 turns to failures between programs: limiting request rates, retrying failed calls, and making repeated requests safe.

Exercises

1. Add a test to thread_pool_v2.rs that measures four jobs sleeping 200 ms each, and asserts that they finish in under 400 ms.
2. Add submit from listing 18.13 to version 4, returning the job’s result through a channel.
3. Make version 4’s queue bounded with sync_channel, and add a try_execute method that returns the job instead of blocking when the queue is full.
4. Write map_order so that it returns Result<Vec<R>, E> for a mapper of type Fn(T) -> Result<R, E>, and stops sending jobs after the first error.