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 axum::{extract::{Path, State}, http::StatusCode, Json};
use dashmap::DashMap;
use serde::Serialize;
use std::sync::Arc;

Request coalescing in an Axum handler

caching concurrency request-coalescing
Advanced 8 steps
go 
package config
 
import (
	"fmt"

A thread-safe config singleton in Go

singleton concurrency environment-variables
Intermediate 7 steps
rust 
use std::net::Ipv4Addr;
use std::str::FromStr;
 
#[derive(Debug, Clone, Copy)]

Parsing and matching IPv4 CIDR ranges in Rust

bitwise-operations parsing error-handling
Intermediate 8 steps
javascript 
const express = require('express');
const app = express();
 
app.get('/health', (req, res) => res.json({ status: 'ok' }));

Graceful shutdown in an Express server

graceful-shutdown signal-handling connection-tracking
Advanced 9 steps
rust 
use std::sync::OnceLock;
 
static CRC_TABLE: OnceLock<[u32; 256]> = OnceLock::new();
 

Table-driven CRC32 with lazy init in Rust

checksums lazy-initialization lookup-tables
Intermediate 8 steps
rust 
use axum::{
    body::Body,
    extract::Path,
    http::{header, HeaderMap, HeaderValue, StatusCode},

HTTP range requests for video streaming in Axum

http-range-requests streaming async-io
Advanced 8 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