ThreadPool
- Rust course
- Advanced
- Lesson 79/81
- 7 min
Build your own ThreadPool with Mutex and mpsc.
Final task: ThreadPool
Opening a new thread for each request is expensive. A thread pool starts a fixed number of workers and distributes jobs through a channel. The building blocks: an mpsc channel, Arc<Mutex<Receiver>> to share the receiver, and Box<dyn FnOnce() + Send> as the job type:
type Is = Box<dyn FnOnce() + Send + 'static>;
struct Havuz {
iscler: Vec<thread::JoinHandle<()>>,
gonderici: Option<mpsc::Sender<Is>>,
}The Option is needed to drop the sender during Drop and tell the workers "the work is over".
Parts of the pool
Workers pull jobs from the receiver and leave the loop when the channel closes. Drop drops the sender and waits for all workers:
impl Havuz {
fn new(n: usize) -> Self {
let (tx, rx) = mpsc::channel::<Is>();
let rx = Arc::new(Mutex::new(rx));
let iscler = (0..n).map(|_| {
let rx = Arc::clone(&rx);
thread::spawn(move || loop {
let is = rx.lock().unwrap().recv();
match is {
Ok(f) => f(),
Err(_) => break, // the channel closed
}
})
}).collect();
Havuz { iscler, gonderici: Some(tx) }
}
fn calistir<F: FnOnce() + Send + 'static>(&self, f: F) {
self.gonderici.as_ref().unwrap().send(Box::new(f)).unwrap();
}
}Practice Thread pool
Write the Havuz type above with new, calistir and Drop. Give a 2-worker pool 4 jobs, each sending i * i to a separate channel. Drop the pool, collect the results, sort and print them.
Starter code
use std::sync::{mpsc, Arc, Mutex};
use std::thread;
fn main() {
// buraya yaz
}Sample solution
use std::sync::{mpsc, Arc, Mutex};
use std::thread;
type Is = Box<dyn FnOnce() + Send + 'static>;
struct Havuz {
iscler: Vec<thread::JoinHandle<()>>,
gonderici: Option<mpsc::Sender<Is>>,
}
impl Havuz {
fn new(n: usize) -> Self {
let (tx, rx) = mpsc::channel::<Is>();
let rx = Arc::new(Mutex::new(rx));
let iscler = (0..n)
.map(|_| {
let rx = Arc::clone(&rx);
thread::spawn(move || loop {
let is = rx.lock().unwrap().recv();
match is {
Ok(f) => f(),
Err(_) => break,
}
})
})
.collect();
Havuz { iscler, gonderici: Some(tx) }
}
fn calistir<F: FnOnce() + Send + 'static>(&self, f: F) {
self.gonderici.as_ref().unwrap().send(Box::new(f)).unwrap();
}
}
impl Drop for Havuz {
fn drop(&mut self) {
drop(self.gonderici.take());
for i in self.iscler.drain(..) {
i.join().unwrap();
}
}
}
fn main() {
let (tx, rx) = mpsc::channel();
let havuz = Havuz::new(2);
for i in 0..4 {
let tx = tx.clone();
havuz.calistir(move || tx.send(i * i).unwrap());
}
drop(havuz);
drop(tx);
let mut sonuc: Vec<i32> = rx.iter().collect();
sonuc.sort();
println!("{:?}", sonuc);
}Expected output
[0, 1, 4, 9]
Questions and answers
Review the lesson questions together with their answers. To test yourself, solve the lesson interactively.
1 Why is it important to release the lock right after <code>recv</code> in the worker's loop?
- If the lock is held while the job runs, other workers can't take jobs and the pool works like a single thread Correct answer
- It panics
- It doesn't compile
- It doesn't matter
Explanation: In let is = rx.lock().unwrap().recv(); the guard is a temporary dropped at the end of the line; writing it in while let instead holds the lock for the whole body.
2 Drop the sender
impl Drop for Havuz {
fn drop(&mut self) {
drop(self.gonderici.____());
for i in self.iscler.drain(..) {
i.join().unwrap();
}
}
}impl Drop for Havuz {
fn drop(&mut self) {
drop(self.gonderici.take());
for i in self.iscler.drain(..) {
i.join().unwrap();
}
}
}Explanation: Option::take takes the sender and leaves None; the channel closes.
3 Piece and role
| Term | Meaning |
|---|---|
Box<dyn FnOnce() + Send> |
Job type |
Arc<Mutex<Receiver>> |
Shared receiver |
recv() → Err |
Channel closed |
Drop |
Clean shutdown |