use std::sync::mpsc::{self, SyncSender};
use std::sync::{Arc, Mutex};
use std::thread;
mod worker;
use worker::{Message, Worker};
pub type Job = Box<dyn FnOnce() + Send + 'static>;
pub struct ThreadPool {
workers: Vec<Worker>,
sender: SyncSender<Message>,
}
impl ThreadPool {
pub fn new(size: usize, capacity: usize) -> ThreadPool {
assert!(size > 0, "pool needs at least one worker");
let (sender, receiver) = mpsc::sync_channel::<Message>(capacity);
let receiver = Arc::new(Mutex::new(receiver));
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, Arc::clone(&receiver)));
}
ThreadPool { workers, sender }
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
// Blocks once the bounded buffer is full: this is the backpressure.
self.sender
.send(Message::NewJob(Box::new(f)))
.expect("all workers have stopped");
}
}
impl Drop for ThreadPool {
fn drop(&mut self) {
for _ in &self.workers {
self.sender.send(Message::Terminate).ok();
}
for worker in &mut self.workers {
if let Some(handle) = worker.handle.take() {
handle.join().expect("worker thread panicked");
}
}
}
}
use std::sync::mpsc::Receiver;
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use crate::Job;
pub enum Message {
NewJob(Job),
Terminate,
}
pub struct Worker {
pub id: usize,
pub handle: Option<JoinHandle<()>>,
}
impl Worker {
pub fn new(id: usize, receiver: Arc<Mutex<Receiver<Message>>>) -> Worker {
let handle = thread::spawn(move || loop {
let message = {
let guard = receiver.lock().expect("receiver mutex poisoned");
guard.recv()
};
match message {
Ok(Message::NewJob(job)) => {
job();
}
Ok(Message::Terminate) => {
break;
}
Err(_) => {
// Senders dropped: treat as shutdown.
break;
}
}
});
Worker {
id,
handle: Some(handle),
}
}
}
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::Duration;
use pool::ThreadPool;
fn main() {
// 4 workers, queue capacity 2 -> at most 6 tasks alive before producer parks.
let pool = ThreadPool::new(4, 2);
let completed = Arc::new(AtomicUsize::new(0));
for task_id in 0..11 {
let done = Arc::clone(&completed);
pool.execute(move || {
thread::sleep(Duration::from_millis(50));
let n = done.fetch_add(1, Ordering::SeqCst) + 1;
println!("task {} finished (total {})", task_id, n);
});
println!("submitted task {}", task_id);
}
// Dropping the pool sends Terminate to every worker and joins them.
drop(pool);
println!("all done: {}", completed.load(Ordering::SeqCst));
}
This snippet builds a fixed-size thread pool whose task queue is bounded, so submitting work applies backpressure instead of letting an unbounded backlog grow without limit. The core idea is that a SyncSender with a small buffer will block the producer once the buffer fills, which naturally throttles fast producers to the rate the workers can drain. That property is what makes the pool safe under load: memory stays bounded and slow consumers push back on the caller.
In ThreadPool, the constructor creates a mpsc::sync_channel(capacity) and wraps the Receiver in an Arc<Mutex<..>> so every worker can share the single receive end. Standard-library mpsc is multi-producer but single-consumer, so the Mutex serializes the recv() calls across the worker threads, effectively emulating an MPMC queue. Each spawned worker loops, locks the receiver just long enough to pull one Job, releases the lock, then runs the closure outside the lock so long tasks never block other workers from dequeuing.
Jobs are typed as Box<dyn FnOnce() + Send + 'static>, the canonical boxed-closure shape for a work queue: FnOnce because a job runs exactly once, Send so it can cross the thread boundary, and 'static so it owns its captures. execute sends over the SyncSender; when the buffer is full the send blocks, which is the backpressure in action.
Shutdown is handled by the Message enum in worker.rs. Rather than relying on channel disconnection alone, Drop for ThreadPool sends one Message::Terminate per worker, then joins each JoinHandle. Sending exactly N terminate messages guarantees every worker observes a stop signal even while others are mid-task, and the take() on the optional handle makes the join idempotent. A worker treats a recv() error as an implicit terminate, covering the case where senders are dropped.
The main.rs tab shows the pool draining a batch of jobs and exercising the bound: with capacity two, the eleventh in-flight submission parks the producer until a worker frees a slot. Trade-offs to note: the shared Mutex is a mild contention point under very high throughput, where a dedicated crossbeam MPMC channel would scale better; and panicking jobs currently kill their worker thread, so production code often wraps the call in catch_unwind. This pattern fits CPU-bound or I/O batch work where a stable, bounded number of threads must be reused.
Related snips
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
package pools
import (
"bytes"
"sync"
)sync.Pool for bytes.Buffer to reduce allocations in hot paths
# app/channels/chat_channel.rb
class ChatChannel < ApplicationCable::Channel
def subscribed
# Subscribe to a specific room
room = Room.find(params[:room_id])ActionCable for real-time WebSocket communication
Share this code
Here's the card — post it anywhere.
