ThreadPool
- Rust kursu
- İleri seviye
- Ders 79/81
- 7 dk
Kendi ThreadPool'unu Mutex ve mpsc ile yazarsın.
Bitirme görevi: ThreadPool
Her istek için yeni thread açmak pahalıdır. İş parçacığı havuzu sabit sayıda işçi başlatır ve işleri bir kanaldan dağıtır. Yapı taşları: mpsc kanalı, alıcıyı paylaşmak için Arc<Mutex<Receiver>> ve iş türü olarak Box<dyn FnOnce() + Send>:
type Is = Box<dyn FnOnce() + Send + 'static>;
struct Havuz {
iscler: Vec<thread::JoinHandle<()>>,
gonderici: Option<mpsc::Sender<Is>>,
}Option, Drop sırasında göndericiyi bırakıp işçilere "iş bitti" demek için gereklidir.
Havuzun parçaları
İşçiler alıcıdan iş çeker; kanal kapanınca döngüden çıkar. Drop göndericiyi bırakır ve tüm işçileri bekler:
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, // kanal kapandı
}
})
}).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();
}
}Uygulama İş parçacığı havuzu
Yukarıdaki Havuz türünü new, calistir ve Drop ile yaz. 2 işçili bir havuza, her biri i * i sonucunu ayrı bir kanala gönderen 4 iş ver. Havuzu bırak, sonuçları topla, sırala ve yazdır.
Başlangıç kodu
use std::sync::{mpsc, Arc, Mutex};
use std::thread;
fn main() {
// buraya yaz
}Örnek çözüm
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);
}Beklenen çıktı
[0, 1, 4, 9]
Soru ve cevaplar
Dersin sorularını cevaplarıyla birlikte inceleyebilirsin. Kendini denemek istersen dersi interaktif olarak çöz.
1 İşçinin döngüsünde kilidi <code>recv</code> sonrasında hemen bırakmak neden önemlidir?
- Kilit iş çalışırken tutulursa diğer işçiler iş alamaz ve havuz tek thread gibi çalışır Doğru cevap
- Panic olur
- Derlenmez
- Önemli değil
Açıklama: let is = rx.lock().unwrap().recv(); satırında guard geçici değerdir ve satır sonunda bırakılır; while let içinde yazmak ise kilidi tüm gövde boyunca tutar.
2 Göndericiyi bırak
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();
}
}
}Açıklama: Option::take göndericiyi alıp yerine None bırakır; kanal kapanır.
3 Parça ve görevi
| Kavram | Anlamı |
|---|---|
Box<dyn FnOnce() + Send> |
İş türü |
Arc<Mutex<Receiver>> |
Paylaşılan alıcı |
recv() → Err |
Kanal kapandı |
Drop |
Düzgün kapanış |