rust 51 lines · 9 steps

Building a thread pool in Rust

A fixed set of worker threads share one job channel, executing closures as they arrive and shutting down cleanly.

Explained by highlit
Advanced concurrency channels thread-pool ownership graceful-shutdown
1use std::sync::{mpsc, Arc, Mutex};
2use std::thread;
3use std::time::Duration;
4 
5type Job = Box<dyn FnOnce() + Send + 'static>;
6 
7pub struct WorkerPool {
8 workers: Vec<thread::JoinHandle<()>>,
9 sender: Option<mpsc::Sender<Job>>,
10}
11 
12impl WorkerPool {
13 pub fn new(size: usize) -> Self {
14 let (sender, receiver) = mpsc::channel::<Job>();
15 let receiver = Arc::new(Mutex::new(receiver));
16 let mut workers = Vec::with_capacity(size);
17 
18 for id in 0..size {
19 let receiver = Arc::clone(&receiver);
20 workers.push(thread::spawn(move || loop {
21 let job = match receiver.lock().unwrap().recv() {
22 Ok(job) => job,
23 Err(_) => break,
24 };
25 eprintln!("worker {id} picked up a job");
26 job();
27 }));
28 }
29 
30 WorkerPool { workers, sender: Some(sender) }
31 }
32 
33 pub fn submit<F>(&self, task: F)
34 where
35 F: FnOnce() + Send + 'static,
36 {
37 if let Some(sender) = &self.sender {
38 sender.send(Box::new(task)).expect("pool has shut down");
39 }
40 }
41}
42 
43impl Drop for WorkerPool {
44 fn drop(&mut self) {
45 drop(self.sender.take());
46 for worker in self.workers.drain(..) {
47 let _ = worker.join();
48 }
49 thread::sleep(Duration::from_millis(1));
50 }
51}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Wrapping a channel receiver in Arc<Mutex<...>> lets many threads pull from one shared work queue safely.
  2. 2Boxing closures as a trait object turns arbitrary tasks into a single sendable job type.
  3. 3Dropping the sender closes the channel, which is the signal worker loops use to exit.

Related explainers

rust 
use serde::Deserialize;
 
#[derive(Debug, Deserialize)]
#[serde(untagged)]

Parsing flexible JSON shapes with serde

deserialization enums json
Intermediate 6 steps
go 
package streaming
 
import (
	"bufio"

Streaming NDJSON logs over HTTP in Go

http-streaming channels select
Advanced 10 steps
rust 
use std::cmp::{Ordering, Reverse};
use std::collections::BinaryHeap;
use std::fs::File;
use std::io::{self, BufRead, BufReader, BufWriter, Lines, Write};

K-way merge of sorted logs in Rust

binary-heap k-way-merge streaming-io
Intermediate 8 steps
rust 
use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel
Advanced 7 steps
rust 
use std::f64::consts::PI;
 
#[derive(Debug, Clone, Copy)]
pub struct LatLng {

Building geographic bounding boxes in Rust

geospatial value-types option
Intermediate 7 steps
go 
func (w *Watcher) resetDebounce(d time.Duration) {
	if !w.timer.Stop() {
		select {
		case <-w.timer.C:

Debouncing a stream of events in Go

debounce timers channels
Advanced 7 steps

Share this explainer

Here's the card — post it anywhere.

Building a thread pool in Rust — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code