37. Project: a thread pool from scratch

The same design as the C guide's §23.1 thread pool: N workers take jobs from a shared queue. Compare the shutdown logic: in Rust it lives in Drop, so a pool cannot be leaked with its threads still running.

// src/bin/pool.rs
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};

type Job = Box<dyn FnOnce() + Send + 'static>;   // any closure that can move to a thread and run once

pub struct ThreadPool {
    sender: Option<Sender<Job>>,                 // Option so Drop can take it and close the channel
    workers: Vec<JoinHandle<()>>,
}

impl ThreadPool {
    pub fn new(size: usize) -> Self {
        assert!(size > 0);
        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver: Arc<Mutex<Receiver<Job>>> = Arc::new(Mutex::new(receiver));
        let workers = (0..size)
            .map(|id| {
                let rx = Arc::clone(&receiver);
                thread::Builder::new()
                    .name(format!("pool-{id}"))
                    .spawn(move || loop {
                        // The guard is a temporary, dropped at the end of this statement,
                        // so the lock is NOT held while the job runs.
                        let msg = rx.lock().unwrap_or_else(|e| e.into_inner()).recv();
                        match msg {
                            Ok(job) => {
                                // Isolate panics: a bad job must not kill the worker.
                                if std::panic::catch_unwind(std::panic::AssertUnwindSafe(job)).is_err() {
                                    eprintln!("pool-{id}: job panicked");
                                }
                            }
                            Err(_) => break, // channel closed: shut down
                        }
                    })
                    .expect("spawn worker")
            })
            .collect();
        Self { sender: Some(sender), workers }
    }

    pub fn execute<F: FnOnce() + Send + 'static>(&self, f: F) {
        self.sender.as_ref().expect("pool is running").send(Box::new(f)).expect("workers alive");
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take());           // close the channel: every recv() now returns Err
        for w in self.workers.drain(..) {
            let _ = w.join();               // wait for queued jobs to finish
        }
    }
}

fn main() {
    let results = Arc::new(Mutex::new(Vec::new()));
    {
        let pool = ThreadPool::new(4);
        for i in 0..8u64 {
            let results = Arc::clone(&results);
            pool.execute(move || {
                if i == 5 { panic!("job {i} failed"); }
                let fib = (0..i).fold((0u64, 1u64), |(a, b), _| (b, a + b)).0;
                results.lock().unwrap().push((i, fib));
            });
        }
    } // pool dropped: waits for all jobs
    let mut r = results.lock().unwrap().clone();
    r.sort();
    println!("{r:?}"); // 7 results; job 5 panicked and was isolated
}

thread 'pool-3' panicked at src/bin/pool.rs:65:29:
job 5 failed
pool-3: job panicked
[(0, 0), (1, 1), (2, 1), (3, 2), (4, 3), (6, 8), (7, 13)]

Stretch goals: use a bounded sync_channel so execute applies backpressure. Return a handle from execute that gives back the job's result (a one-shot channel). Then compare with rayon::ThreadPool.