Rust
ThreadPool
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();
}
}
Exercise 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.
Show example 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);
}
Test yourself
-
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
- It panics
- It doesn't compile
- It doesn't matter
Show answer
If the lock is held while the job runs, other workers can't take jobs and the pool works like a single thread — 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.
-
Drop the sender
impl Drop for Havuz {
fn drop(&mut self) {
drop(self.gonderici.____());
for i in self.iscler.drain(..) {
i.join().unwrap();
}
}
}
Show answer
impl Drop for Havuz {
fn drop(&mut self) {
drop(self.gonderici.take());
for i in self.iscler.drain(..) {
i.join().unwrap();
}
}
}
-
Piece and role
- Box<dyn FnOnce() + Send>
- Job type
- Arc<Mutex<Receiver>>
- Shared receiver
- recv() → Err
- Channel closed
- Drop
- Clean shutdown
Next lesson: derive macro
