Thread Pool

A thread pool maintains a fixed set of pre-spawned OS worker threads that wait on a shared synchronization queue for boxed closures (Box<dyn FnOnce() + Send + 'static>) and provides a Future-like result handle (JobHandle<T>) plus deterministic graceful shutdown in Drop.

Thread Pool Architecture

+-------------------------------------------------------------------+
| ThreadPool                                                        |
|   sender: Option<mpsc::Sender<Job>>                               |
|   workers: Vec<JoinHandle<()>>                                    |
+------------------------------+------------------------------------+
                               |
              mpsc::channel()  |  Arc<Mutex<mpsc::Receiver<Job>>>
                               v
         +---------------------+---------------------+
         |                     |                     |
         v                     v                     v
   +-----------+         +-----------+         +-----------+
   | Worker 0  |         | Worker 1  |         | Worker 2  |
   +-----------+         +-----------+         +-----------+

Graceful Shutdown in Drop:
  1. Take `self.sender` and drop it -> closes channel
  2. Workers receive `Err(RecvError)` once queued jobs finish
  3. Join every worker `JoinHandle`

Complete Thread Pool Implementation

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

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

pub struct TaskHandle<T> {
    rx: mpsc::Receiver<T>,
}

impl<T> TaskHandle<T> {
    pub fn wait(self) -> Result<T, mpsc::RecvError> {
        self.rx.recv()
    }
}

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

impl ThreadPool {
    pub fn new(num_workers: usize) -> Self {
        assert!(num_workers > 0, "ThreadPool requires at least 1 worker");

        let (sender, receiver) = mpsc::channel::<Job>();
        let shared_rx = Arc::new(Mutex::new(receiver));
        let mut workers = Vec::with_capacity(num_workers);

        for id in 0..num_workers {
            let rx = Arc::clone(&shared_rx);
            let handle = thread::Builder::new()
                .name(format!("pool-worker-{id}"))
                .spawn(move || loop {
                    let message = {
                        let guard = rx.lock().expect("receiver mutex poisoned");
                        guard.recv()
                    };
                    match message {
                        Ok(job) => job(),
                        Err(_) => break,
                    }
                })
                .expect("failed to spawn worker thread");
            workers.push(handle);
        }

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

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        if let Some(tx) = &self.sender {
            tx.send(Box::new(f)).expect("thread pool has shut down");
        }
    }

    pub fn spawn_with_handle<F, T>(&self, f: F) -> TaskHandle<T>
    where
        F: FnOnce() -> T + Send + 'static,
        T: Send + 'static,
    {
        let (result_tx, result_rx) = mpsc::sync_channel(1);
        self.execute(move || {
            let output = f();
            let _ = result_tx.send(output);
        });
        TaskHandle { rx: result_rx }
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        // Drop the sender first so workers observe channel disconnection
        drop(self.sender.take());
        for handle in self.workers.drain(..) {
            let _ = handle.join();
        }
    }
}

fn is_prime(n: u64) -> bool {
    if n < 2 {
        return false;
    }
    if n % 2 == 0 {
        return n == 2;
    }
    let mut d = 3;
    while d * d <= n {
        if n % d == 0 {
            return false;
        }
        d += 2;
    }
    true
}

fn main() {
    let pool = ThreadPool::new(4);
    let ranges = [
        (1u64, 25_000u64),
        (25_001, 50_000),
        (50_001, 75_000),
        (75_001, 100_000),
    ];

    let handles: Vec<TaskHandle<(u64, u64, usize)>> = ranges
        .into_iter()
        .map(|(lo, hi)| {
            pool.spawn_with_handle(move || {
                let count = (lo..=hi).filter(|&x| is_prime(x)).count();
                (lo, hi, count)
            })
        })
        .collect();

    let mut total_primes = 0;
    for handle in handles {
        let (lo, hi, count) = handle.wait().unwrap();
        println!("Range {:>6}..={:>6} -> {:>4} primes", lo, hi, count);
        total_primes += count;
    }
    println!("Total primes in 1..=100,000: {}", total_primes);
}

Range      1..= 25000 -> 2762 primes
Range  25001..= 50000 -> 2371 primes
Range  50001..= 75000 -> 2260 primes
Range  75001..=100000 -> 2199 primes
Total primes in 1..=100,000: 9592