🎓 A1Learner
🔍

Rust Thread Pools

A thread pool is a fixed team of worker threads. A pool pays the cost of starting threads once and reuses its threads for every job.

The cost: a new thread for every job

Without a pool, each job spawns a fresh thread. Each spawn reserves about a megabyte of stack:

fn main() {
    let mut handles = Vec::new();
    for n in 1..=6 {
        handles.push(std::thread::spawn(move || n * n));
    }
    let mut squares = Vec::new();
    for h in handles {
        squares.push(h.join().unwrap());
    }
    squares.sort_unstable();
    println!("squares = {:?}", squares);
}

A job is a boxed closure

Any closure the caller hands in becomes one uniform type the channel can carry:

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

One receiver, shared behind a Mutex

A channel receiver has a single owner, so the workers share it. The Mutex is held only for the recv call itself. It is never held while a job runs:

let (sender, receiver) = mpsc::channel::<Job>();
let receiver = Arc::new(Mutex::new(receiver));
// every worker clones the Arc, locks the receiver, takes the next job

A worker is a loop around recv

thread::spawn(move || loop {
    let job = receiver.lock().unwrap().recv();
    match job {
        Ok(job) => job(),
        Err(_) => break, // the sender is gone: the pool is shutting down
    }
})

Drop closes the channel first, then joins

Both fields are Option<...> precisely so Drop can take them out and spend them, in this order:

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take()); // step 1: no job can ever arrive again
        for worker in &mut self.workers {
            if let Some(handle) = worker.thread.take() {
                handle.join().unwrap(); // step 2: wait for each worker to finish
            }
        }
    }
}

Example

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

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

struct Worker {
    thread: Option<thread::JoinHandle<()>>,
}

struct ThreadPool {
    workers: Vec<Worker>,
    sender: Option<mpsc::Sender<Job>>,
}

impl ThreadPool {
    fn new(size: usize) -> ThreadPool {
        assert!(size > 0);
        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));
        let mut workers = Vec::with_capacity(size);
        for _ in 0..size {
            let receiver = Arc::clone(&receiver);
            let handle = thread::spawn(move || loop {
                let job = receiver.lock().unwrap().recv();
                match job {
                    Ok(job) => job(),
                    Err(_) => break,
                }
            });
            workers.push(Worker { thread: Some(handle) });
        }
        ThreadPool { workers, sender: Some(sender) }
    }

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

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.take());
        for worker in &mut self.workers {
            if let Some(handle) = worker.thread.take() {
                handle.join().unwrap();
            }
        }
    }
}

fn main() {
    let (results_tx, results_rx) = mpsc::channel();
    {
        let pool = ThreadPool::new(3);
        for n in 1..=6 {
            let tx = results_tx.clone();
            pool.execute(move || {
                tx.send(n * n).unwrap();
            });
        }
    }
    drop(results_tx);
    let mut squares: Vec<i32> = results_rx.iter().collect();
    squares.sort_unstable();
    println!("squares = {:?}", squares);
    println!("jobs done = {}", squares.len());
    println!("pool shut down cleanly");
}

Key points

  • A pool owns its threads.
  • ThreadPool::new(4) starts four workers up front.
  • It reuses them for every job.
  • No thread is created per job.
  • execute(closure) boxes the closure and sends it down the channel.
  • Whichever worker is free takes it.
  • The worker runs it, then loops back for more.
  • Sizing: pools are small and fixed on purpose.
  • thread::available_parallelism() reports how many threads the machine can truly run at once.
  • That count is a sane default size.
  • Graceful shutdown has two steps.
  • The order is the whole design.
  • Close the job channel first.
  • Workers then exit their loop after their current job.
  • Then join every worker.
  • Joining first would wait forever on workers who are themselves waiting for jobs
  • A job that panics kills its worker thread.
  • The pool silently shrinks by one.
  • Production pools guard jobs against panics.
  • Or they notice and respawn the worker.
  • std ships no ready-made pool.
  • This pattern (about 60 lines) is the standard one.
  • It comes from the Rust Book's final project.
  • The rayon crate builds a data-parallel pool on the same ideas.
  • It fits jobs like "apply this to every element".
  • A thread pool keeps a fixed team of workers and hands each new job to a free worker.

In this example, ThreadPool::new(3) starts three workers. All three share one job receiver behind an Arc<Mutex<...>>. The Mutex exists only because a receiver has one owner. It is held just for the recv call. It is never held while a job runs. Six jobs go in. Each computes a square. Each sends its square back on a separate results channel. Then main leaves the block and the pool drops. Drop first takes and drops the sender. That closes the job channel. Each worker finishes its current job. Then it sees recv fail and breaks its loop. Only then does Drop join the three threads. That order is the whole secret of graceful shutdown. The results are collected and sorted. The output is identical on every run. It shows all six squares and six jobs done. It ends with a clean shutdown.

Note

This pool fits requests.

Try it Yourself

Edit the code, then press Run.

Running is simulated · Rust Playground

Exercise

Why must the pool's Drop drop the sender before it joins the workers?