ThreadPool

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.

Örnek çözümü göster

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);
}

Kendini test et

İşç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
- Panic olur
- Derlenmez
- Önemli değil
Cevabı göster

Kilit iş çalışırken tutulursa diğer işçiler iş alamaz ve havuz tek thread gibi çalışır — 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.

Göndericiyi bırak

impl Drop for Havuz {
    fn drop(&mut self) {
        drop(self.gonderici.____());
        for i in self.iscler.drain(..) {
            i.join().unwrap();
        }
    }
}

Cevabı göster

impl Drop for Havuz {
    fn drop(&mut self) {
        drop(self.gonderici.take());
        for i in self.iscler.drain(..) {
            i.join().unwrap();
        }
    }
}

Parça ve görevi

Box<dyn FnOnce() + Send>

İş türü
Arc<Mutex<Receiver>>

Paylaşılan alıcı
recv() → Err

Kanal kapandı
Drop

Düzgün kapanış

Sıradaki ders: derive macro

Deneyiminizi Geliştiriyoruz

Çerez Tercihleri

Kesinlikle Gerekli Çerezler

Analitik Çerezler