Thread Pools - Concepts

A thread pool is a concurrency pattern that maintains a fixed, pre-allocated set of worker OS threads waiting for tasks on a shared queue. Instead of paying the OS syscall and 2 MiB stack-allocation cost of std::thread::spawn on every incoming request (and risking memory exhaustion if 50,000 requests arrive at once), callers submit boxed closures (Box<dyn FnOnce() + Send + 'static>) to the pool, and idle workers execute them sequentially.

Thread Pool Architecture

Caller Threads                          Shared Channel Queue             Pre-Spawned Worker Threads
+------------------+                    +-------------------+           +--------------------+
| pool.execute(j1) +---> Sender::send ->| [ Job3 | Job2 |Job1] --lock-->| Worker 0 (runs j1) |
| pool.execute(j2) |                    +-------------------+  \        +--------------------+
+------------------+                                            +-lock->| Worker 1 (runs j2) |
                                                                        +--------------------+
Graceful Shutdown (`Drop for ThreadPool`):
  1. Drop `Sender<Job>` ---> `rx.recv()` returns `Err` in all workers once queue drains!
  2. Join every worker's `JoinHandle<()>` so all in-flight jobs finish cleanly!

Building a Complete Graceful-Shutdown Thread Pool

A production-ready thread pool requires two careful ownership patterns:

1. Shared Receiver<Job> across workers: Because mpsc::Receiver<Job> is single-consumer (!Sync), wrapping it in Arc<Mutex<mpsc::Receiver<Job>>> allows multiple worker threads to lock the receiver briefly, pull the next Job, drop the MutexGuard immediately, and then execute the Job in parallel!
2. Graceful shutdown in Drop: Storing sender: Option<mpsc::Sender<Job>> allows ThreadPool::drop to drop(self.sender.take()) first (closing the channel so workers exit their recv() loops once queued jobs finish) and then join each worker's JoinHandle<()>:

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

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

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

impl ThreadPool {
    pub fn new(size: usize) -> Self {
        assert!(size > 0, "pool size must be at least 1");
        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));
        let mut workers = Vec::with_capacity(size);

        for id in 0..size {
            let rx = Arc::clone(&receiver);
            workers.push(thread::spawn(move || {
                loop {
                    // Lock the receiver ONLY long enough to pull one Job;
                    // the MutexGuard drops at the semicolon before `job()` runs!
                    let message = rx.lock().unwrap().recv();
                    match message {
                        Ok(job) => job(),
                        Err(_) => {
                            // Sender was dropped and queue is empty -> exit worker loop!
                            let _ = id;
                            break;
                        }
                    }
                }
            }));
        }

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

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        if let Some(sender) = &self.sender {
            sender.send(Box::new(f)).unwrap();
        }
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        // 1. Drop the Sender FIRST so workers receive `Err` once the queue drains:
        drop(self.sender.take());

        // 2. Join every worker thread to guarantee all queued jobs finish:
        for handle in self.workers.drain(..) {
            handle.join().unwrap();
        }
    }
}

fn main() {
    let results = Arc::new(Mutex::new(Vec::new()));

    {
        let pool = ThreadPool::new(3);
        for task_id in 1..=6 {
            let results = Arc::clone(&results);
            pool.execute(move || {
                results.lock().unwrap().push(task_id * 10);
            });
        }
    } // `pool` drops here, draining all 6 jobs and joining all 3 workers!

    let mut out = results.lock().unwrap().clone();
    out.sort();
    println!("completed pool jobs: {:?}", out);
}

completed pool jobs: [10, 20, 30, 40, 50, 60]

Critical Lock-Scope Trap in Worker Loops

Notice why we wrote:

let message = rx.lock().unwrap().recv();
match message { ... }

instead of while let Ok(job) = rx.lock().unwrap().recv() { job(); }: in while let, a temporary MutexGuard created in the condition expression can remain alive across the loop body { job(); }, holding the receiver lock while job() executes and serializing the entire thread pool to 1 worker at a time! Binding let message = rx.lock().unwrap().recv(); in its own statement guarantees the MutexGuard is dropped before job() starts.