Thread Pool

A thread pool keeps a fixed number of worker threads alive and hands them jobs from a queue, so a program doesn't pay for starting a thread per task. Rust's standard library has threads and channels but no thread pool, so it's a good project for seeing how the pieces from the concurrency lessons fit together. We build ThreadPool with this API:

Rust

let pool = ThreadPool::new(4);
pool.execute(|| println!("fire and forget"));
let handle = pool.submit(|| expensive_calculation());
let result = handle.join();   // Ok(value), or Err if the job panicked
drop(pool);                   // waits for queued jobs, then stops the workers

Step	Topics
1. workers and a queue	`thread::spawn`, `mpsc` channels, `Arc<Mutex<Receiver>>`, `Box<dyn FnOnce() + Send>`
2. graceful shutdown	`Drop`, `Option::take`, joining threads
3. results	a channel per job, generic `submit<T>`
4. panicking jobs	`panic::catch_unwind`, `AssertUnwindSafe`
5. the complete pool	everything together on a realistic batch job

In real projects you'd reach for rayon for data parallelism or Tokio's spawn_blocking for blocking work in async code. Building one yourself shows what those crates do underneath.

Step 1: Workers pulling jobs from a queue

A job is any closure that runs once on another thread: Box<dyn FnOnce() + Send + 'static>. FnOnce because it's called once, Send because it moves to another thread, 'static because it may outlive the caller's stack frame.

The queue is an mpsc channel. A channel has only one receiver, so the workers share it through Arc<Mutex<Receiver<Job>>>: whichever worker gets the lock takes the next job. The lock is held only while receiving, not while running the job.

Rust

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

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

struct ThreadPool {
    sender: Sender<Job>,
}

impl ThreadPool {
    fn new(size: usize) -> ThreadPool {
        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));
        for _ in 0..size {
            let receiver: Arc<Mutex<Receiver<Job>>> = Arc::clone(&receiver);
            thread::spawn(move || {
                loop {
                    // The guard is a temporary, so the lock is released at the end of this statement.
                    let job = receiver.lock().unwrap().recv();
                    match job {
                        Ok(job) => job(),
                        Err(_) => break, // the sender is gone: no more jobs will come
                    }
                }
            });
        }
        ThreadPool { sender }
    }

    fn execute(&self, job: impl FnOnce() + Send + 'static) {
        self.sender.send(Box::new(job)).unwrap();
    }
}

fn main() {
    let pool = ThreadPool::new(4);
    let done = Arc::new(AtomicUsize::new(0));
    for _ in 0..8 {
        let done = Arc::clone(&done);
        pool.execute(move || {
            thread::sleep(Duration::from_millis(50));
            done.fetch_add(1, Ordering::SeqCst);
        });
    }
    thread::sleep(Duration::from_millis(300)); // crude: wait for the jobs (fixed in step 2)
    println!("{} jobs done", done.load(Ordering::SeqCst));
}

Output

8 jobs done

Eight 50 ms jobs on four workers take about 100 ms. The sleep in main is a placeholder: without it, main would return and end the process while jobs are still running.

Step 2: Graceful shutdown in `Drop`

A graceful shutdown stops accepting jobs and lets queued ones finish. In Rust this belongs in Drop, so it happens automatically when the pool goes out of scope:

Drop the Sender. Workers finish the jobs still in the channel; then recv() returns Err and their loops end.
Join every worker thread.

Both fields need to be moved out of &mut self inside drop, which is what Option::take is for.

Rust

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

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

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

impl ThreadPool {
    fn new(size: usize) -> ThreadPool {
        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));
        let workers = (0..size)
            .map(|id| {
                let receiver = Arc::clone(&receiver);
                let handle = thread::spawn(move || {
                    let mut jobs_run = 0;
                    loop {
                        let next = receiver.lock().unwrap().recv(); // the lock is released at the end of this statement
                        let Ok(job) = next else { break };
                        job();
                        jobs_run += 1;
                    }
                    jobs_run // returned through join()
                });
                (id, Some(handle))
            })
            .collect();
        ThreadPool { sender: Some(sender), workers }
    }

    fn execute(&self, job: impl FnOnce() + Send + 'static) {
        self.sender.as_ref().expect("pool is shut down").send(Box::new(job)).unwrap();
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take()); // 1. close the queue
        let mut total = 0;
        for (_, handle) in &mut self.workers {
            if let Some(handle) = handle.take() {
                total += handle.join().unwrap(); // 2. wait for the worker
            }
        }
        println!("pool shut down after {total} jobs on {} workers", self.workers.len());
    }
}

fn main() {
    let results = Arc::new(Mutex::new(Vec::new()));
    {
        let pool = ThreadPool::new(3);
        for n in 1..=6u64 {
            let results = Arc::clone(&results);
            pool.execute(move || {
                thread::sleep(Duration::from_millis(20 * n));
                results.lock().unwrap().push(n * n);
            });
        }
        println!("all jobs queued");
    } // `pool` is dropped here: Drop waits for every queued job

    let mut results = results.lock().unwrap().clone();
    results.sort();
    println!("{results:?}");
}

Output

all jobs queued
pool shut down after 6 jobs on 3 workers
[1, 4, 9, 16, 25, 36]

The worker loop receives in its own statement, let next = receiver.lock().unwrap().recv();, on purpose. The MutexGuard is a temporary, and a temporary lives until the end of the statement that created it. The shorter while let Ok(job) = receiver.lock().unwrap().recv() { job(); } compiles too, but there the guard lives until the end of the loop body: the lock stays held while the job runs, and the pool quietly runs one job at a time. (The Rust Book's thread pool chapter points out the same trap.)

Step 3: Getting results back

Callers usually want the job's result, not just a guarantee that it ran. Our submit returns a JobHandle<T>. Each submitted job gets its own one-shot channel: the job sends its result, and join waits for it. submit is generic over the result type T, but the queue still carries plain Jobs: the closure that goes on the queue captures the sender and hides T inside it.

Rust

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

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

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

struct JobHandle<T> {
    result: Receiver<T>,
}

impl<T> JobHandle<T> {
    fn join(self) -> T {
        self.result.recv().expect("the job never finished")
    }
}

impl ThreadPool {
    fn new(size: usize) -> ThreadPool {
        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 {
                        let next = receiver.lock().unwrap().recv(); // the lock is released at the end of this statement
                        let Ok(job) = next else { break };
                        job();
                    }
                })
            })
            .collect();
        ThreadPool { sender: Some(sender), workers }
    }

    fn submit<T, F>(&self, task: F) -> JobHandle<T>
    where
        F: FnOnce() -> T + Send + 'static,
        T: Send + 'static,
    {
        let (result_tx, result_rx) = mpsc::channel();
        let job: Job = Box::new(move || {
            let _ = result_tx.send(task()); // the caller may have dropped the handle
        });
        self.sender.as_ref().unwrap().send(job).unwrap();
        JobHandle { result: result_rx }
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take());
        for worker in self.workers.drain(..) {
            worker.join().unwrap();
        }
    }
}

fn is_prime(n: u64) -> bool {
    n >= 2 && (2..).take_while(|d| d * d <= n).all(|d| n % d != 0)
}

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

    // Different jobs can return different types.
    let primes = pool.submit(|| (1..50_000u64).filter(|&n| is_prime(n)).count());
    let greeting = pool.submit(|| String::from("hello from a worker"));
    let chunks: Vec<JobHandle<u64>> = (0..4u64)
        .map(|chunk| pool.submit(move || (chunk * 250_000..(chunk + 1) * 250_000).sum()))
        .collect();

    println!("{} primes below 50,000", primes.join());
    println!("{}", greeting.join());
    let total: u64 = chunks.into_iter().map(JobHandle::join).sum();
    println!("sum of 0..1,000,000 = {total}");
}

Output

5133 primes below 50,000
hello from a worker
sum of 0..1,000,000 = 499999500000

Step 4: Surviving a panicking job

If a job panics, its worker thread dies, and the pool silently shrinks. A robust pool catches the panic and reports it through the job's handle, which std::panic::catch_unwind makes possible, which stops a panic from unwinding past a point and returns it as an Err.

catch_unwind requires the closure to be UnwindSafe, a marker trait that warns you when a panic could leave shared data half-updated. Our job owns everything it touches, so wrapping it in AssertUnwindSafe is correct here.

Rust

use std::panic::{self, AssertUnwindSafe};
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};

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

#[derive(Debug)]
struct JobPanicked(String);

struct JobHandle<T> {
    result: Receiver<Result<T, JobPanicked>>,
}

impl<T> JobHandle<T> {
    fn join(self) -> Result<T, JobPanicked> {
        self.result.recv().expect("the job never finished")
    }
}

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

impl ThreadPool {
    fn new(size: usize) -> ThreadPool {
        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 {
                        let next = receiver.lock().unwrap().recv(); // the lock is released at the end of this statement
                        let Ok(job) = next else { break };
                        job();
                    }
                })
            })
            .collect();
        ThreadPool { sender: Some(sender), workers }
    }

    fn submit<T, F>(&self, task: F) -> JobHandle<T>
    where
        F: FnOnce() -> T + Send + 'static,
        T: Send + 'static,
    {
        let (result_tx, result_rx) = mpsc::channel();
        let job: Job = Box::new(move || {
            let outcome = panic::catch_unwind(AssertUnwindSafe(task)).map_err(|payload| {
                // A panic payload is usually a &str or a String.
                let message = payload
                    .downcast_ref::<&str>()
                    .map(|s| s.to_string())
                    .or_else(|| payload.downcast_ref::<String>().cloned())
                    .unwrap_or_else(|| String::from("unknown panic"));
                JobPanicked(message)
            });
            let _ = result_tx.send(outcome);
        });
        self.sender.as_ref().unwrap().send(job).unwrap();
        JobHandle { result: result_rx }
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take());
        for worker in self.workers.drain(..) {
            worker.join().unwrap();
        }
    }
}

fn parse_quantity(text: &'static str) -> u32 {
    text.parse().unwrap_or_else(|_| panic!("bad quantity {text:?}"))
}

fn main() {
    panic::set_hook(Box::new(|_| {})); // keep the demo output quiet; the handle reports the panic

    let pool = ThreadPool::new(1); // one worker: it must survive the panic to run the last job
    let first = pool.submit(|| parse_quantity("12"));
    let second = pool.submit(|| parse_quantity("twelve"));
    let third = pool.submit(|| parse_quantity("7"));

    println!("{:?}", first.join());
    println!("{:?}", second.join());
    println!("{:?}", third.join());
}

Output

Ok(12)
Err(JobPanicked("bad quantity \"twelve\""))
Ok(7)

With a single worker, the third job only runs because the worker survived the second. panic::set_hook replaces the default "thread panicked at ..." message on stderr; a real application would log it instead.

Step 5: The complete pool

The final version combines everything and adds execute for fire-and-forget jobs and a Stats count kept with atomics. It's then used for a typical batch job: validating a list of orders in parallel, collecting the results in the original order even though jobs finish in any order.

Rust

use std::panic::{self, AssertUnwindSafe};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

// ---------- pool.rs ----------

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

#[derive(Debug)]
pub struct JobPanicked(pub String);

pub struct JobHandle<T> {
    result: Receiver<Result<T, JobPanicked>>,
}

impl<T> JobHandle<T> {
    pub fn join(self) -> Result<T, JobPanicked> {
        self.result.recv().expect("the job never finished")
    }
}

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

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

impl ThreadPool {
    /// Creates a pool with `size` worker threads.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0, "a pool needs at least one thread");
        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));
        let stats = Arc::new(Stats::default());
        let workers = (0..size)
            .map(|id| {
                let receiver = Arc::clone(&receiver);
                let stats = Arc::clone(&stats);
                thread::Builder::new()
                    .name(format!("pool-worker-{id}"))
                    .spawn(move || {
                        loop {
                            let next = receiver.lock().unwrap().recv(); // the lock is released at the end of this statement
                            let Ok(job) = next else { break };
                            let counter = match panic::catch_unwind(AssertUnwindSafe(job)) {
                                Ok(()) => &stats.completed,
                                Err(_) => &stats.panicked,
                            };
                            counter.fetch_add(1, Ordering::Relaxed);
                        }
                    })
                    .expect("failed to spawn a worker thread")
            })
            .collect();
        ThreadPool { sender: Some(sender), workers, stats }
    }

    /// Runs `job` on the pool without waiting for it.
    pub fn execute(&self, job: impl FnOnce() + Send + 'static) {
        self.sender.as_ref().expect("pool is shut down").send(Box::new(job)).expect("workers are gone");
    }

    /// Runs `task` on the pool; the returned handle gives its result.
    pub fn submit<T, F>(&self, task: F) -> JobHandle<T>
    where
        F: FnOnce() -> T + Send + 'static,
        T: Send + 'static,
    {
        let (result_tx, result_rx) = mpsc::channel();
        self.execute(move || {
            let outcome = panic::catch_unwind(AssertUnwindSafe(task)).map_err(|payload| {
                let message = payload
                    .downcast_ref::<&str>()
                    .map(|s| s.to_string())
                    .or_else(|| payload.downcast_ref::<String>().cloned())
                    .unwrap_or_else(|| String::from("unknown panic"));
                JobPanicked(message)
            });
            let failed = outcome.is_err();
            let _ = result_tx.send(outcome);
            if failed {
                // Re-raise so the worker's own counter sees it; the handle already has the details.
                panic::resume_unwind(Box::new("job panicked"));
            }
        });
        JobHandle { result: result_rx }
    }

    /// (completed, panicked) jobs so far.
    pub fn stats(&self) -> (usize, usize) {
        (self.stats.completed.load(Ordering::Relaxed), self.stats.panicked.load(Ordering::Relaxed))
    }
}

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

// ---------- main.rs ----------

#[derive(Debug, Clone)]
struct Order {
    id: u32,
    sku: &'static str,
    quantity: u32,
    unit_cents: u64,
}

// Pretend each check calls an inventory service.
fn validate(order: Order) -> Result<u64, String> {
    thread::sleep(Duration::from_millis(40));
    if order.quantity == 0 {
        return Err(format!("order {}: quantity is zero", order.id));
    }
    if order.sku.is_empty() {
        panic!("order {} has no SKU", order.id); // a bug, not a validation error
    }
    Ok(order.quantity as u64 * order.unit_cents)
}

fn main() {
    panic::set_hook(Box::new(|_| {}));

    let orders = vec![
        Order { id: 1, sku: "LAMP-7", quantity: 2, unit_cents: 4_500 },
        Order { id: 2, sku: "MUG-1", quantity: 0, unit_cents: 900 },
        Order { id: 3, sku: "DESK-2", quantity: 1, unit_cents: 32_000 },
        Order { id: 4, sku: "", quantity: 3, unit_cents: 100 },
        Order { id: 5, sku: "PEN-9", quantity: 10, unit_cents: 150 },
        Order { id: 6, sku: "CHAIR-4", quantity: 4, unit_cents: 12_500 },
        Order { id: 7, sku: "LAMP-7", quantity: 1, unit_cents: 4_500 },
        Order { id: 8, sku: "MUG-1", quantity: 6, unit_cents: 900 },
    ];

    let start = Instant::now();
    let pool = ThreadPool::new(4);
    let (log_tx, log_rx) = mpsc::channel();
    pool.execute(move || log_tx.send(String::from("audit log: batch started")).unwrap());

    // Submit everything first, then wait: the handles keep the original order.
    let handles: Vec<(u32, JobHandle<Result<u64, String>>)> =
        orders.iter().cloned().map(|order| (order.id, pool.submit(move || validate(order)))).collect();

    let mut total_cents = 0;
    for (id, handle) in handles {
        match handle.join() {
            Ok(Ok(cents)) => {
                total_cents += cents;
                println!("order {id}: ok, {cents} cents");
            }
            Ok(Err(reason)) => println!("order {id}: rejected, {reason}"),
            Err(JobPanicked(message)) => println!("order {id}: crashed, {message}"),
        }
    }
    println!("{}", log_rx.recv().unwrap());
    println!("total accepted: {total_cents} cents");

    drop(pool); // wait for any remaining jobs before reading the final numbers
    println!("8 jobs of 40 ms took under 200 ms: {}", start.elapsed() < Duration::from_millis(200));
}

Output

order 1: ok, 9000 cents
order 2: rejected, order 2: quantity is zero
order 3: ok, 32000 cents
order 4: crashed, order 4 has no SKU
order 5: ok, 1500 cents
order 6: ok, 50000 cents
order 7: ok, 4500 cents
order 8: ok, 5400 cents
audit log: batch started
total accepted: 102400 cents
8 jobs of 40 ms took under 200 ms: true

Eight 40 ms jobs on four workers take about 80 ms instead of 320 ms. A validation failure (Ok(Err(...))) and a crash (Err(JobPanicked)) are reported differently, as they should be: one is bad input, the other a bug.

Ideas for extending it

A bounded queue with mpsc::sync_channel(capacity), so execute blocks when too much work is waiting (back pressure), like a ThreadPoolExecutor with an ArrayBlockingQueue.
shutdown_now() that drops queued jobs, using a shared AtomicBool checked before each job.
Work stealing, where each worker has its own queue and idle workers take from busy ones. The crossbeam-deque crate provides the queues; this is how rayon schedules work.
Replace Arc<Mutex<Receiver>> with a crossbeam-channel receiver, which can be cloned and shared without a lock.

Summary

A job is a Box<dyn FnOnce() + Send + 'static>; workers share one channel receiver behind Arc<Mutex<...>> and hold the lock only while receiving.
Dropping the sender is the shutdown signal: workers drain the queue, recv() fails, and Drop joins them. Option::take moves fields out in drop.
submit<T> hides the result type inside the job closure and hands back a receiver for the result.
catch_unwind keeps a panicking job from killing its worker, and the panic becomes an Err for the caller.
For production code, rayon (data parallelism) and tokio::task::spawn_blocking (blocking work in async code) are the standard choices.

Next: Full-Text Search Engine.

Step 1: Workers pulling jobs from a queue

Step 2: Graceful shutdown in Drop

Step 3: Getting results back

Step 4: Surviving a panicking job

Step 5: The complete pool

Ideas for extending it

Summary

Step 2: Graceful shutdown in `Drop`