A thread pool

A thread pool is a fixed set of workers that share one queue of jobs. The caller sends closures, the workers take them one at a time, and the pool shuts down by closing the queue and joining every worker. Building one uses the two patterns of the chapter together: a channel for the handover and a lock around the receiver.

This lesson starts with one worker over a channel, gives the receiver to several workers through an Arc, wraps the whole thing in a struct whose destructor shuts it down, and ends with the compiler error for a receiver that was moved into the first worker.

One worker and a queue

The queue is an ordinary channel of jobs. The worker loops over the receiver, does the work and returns the accumulated result when the channel closes, which happens when the sender is dropped.

The program sends three numbers, closes the queue and prints the sum the worker computed. The join waits for the last job.

use std::sync::mpsc; use std::thread; fn main() { let (jobs, work) = mpsc::channel::<u32>(); let worker = thread::spawn(move || { let mut sum = 0; for job in work { sum += job; } sum }); for job in [1, 2, 3] { jobs.send(job).unwrap(); } drop(jobs); println!("{}", worker.join().unwrap()); }

rustc --edition 2024 one.rs
./one

6

for job in work ends when every sender is gone, which is the shutdown condition of the whole design. The lesson on channels explains the close that the drop causes.

Several workers on one queue

A receiver cannot be cloned, so the workers share one through an Arc<Mutex<Receiver>>. Each worker takes the lock, receives one job, releases the lock and runs the job; the lock is held only for the receive.

Three workers square six jobs and report through a second channel. The results are collected and sorted, so the output does not depend on which worker took which job.

use std::sync::{mpsc, Arc, Mutex}; use std::thread; fn main() { let (jobs, work) = mpsc::channel::<u32>(); let work = Arc::new(Mutex::new(work)); let (results, output) = mpsc::channel::<u32>(); let mut workers = Vec::new(); for _ in 0..3 { let work = Arc::clone(&work); let results = results.clone(); workers.push(thread::spawn(move || { loop { let job = { let guard = work.lock().unwrap(); guard.recv() }; match job { Ok(value) => results.send(value * value).unwrap(), Err(_) => break, } } })); } drop(results); for job in 1..=6 { jobs.send(job).unwrap(); } drop(jobs); for worker in workers { worker.join().unwrap(); } let mut values: Vec<u32> = output.iter().collect(); values.sort(); println!("{values:?}"); }

rustc --edition 2024 three.rs
./three

[1, 4, 9, 16, 25, 36]

The inner block around the lock is deliberate: the guard is dropped at its end, before the job runs, so two workers can compute at the same time while only one waits on the queue. The lesson on Mutex and Arc is the pair of types, and the previous lesson explains why the scope cannot replace them here: the pool outlives the calls that use it.

A pool with a shutdown

A pool that a program keeps for its whole life is a struct: a sender and a vector of handles. The destructor drops the sender, which closes the queue, and joins every worker, which is the shutdown in two lines.

The job type is a boxed closure that is Send and 'static, which is what a queue of work needs. The program runs three jobs that report to a channel, drops the pool and prints the sorted results.

use std::sync::{mpsc, Arc, Mutex}; use std::thread::{self, JoinHandle}; type Job = Box<dyn FnOnce() + Send + 'static>; struct Pool { sender: Option<mpsc::Sender<Job>>, workers: Vec<JoinHandle<()>>, } impl Pool { fn new(size: usize) -> Self { let (sender, receiver) = mpsc::channel::<Job>(); let receiver = Arc::new(Mutex::new(receiver)); let mut workers = Vec::new(); for _ in 0..size { let receiver = Arc::clone(&receiver); workers.push(thread::spawn(move || { while let Ok(job) = receiver.lock().unwrap().recv() { job(); } })); } Pool { sender: Some(sender), workers } } fn execute(&self, job: impl FnOnce() + Send + 'static) { self.sender.as_ref().unwrap().send(Box::new(job)).unwrap(); } } impl Drop for Pool { fn drop(&mut self) { drop(self.sender.take()); for worker in self.workers.drain(..) { worker.join().unwrap(); } } } fn main() { let (sender, receiver) = mpsc::channel(); let pool = Pool::new(2); for value in [1, 2, 3] { let sender = sender.clone(); pool.execute(move || sender.send(value * 10).unwrap()); } drop(sender); drop(pool); let mut values: Vec<i32> = receiver.iter().collect(); values.sort(); println!("{values:?}"); }

rustc --edition 2024 pool.rs
./pool

[10, 20, 30]

drop(pool) is the shutdown: the destructor closes the queue and waits for the workers, so every job has run before the receiver is drained. The lesson on Drop covers the destructor that does the work, and a job that panics poisons the receiver lock, which the mutex lesson shows how to survive.

A pool is also the place to put the queue's bound: a sender that blocks when the workers are behind is the back pressure of the previous lesson, and a program that would rather fail than wait uses try_send and reports the busy state to its caller.

A job that panics takes its worker with it, and the panic poisons the lock around the receiver, so the other workers stop at their next receive. A pool for a long-running service therefore catches panics inside the job with catch_unwind, or gives each job its own task, which is what the async runtime of a later lesson does.

The size of the pool is a decision: one worker makes the order of the jobs predictable, one per core uses the machine, and more than that only helps when the jobs block. A pool of zero has no receiver left, so the first execute fails and the unwrap in the method panics on the spot.

A receiver moved into one worker

The receiver is not cloneable, so the obvious version of two workers does not compile: the first spawn takes it and the second has nothing to take. The compiler reports E0382 and names the type that has no Copy.

The program gives the receiver to the first thread and then to the second. The compiler stops at the second closure.

use std::sync::mpsc; use std::thread; fn main() { let (_jobs, work) = mpsc::channel::<u32>(); let first = thread::spawn(move || work.recv().unwrap()); let second = thread::spawn(move || work.recv().unwrap()); first.join().unwrap(); second.join().unwrap(); }

rustc --edition 2024 moved_err.rs

error[E0382]: use of moved value: `work`
--> moved_err.rs:7:32
|
5 | let (_jobs, work) = mpsc::channel::<u32>();
| ---- move occurs because `work` has type `std::sync::mpsc::Receiver<u32>`, which does not implement the `Copy` trait
6 | let first = thread::spawn(move || work.recv().unwrap());
| ------- ---- variable moved due to use in closure
| |
| value moved into closure here
7 | let second = thread::spawn(move || work.recv().unwrap());
| ^^^^^^^ ---- use occurs due to use in closure
| |
| value used here after move

error: aborting due to 1 previous error

For more information about this error, try `rustc --explain E0382`.

The fix is the Arc<Mutex<Receiver>> of the second section, and the error is the reason the pool needs a lock at all. The next lesson adds a bound to the same queue and a reader-writer lock beside it.

Exercises

  1. Give the pool a join method and call it before the destructor runs.
  2. Send jobs that compute a prime check and return a boolean through a results channel.
  3. Make a job panic and find out what happens to the pool on the next job.

Questions

A new set of questions in a new order every time. Choose one answer for each, then check.

  1. What does the worker loop end on?

    Answer

    The channel closing when every sender is dropped

    The receive returns an error once no sender exists, and the loop breaks.

  2. Why does the shared receiver need a Mutex?

    Answer

    A Receiver cannot be cloned, so the workers take turns receiving

    One receiver is shared, and the lock guarantees one receive at a time.

  3. What does the destructor of the pool do?

    Answer

    Closes the queue and joins the workers

    It drops the sender and joins every handle, so all jobs finish first.

  4. What is the job type of the pool?

    Answer

    A boxed closure that is Send and lives for the whole program

    The queue holds trait objects, so each job is boxed and must be Send.

  5. What does this program print? use std::sync::mpsc; use std::thread; fn main() { let (jobs, work) = mpsc::channel::<u32>(); let worker = thread::spawn(move || { let mut count = 0; for _ in work { count += 1; } count }); for value in 1..=4 { jobs.send(value).unwrap(); } drop(jobs); println!("{}", worker.join().unwrap()); }

    Answer

    4

    The worker counts four jobs and the drop closes the queue.

  6. What does this program print? use std::sync::mpsc; use std::thread; fn main() { let (jobs, work) = mpsc::channel::<i32>(); let worker = thread::spawn(move || { let mut sum = 0; for value in work { sum += value; } sum }); jobs.send(2).unwrap(); jobs.send(3).unwrap(); drop(jobs); println!("{}", worker.join().unwrap()); }

    Answer

    5

    The two values are added by the worker after the channel closes.

  7. What does this program print? use std::sync::mpsc; fn main() { let (jobs, work) = mpsc::channel::<i32>(); jobs.send(7).unwrap(); drop(jobs); let values: Vec<i32> = work.iter().collect(); println!("{values:?}"); }

    Answer

    [7]

    The value is queued before the close, and the iterator drains it.

  8. What does this program print? use std::sync::{Arc, Mutex}; fn main() { let value = Arc::new(Mutex::new(0)); { let mut guard = value.lock().unwrap(); *guard += 2; } println!("{}", value.lock().unwrap()); }

    Answer

    2

    The block ends the lock before the print takes it again.

  9. What is the first line of the compiler output for this program? use std::sync::mpsc; use std::thread; fn main() { let (_jobs, work) = mpsc::channel::<u32>(); let first = thread::spawn(move || work.recv().unwrap()); let second = thread::spawn(move || work.recv().unwrap()); first.join().unwrap(); second.join().unwrap(); }

    Answer

    error[E0382]: use of moved value: work

    The first closure takes the receiver by move and the second cannot use it.

  10. What is the first line of the compiler output for this program? use std::sync::mpsc; fn main() { let (jobs, work) = mpsc::channel::<i32>(); let copy = work.clone(); println!("{copy:?}"); drop(jobs); }

    Answer

    error[E0599]: no method named clone found for struct std::sync::mpsc::Receiver<T> in the current scope

    The receiver has no clone, which is why a pool wraps it in a mutex.

The code on this page was built with rustc 1.96.1 and run on Alpine Linux 3.24.3 (x86_64) in a Docker container.

Article author: Arthur Isaev