Rust Programlama Dili

Tek İş Parçacıklı Sunucudan Çok İş Parçacıklı Sunucuya Geçiş

Şu anda sunucu her isteği sırayla işliyor; yani ilk bağlantının işlenmesi tamamlanmadan ikinci bir bağlantıyı işlemeye geçmiyor. Sunucu giderek daha fazla istek aldığında, bu sıralı çalıştırma biçimi gitgide daha verimsiz hâle gelecektir. Sunucu işlenmesi uzun süren bir istek alırsa, sonradan gelen yeni istekler hızlıca işlenebilecek olsa bile uzun isteğin bitmesini beklemek zorunda kalacaktır. Bu sorunu çözmemiz gerekecek; ancak önce sorunu çalışırken gözlemleyelim.

Yavaş Bir İsteği Taklit Etmek

Yavaş işlenen bir isteğin, mevcut sunucu uygulamamıza yapılan diğer istekleri nasıl etkileyebileceğine bakalım. Liste 21-10, sunucunun yanıt vermeden önce beş saniye uyumasına neden olacak yapay bir yavaş yanıtla /sleep yoluna gelen bir isteğin işlenmesini uygular.

use std::{
    fs,
    io::{BufReader, prelude::*},
    net::{TcpListener, TcpStream},
    thread,
    time::Duration,
};
// --snip--

fn main() {
    let dinleyici = TcpListener::bind("127.0.0.1:7878").unwrap();

    for akis in dinleyici.incoming() {
        let akis = akis.unwrap();

        baglantiyi_isle(akis);
    }
}

fn baglantiyi_isle(mut akis: TcpStream) {
    // --snip--

    let tamponlu_okuyucu = BufReader::new(&akis);
    let istek_satiri = tamponlu_okuyucu.lines().next().unwrap().unwrap();

    let (durum_satiri, dosya_adi) = match &istek_satiri[..] {
        "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "merhaba.html"),
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));
            ("HTTP/1.1 200 OK", "merhaba.html")
        }
        _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
    };

    // --snip--

    let icerik = fs::read_to_string(dosya_adi).unwrap();
    let uzunluk = icerik.len();

    let yanit =
        format!("{durum_satiri}\r\nContent-Length: {uzunluk}\r\n\r\n{icerik}");

    akis.write_all(yanit.as_bytes()).unwrap();
}

Artık üç durumumuz olduğu için if yapısından match yapısına geçtik. String sabiti (string literal) değerleriyle desen eşleştirmesi yapabilmek için istek_satiri değişkeninin bir dilimi (&istek_satiri[..]) üzerinde açıkça eşleştirme yapmamız gerekir: istek_satiri bir String, desenlerdeki string sabitleri ise &str türündedir ve match, eşitlik operatörünün (==) yaptığı gibi otomatik referans alma ve başvuru çözme (dereferencing) işlemi yapmaz.

İlk kol, Liste 21-9’daki if bloğuyla aynıdır. İkinci kol, /sleep yoluna yapılan bir istekle eşleşir. Bu istek alındığında sunucu, başarılı HTML sayfasını oluşturup göndermeden önce beş saniye uyuyacaktır. Üçüncü kol ise Liste 21-9’daki else bloğuyla aynıdır.

Sunucumuzun ne kadar ilkel olduğunu görebilirsiniz: Gerçek kütüphaneler farklı isteklerin tanınmasını çok daha özlü bir biçimde, çok daha az kodla ele alır!

cargo run kullanarak sunucuyu başlatın. Ardından iki tarayıcı penceresi açın: biri http://127.0.0.1:7878, diğeri ise http://127.0.0.1:7878/sleep için olsun. Daha önce yaptığınız gibi / URI adresini birkaç kez girerseniz hızlıca yanıt verdiğini görürsünüz. Ancak önce /sleep adresine girip hemen ardından / sayfasını yüklerseniz, / sayfasının yüklenmeden önce sleep çağrısının beş saniyelik uykusunu tamamen bitirmesini beklediğini görürsünüz.

Yavaş bir isteğin arkasında diğer isteklerin birikmesini önlemek için Bölüm 17’de yaptığımız gibi async kullanmak da dahil olmak üzere başvurabileceğimiz birden fazla teknik vardır; bizim uygulayacağımız teknik ise bir thread havuzu (thread pool) olacaktır.

Thread Havuzuyla İşleme Kapasitesini Artırmak

Thread havuzu (thread pool), bir görevi ele almaya hazır olarak bekleyen, önceden başlatılmış bir thread grubudur. Program yeni bir görev aldığında, havuzdaki thread’lerden birini bu göreve atar ve o thread görevi işler. Havuzda kalan diğer thread’ler, ilk thread çalışmaya devam ederken gelen başka görevleri ele almak üzere hazırdır. İlk thread görevini işlemeyi bitirdiğinde, yeni bir görevi üstlenmeye hazır biçimde boşta bekleyen thread’ler havuzuna geri döner. Bir thread havuzu, bağlantıları eşzamanlı olarak işlemenize olanak tanıyarak sunucunuzun işleme kapasitesini (throughput) artırır.

Bizi hizmet reddi (Denial of Service — DoS) saldırılarından korumak için havuzdaki thread sayısını küçük bir sayıyla sınırlandıracağız. Programımız gelen her istek için yeni bir thread oluştursaydı, sunucumuza 10 milyon istek gönderen biri sunucumuzun tüm kaynaklarını tüketip isteklerin işlenmesini durma noktasına getirerek büyük bir hasara yol açabilirdi.

Dolayısıyla sınırsız sayıda thread başlatmak yerine, havuzda bekleyen sabit sayıda thread bulunduracağız. Gelen istekler işlenmek üzere havuza gönderilir. Havuz, gelen isteklerden oluşan bir kuyruk tutar. Havuzdaki thread’lerin her biri bu kuyruktan bir istek çeker, isteği işler ve ardından kuyruktan başka bir istek ister. Bu tasarımla en fazla N isteği eşzamanlı olarak işleyebiliriz; burada N thread sayısıdır. Thread’lerin her biri uzun süren bir isteğe yanıt veriyorsa, sonradan gelen istekler yine kuyrukta birikebilir; ancak o noktaya ulaşmadan önce ele alabileceğimiz uzun süreli istek sayısını artırmış oluruz.

Bu teknik, bir web sunucusunun işleme kapasitesini artırmanın pek çok yolundan yalnızca biridir. İnceleyebileceğiniz diğer seçenekler arasında fork/join modeli, tek thread’li asenkron G/Ç (single-threaded async I/O) modeli ve çok thread’li asenkron G/Ç (multithreaded async I/O) modeli yer alır. Bu konuya ilgi duyuyorsanız diğer çözümler hakkında daha fazla okuma yapabilir ve bunları uygulamayı deneyebilirsiniz; Rust gibi alt seviyeli bir dille bu seçeneklerin tamamı mümkündür.

Bir thread havuzu uygulamaya başlamadan önce, havuzu kullanmanın nasıl görünmesi gerektiği üzerine konuşalım. Kod tasarlamaya çalışırken önce istemci arayüzünü (client interface) yazmak tasarımınıza yön vermeye yardımcı olabilir. Kodun API’sini, onu çağırmak istediğiniz biçimde yapılandırarak yazın; işlevselliği uygulayıp ardından açık (public) API’yi tasarlamak yerine, önce bu yapıyı kurup işlevselliği onun içinde uygulayın.

Bölüm 12’deki projede test güdümlü geliştirmeyi (test-driven development) kullandığımıza benzer biçimde, burada da derleyici yönlendirmeli geliştirme (compiler-driven development) kullanacağız. İstediğimiz fonksiyonları çağıran kodu yazacağız ve ardından kodun çalışmasını sağlamak üzere bir sonraki adımda neyi değiştirmemiz gerektiğini belirlemek için derleyiciden gelen hatalara bakacağız. Ancak bunu yapmadan önce, başlangıç noktası olarak kullanmayacağımız tekniği inceleyelim.

Her İstek İçin Bir Thread Başlatmak

İlk olarak, her bağlantı için gerçekten yeni bir thread oluştursaydı kodumuzun nasıl görüneceğini inceleyelim. Daha önce belirttiğimiz gibi, sınırsız sayıda thread başlatma olasılığının doğuracağı sorunlar nedeniyle nihai planımız bu değil; ancak önce çalışan çok thread’li (multithreaded) bir sunucu elde etmek için iyi bir başlangıç noktasıdır. Ardından iyileştirme olarak thread havuzunu ekleyeceğiz ve iki çözümü karşılaştırmak daha kolay olacak.

Liste 21-11, for döngüsü içinde her akışı ele almak üzere yeni bir thread başlatmak için main fonksiyonunda yapılacak değişiklikleri göstermektedir.

use std::{
    fs,
    io::{BufReader, prelude::*},
    net::{TcpListener, TcpStream},
    thread,
    time::Duration,
};

fn main() {
    let dinleyici = TcpListener::bind("127.0.0.1:7878").unwrap();

    for akis in dinleyici.incoming() {
        let akis = akis.unwrap();

        thread::spawn(|| {
            baglantiyi_isle(akis);
        });
    }
}

fn baglantiyi_isle(mut akis: TcpStream) {
    let tamponlu_okuyucu = BufReader::new(&akis);
    let istek_satiri = tamponlu_okuyucu.lines().next().unwrap().unwrap();

    let (durum_satiri, dosya_adi) = match &istek_satiri[..] {
        "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "merhaba.html"),
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));
            ("HTTP/1.1 200 OK", "merhaba.html")
        }
        _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
    };

    let icerik = fs::read_to_string(dosya_adi).unwrap();
    let uzunluk = icerik.len();

    let yanit =
        format!("{durum_satiri}\r\nContent-Length: {uzunluk}\r\n\r\n{icerik}");

    akis.write_all(yanit.as_bytes()).unwrap();
}

Bölüm 16’da öğrendiğiniz gibi thread::spawn, yeni bir thread oluşturur ve ardından kapanış içindeki kodu bu yeni thread’de çalıştırır. Bu kodu çalıştırıp tarayıcınızda /sleep sayfasını ve hemen ardından iki tarayıcı sekmesinde daha / sayfasını yüklerseniz, / yoluna yapılan isteklerin gerçekten de /sleep isteğinin bitmesini beklemek zorunda kalmadığını görürsünüz. Ne var ki belirttiğimiz gibi, herhangi bir sınır olmaksızın yeni thread’ler oluşturacağınız için bu yaklaşım eninde sonunda sistemi aşırı yükleyecektir.

Bölüm 17’den hatırlayacağınız üzere, bu tam da async ve await’in parladığı türden bir senaryodur! Thread havuzunu inşa ederken bunu aklınızda bulundurun ve async kullanılsaydı nelerin farklı görüneceğini, nelerin aynı kalacağını düşünün.

Sınırlı Sayıda Thread Oluşturmak

Thread’lerden thread havuzuna geçişin API’mizi kullanan kodda büyük değişiklikler gerektirmemesi için thread havuzumuzun benzer ve tanıdık bir biçimde çalışmasını istiyoruz. Liste 21-12, thread::spawn yerine kullanmak istediğimiz IsParcacigiHavuzu struct’ı için tasarladığımız varsayımsal arayüzü göstermektedir.

use std::{
    fs,
    io::{BufReader, prelude::*},
    net::{TcpListener, TcpStream},
    thread,
    time::Duration,
};

fn main() {
    let dinleyici = TcpListener::bind("127.0.0.1:7878").unwrap();
    let havuz = IsParcacigiHavuzu::new(4);

    for akis in dinleyici.incoming() {
        let akis = akis.unwrap();

        havuz.calistir(|| {
            baglantiyi_isle(akis);
        });
    }
}

fn baglantiyi_isle(mut akis: TcpStream) {
    let tamponlu_okuyucu = BufReader::new(&akis);
    let istek_satiri = tamponlu_okuyucu.lines().next().unwrap().unwrap();

    let (durum_satiri, dosya_adi) = match &istek_satiri[..] {
        "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "merhaba.html"),
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));
            ("HTTP/1.1 200 OK", "merhaba.html")
        }
        _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
    };

    let icerik = fs::read_to_string(dosya_adi).unwrap();
    let uzunluk = icerik.len();

    let yanit =
        format!("{durum_satiri}\r\nContent-Length: {uzunluk}\r\n\r\n{icerik}");

    akis.write_all(yanit.as_bytes()).unwrap();
}

Yapılandırılabilir sayıda —bu örnekte dört— thread içeren yeni bir thread havuzu oluşturmak için IsParcacigiHavuzu::new kullanıyoruz. Ardından for döngüsünde havuz.calistir, havuzun her akış için çalıştırması gereken bir kapanış alması bakımından thread::spawn ile benzer bir arayüze sahiptir. havuz.calistir metodunu, kapanışı alıp çalıştırılmak üzere havuzdaki bir thread’e verecek şekilde uygulamamız gerekiyor. Bu kod henüz derlenmeyecektir; ancak derleyicinin onu nasıl düzelteceğimiz konusunda bize rehberlik edebilmesi için derlemeyi deneyeceğiz.

Derleyici Yönlendirmeli Geliştirme ile IsParcacigiHavuzu Oluşturmak

Liste 21-12’deki değişiklikleri src/main.rs dosyasına uygulayın; ardından geliştirme sürecimizi yönlendirmek için cargo check komutunun verdiği derleyici hatalarından yararlanalım. Aldığımız ilk hata şöyledir:

$ cargo check
    Checking merhaba v0.1.0 (file:///projects/merhaba)
error[E0433]: cannot find type `IsParcacigiHavuzu` in this scope
  --> src/main.rs:11:17
   |
11 |     let havuz = IsParcacigiHavuzu::new(4);
   |                 ^^^^^^^^^^^^^^^^^ use of undeclared type `IsParcacigiHavuzu`

For more information about this error, try `rustc --explain E0433`.
error: could not compile `merhaba` (bin "merhaba") due to 1 previous error

Harika! Bu hata bize bir IsParcacigiHavuzu türüne veya modülüne ihtiyacımız olduğunu söylüyor; şimdi bunu oluşturalım. IsParcacigiHavuzu uygulamamız, web sunucumuzun yaptığı işin türünden bağımsız olacaktır. Bu nedenle IsParcacigiHavuzu uygulamamızı barındırmak üzere merhaba paketini yalnızca bir ikili crate (binary crate) olmaktan çıkarıp bir kütüphane crate’i (library crate) de içerecek hâle getirelim. Kütüphane crate’ine geçtikten sonra, bu ayrı thread havuzu kütüphanesini yalnızca web isteklerine yanıt vermek için değil, thread havuzu kullanarak yapmak istediğimiz herhangi bir iş için de kullanabiliriz.

Şimdilik sahip olabileceğimiz en basit IsParcacigiHavuzu struct tanımını içeren aşağıdaki src/lib.rs dosyasını oluşturun:

pub struct IsParcacigiHavuzu;

Ardından src/main.rs dosyasının en üstüne aşağıdaki kodu ekleyerek IsParcacigiHavuzu türünü kütüphane crate’inden kapsama dahil edecek şekilde main.rs dosyasını düzenleyin:

use merhaba::IsParcacigiHavuzu;
use std::{
    fs,
    io::{BufReader, prelude::*},
    net::{TcpListener, TcpStream},
    thread,
    time::Duration,
};

fn main() {
    let dinleyici = TcpListener::bind("127.0.0.1:7878").unwrap();
    let havuz = IsParcacigiHavuzu::new(4);

    for akis in dinleyici.incoming() {
        let akis = akis.unwrap();

        havuz.calistir(|| {
            baglantiyi_isle(akis);
        });
    }
}

fn baglantiyi_isle(mut akis: TcpStream) {
    let tamponlu_okuyucu = BufReader::new(&akis);
    let istek_satiri = tamponlu_okuyucu.lines().next().unwrap().unwrap();

    let (durum_satiri, dosya_adi) = match &istek_satiri[..] {
        "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "merhaba.html"),
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));
            ("HTTP/1.1 200 OK", "merhaba.html")
        }
        _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
    };

    let icerik = fs::read_to_string(dosya_adi).unwrap();
    let uzunluk = icerik.len();

    let yanit =
        format!("{durum_satiri}\r\nContent-Length: {uzunluk}\r\n\r\n{icerik}");

    akis.write_all(yanit.as_bytes()).unwrap();
}

Bu kod hâlâ çalışmayacaktır; ancak ele almamız gereken bir sonraki hatayı görmek için kodu yeniden kontrol edelim:

$ cargo check
    Checking merhaba v0.1.0 (file:///projects/merhaba)
error[E0599]: no associated function or constant named `new` found for struct `IsParcacigiHavuzu` in the current scope
  --> src/main.rs:12:36
   |
12 |     let havuz = IsParcacigiHavuzu::new(4);
   |                                    ^^^ associated function or constant not found in `IsParcacigiHavuzu`

For more information about this error, try `rustc --explain E0599`.
error: could not compile `merhaba` (bin "merhaba") due to 1 previous error

Bu hata, bir sonraki adımda IsParcacigiHavuzu için new adında bir ilişkili fonksiyon oluşturmamız gerektiğini belirtiyor. Ayrıca new fonksiyonunun argüman olarak 4 değerini kabul edebilen bir parametreye sahip olması ve bir IsParcacigiHavuzu örneği döndürmesi gerektiğini de biliyoruz. Bu özelliklere sahip en basit new fonksiyonunu uygulayalım:

pub struct IsParcacigiHavuzu;

impl IsParcacigiHavuzu {
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        IsParcacigiHavuzu
    }
}

Negatif sayıda thread olmasının hiçbir anlam taşımadığını bildiğimiz için boyut parametresinin türü olarak usize seçtik. Ayrıca Bölüm 3’teki “Tam Sayı Türleri” alt bölümünde ele aldığımız üzere, bu 4 değerini bir thread koleksiyonundaki öğe sayısı olarak kullanacağımızı ve usize türünün tam da bu amaç için var olduğunu biliyoruz.

Kodu bir kez daha kontrol edelim:

$ cargo check
    Checking merhaba v0.1.0 (file:///projects/merhaba)
error[E0599]: no method named `calistir` found for struct `IsParcacigiHavuzu` in the current scope
  --> src/main.rs:17:15
   |
17 |         havuz.calistir(|| {
   |         ------^^^^^^^^ method not found in `IsParcacigiHavuzu`

For more information about this error, try `rustc --explain E0599`.
error: could not compile `merhaba` (bin "merhaba") due to 1 previous error

Şimdi hata, IsParcacigiHavuzu üzerinde bir calistir metodumuz olmadığı için ortaya çıkıyor. “Sınırlı Sayıda Thread Oluşturmak” bölümünden hatırlayacağınız üzere, thread havuzumuzun thread::spawn fonksiyonuna benzer bir arayüze sahip olmasına karar vermiştik. Ek olarak calistir metodunu, kendisine verilen kapanışı alıp çalıştırması için havuzdaki boşta bekleyen bir thread’e verecek biçimde uygulayacağız.

IsParcacigiHavuzu üzerindeki calistir metodunu parametre olarak bir kapanış alacak şekilde tanımlayacağız. Bölüm 13’teki “Yakalanan Değerleri Kapanışların Dışına Taşımak” alt bölümünden hatırlayacağınız üzere, kapanışları üç farklı trait ile parametre olarak alabiliriz: Fn, FnMut ve FnOnce. Burada ne tür bir kapanış kullanacağımıza karar vermemiz gerekiyor. Sonunda standart kütüphanedeki thread::spawn uygulamasına benzer bir şey yapacağımızı bildiğimiz için thread::spawn imzasının parametresi üzerinde hangi sınırların (bounds) bulunduğuna bakabiliriz. Belgelendirme bize şunu gösterir:

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
    where
        F: FnOnce() -> T,
        F: Send + 'static,
        T: Send + 'static,

Burada bizi ilgilendiren F tür parametresidir; T tür parametresi dönüş değeriyle ilgilidir ve bizim onunla bir işimiz yoktur. spawn fonksiyonunun F üzerindeki trait sınırı olarak FnOnce kullandığını görebiliriz. calistir içinde aldığımız argümanı nihayetinde spawn fonksiyonuna aktaracağımız için muhtemelen bizim de istediğimiz budur. Bir isteği çalıştıracak olan thread, o isteğin kapanışını yalnızca bir kez yürüteceği için —ki bu da FnOnce adındaki Once (bir kez) ifadesiyle tam örtüşür— kullanmak istediğimiz trait’in FnOnce olduğundan daha da emin olabiliriz.

F tür parametresi ayrıca bizim durumumuzda da oldukça yararlı olan Send trait sınırına ve 'static ömür (lifetime) sınırına sahiptir: Kapanışı bir thread’den diğerine aktarmak için Send sınırına, thread’in çalışmasının ne kadar süreceğini bilmediğimiz için de 'static sınırına ihtiyacımız vardır. IsParcacigiHavuzu üzerinde bu sınırlara sahip F türünde jenerik bir parametre alan calistir metodunu oluşturalım:

pub struct IsParcacigiHavuzu;

impl IsParcacigiHavuzu {
    // --snip--
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        IsParcacigiHavuzu
    }

    pub fn calistir<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

FnOnce sonrasında yine de () kullanıyoruz; çünkü buradaki FnOnce, hiç parametre almayan ve birim türü (unit type) () döndüren bir kapanışı temsil eder. Tıpkı fonksiyon tanımlarında olduğu gibi dönüş türü imzadan çıkarılabilir; ancak hiç parametremiz olmasa bile parantezlere yine de ihtiyacımız vardır.

Yine bu, calistir metodunun olabilecek en basit uygulamasıdır: Hiçbir iş yapmaz; ancak şimdilik yalnızca kodumuzun derlenmesini sağlamaya çalışıyoruz. Kodu yeniden kontrol edelim:

$ cargo check
    Checking merhaba v0.1.0 (file:///projects/merhaba)
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.24s

Derleniyor! Ancak cargo run komutunu deneyip tarayıcıda bir istek yaparsanız, tarayıcıda bölümün başında gördüğümüz hataları göreceğinize dikkat edin. Kütüphanemiz henüz calistir metoduna aktarılan kapanışı gerçekten çağırmıyor!

Düşünün: Burada bir kapanış yerine bir future çalıştıracak olsaydık neler farklı olurdu?

new İçinde Thread Sayısını Doğrulamak

new ve calistir parametreleriyle henüz hiçbir şey yapmıyoruz. Şimdi bu fonksiyonların gövdelerini istediğimiz davranışla uygulayalım. Başlangıç olarak new fonksiyonunu düşünelim. Daha önce negatif sayıda thread içeren bir havuzun anlamı olmadığı için boyut parametresinde işaretsiz (unsigned) bir tür seçmiştik. Ne var ki sıfır thread içeren bir havuz da anlamsızdır; buna rağmen sıfır, son derece geçerli bir usize değeridir. Liste 21-13’te gösterildiği gibi, bir IsParcacigiHavuzu örneği döndürmeden önce boyut değerinin sıfırdan büyük olduğunu denetleyen kodu ekleyecek ve sıfır değeri gelirse assert! makrosunu kullanarak programın paniklemesini sağlayacağız.

pub struct IsParcacigiHavuzu;

impl IsParcacigiHavuzu {
    /// Yeni bir IsParcacigiHavuzu oluşturur.
    ///
    /// Boyut, havuzdaki thread sayısıdır.
    ///
    /// # Panics
    ///
    /// `new` fonksiyonu, boyut sıfırsa panikler.
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        assert!(boyut > 0);

        IsParcacigiHavuzu
    }

    // --snip--
    pub fn calistir<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

Ayrıca belgelendirme yorumlarıyla (doc comments) IsParcacigiHavuzu yapımız için biraz belgelendirme de ekledik. Bölüm 14’te ele aldığımız üzere, fonksiyonumuzun hangi durumlarda panikleyebileceğini açıkça belirten bir bölüm ekleyerek iyi belgelendirme pratiklerini izlediğimize dikkat edin. new için üretilen belgelerin nasıl göründüğünü incelemek üzere cargo doc --open komutunu çalıştırıp IsParcacigiHavuzu struct’ına tıklamayı deneyin!

Burada yaptığımız gibi assert! makrosunu eklemek yerine, G/Ç (I/O) projesindeki Liste 12-9’da Yapilandirma::olustur (Config::build) ile yaptığımız gibi new fonksiyonunu build olarak değiştirip bir Result da döndürebilirdik. Ancak bu durumda, hiç thread içermeyen bir thread havuzu oluşturmaya çalışmanın kurtarılamaz bir hata olmasına karar verdik. Kendinizi iddialı hissediyorsanız, new fonksiyonuyla karşılaştırmak üzere aşağıdaki imzaya sahip build adında bir fonksiyon yazmayı deneyin:

pub fn build(boyut: usize) -> Result<IsParcacigiHavuzu, HavuzOlusturmaHatasi> {

Thread’leri Saklayacak Alan Oluşturmak

Artık havuzda saklamak için geçerli sayıda thread’e sahip olduğumuzdan emin olmanın bir yolu bulunduğuna göre, bu thread’leri oluşturabilir ve struct’ı döndürmeden önce IsParcacigiHavuzu struct’ı içinde saklayabiliriz. Peki bir thread’i nasıl “saklarız”? thread::spawn imzasına bir kez daha göz atalım:

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
    where
        F: FnOnce() -> T,
        F: Send + 'static,
        T: Send + 'static,

spawn fonksiyonu bir JoinHandle<T> döndürür; burada T, kapanışın döndürdüğü türdür. Biz de JoinHandle kullanmayı deneyelim ve ne olacağını görelim. Bizim durumumuzda thread havuzuna aktardığımız kapanışlar bağlantıyı işleyecek ve hiçbir değer döndürmeyecektir; dolayısıyla T, birim türü () olacaktır.

Liste 21-14’teki kod derlenecektir ancak henüz hiçbir thread oluşturmaz. IsParcacigiHavuzu tanımını thread::JoinHandle<()> örneklerinden oluşan bir vektör (threadler) tutacak şekilde değiştirdik, vektörü boyut kapasitesiyle ilklendirdik, thread’leri oluşturacak kodu çalıştıracak bir for döngüsü kurduk ve bunları barındıran bir IsParcacigiHavuzu örneği döndürdük.

use std::thread;

pub struct IsParcacigiHavuzu {
    threadler: Vec<thread::JoinHandle<()>>,
}

impl IsParcacigiHavuzu {
    // --snip--
    /// Yeni bir IsParcacigiHavuzu oluşturur.
    ///
    /// Boyut, havuzdaki thread sayısıdır.
    ///
    /// # Panics
    ///
    /// `new` fonksiyonu, boyut sıfırsa panikler.
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        assert!(boyut > 0);

        let mut threadler = Vec::with_capacity(boyut);

        for _ in 0..boyut {
            // thread'leri oluştur ve vektörde sakla
        }

        IsParcacigiHavuzu { threadler }
    }
    // --snip--

    pub fn calistir<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

IsParcacigiHavuzu içindeki vektörün öğe türü olarak thread::JoinHandle kullandığımız için kütüphane crate’inde std::thread modülünü kapsama dahil ettik.

Geçerli bir boyut alındığında IsParcacigiHavuzu yapımız boyut adet öğe tutabilen yeni bir vektör oluşturur. with_capacity fonksiyonu Vec::new ile aynı görevi yerine getirir, ancak önemli bir farkı vardır: Vektörde alanı önceden ayırır (pre-allocates). Vektörde boyut adet öğe saklamamız gerektiğini bildiğimizden, bu bellek ayırma işlemini baştan yapmak, öğeler eklendikçe kendisini yeniden boyutlandıran Vec::new kullanımına kıyasla biraz daha verimlidir.

cargo check komutunu yeniden çalıştırdığınızda derleme başarılı olmalıdır.

IsParcacigiHavuzu’ndan Bir Thread’e Kod Göndermek

Liste 21-14’teki for döngüsünde thread’lerin oluşturulmasına ilişkin bir yorum bırakmıştık. Şimdi thread’leri gerçekte nasıl oluşturacağımıza bakalım. Standart kütüphane, thread oluşturmanın bir yolu olarak thread::spawn fonksiyonunu sunar ve thread::spawn, thread oluşturulur oluşturulmaz çalıştırması gereken bir kod almayı bekler. Ancak bizim durumumuzda thread’leri oluşturup daha sonra göndereceğimiz kodu beklemelerini sağlamak istiyoruz. Standart kütüphanenin thread uygulaması bunu yapmanın herhangi bir yolunu içermez; bunu bizim elle uygulamamız gerekir.

Bu davranışı, IsParcacigiHavuzu ile thread’ler arasına bu yeni davranışı yönetecek yeni bir veri yapısı ekleyerek uygulayacağız. Havuzlama (pooling) uygulamalarında yaygın bir terim olan Worker (çalışan) kavramına karşılık bu veri yapısına Calisan adını vereceğiz. Calisan, çalıştırılması gereken kodu alır ve bu kodu kendi thread’inde çalıştırır.

Bir restoranın mutfağında çalışan insanları düşünün: Çalışanlar müşterilerden sipariş gelene kadar bekler, ardından bu siparişleri alıp hazırlamaktan sorumludurlar.

Thread havuzunda JoinHandle<()> örneklerinden oluşan bir vektör saklamak yerine, Calisan struct’ının örneklerini saklayacağız. Her Calisan tek bir JoinHandle<()> örneği tutacaktır. Ardından Calisan üzerinde, çalıştırılacak kodun kapanışını alıp bunu hâlihazırda çalışmakta olan thread’e yürütülmek üzere gönderecek bir metot uygulayacağız. Ayrıca günlük kaydı tutarken (logging) veya hata ayıklarken havuzdaki farklı Calisan örneklerini birbirinden ayırt edebilmek için her Calisan örneğine bir kimlik vereceğiz.

Bir IsParcacigiHavuzu oluşturduğumuzda gerçekleşecek yeni süreç şöyledir. Calisan yapısını bu şekilde kurduktan sonra kapanışı thread’e gönderen kodu uygulayacağız:

1. Bir kimlik ve bir JoinHandle<()> tutan Calisan struct’ını tanımlamak.
2. IsParcacigiHavuzu yapısını Calisan örneklerinden oluşan bir vektör tutacak şekilde değiştirmek.
3. Bir kimlik numarası alan ve bu kimlik ile boş bir kapanışla başlatılmış bir thread tutan Calisan örneğini döndüren Calisan::new fonksiyonunu tanımlamak.
4. IsParcacigiHavuzu::new içinde, bir kimlik üretmek için for döngüsü sayacını kullanmak, bu kimlik ile yeni bir Calisan oluşturmak ve bu Calisan örneğini vektörde saklamak.

Kendinizi sınamak isterseniz, Liste 21-15’teki koda bakmadan önce bu değişiklikleri kendi başınıza uygulamayı deneyin.

Hazır mısınız? İşte önceki değişiklikleri yapmanın bir yolunu gösteren Liste 21-15.

use std::thread;

pub struct IsParcacigiHavuzu {
    calisanlar: Vec<Calisan>,
}

impl IsParcacigiHavuzu {
    // --snip--
    /// Yeni bir IsParcacigiHavuzu oluşturur.
    ///
    /// Boyut, havuzdaki thread sayısıdır.
    ///
    /// # Panics
    ///
    /// `new` fonksiyonu, boyut sıfırsa panikler.
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        assert!(boyut > 0);

        let mut calisanlar = Vec::with_capacity(boyut);

        for kimlik in 0..boyut {
            calisanlar.push(Calisan::new(kimlik));
        }

        IsParcacigiHavuzu { calisanlar }
    }
    // --snip--

    pub fn calistir<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

struct Calisan {
    kimlik: usize,
    thread: thread::JoinHandle<()>,
}

impl Calisan {
    fn new(kimlik: usize) -> Calisan {
        let thread = thread::spawn(|| {});

        Calisan { kimlik, thread }
    }
}

IsParcacigiHavuzu üzerindeki alanın adını threadler yerine calisanlar olarak değiştirdik; çünkü bu alan artık JoinHandle<()> örnekleri yerine Calisan örneklerini tutuyor. for döngüsündeki sayacı Calisan::new fonksiyonuna argüman olarak veriyor ve oluşturulan her yeni Calisan örneğini calisanlar adlı vektörde saklıyoruz.

Dış kodların (src/main.rs içindeki sunucumuz gibi) IsParcacigiHavuzu içinde bir Calisan struct’ı kullanıldığına dair uygulama ayrıntılarını bilmesine gerek yoktur; bu nedenle Calisan struct’ını ve onun new fonksiyonunu gizli (private) yapıyoruz. Calisan::new fonksiyonu kendisine verdiğimiz kimlik değerini kullanır ve boş bir kapanışla yeni bir thread başlatılarak oluşturulan JoinHandle<()> örneğini thread alanında saklar.

Bu kod derlenecek ve IsParcacigiHavuzu::new fonksiyonuna argüman olarak belirttiğimiz sayıda Calisan örneğini saklayacaktır. Ancak calistir içinde aldığımız kapanışı hâlâ işlemiyor. Şimdi bunu nasıl yapacağımıza bakalım.

Kanallar Aracılığıyla Thread’lere İstek Göndermek

Ele alacağımız bir sonraki sorun, thread::spawn fonksiyonuna verilen kapanışların kesinlikle hiçbir iş yapmamasıdır. Şu anda çalıştırmak istediğimiz kapanışı calistir metodunda alıyoruz. Oysa IsParcacigiHavuzu oluşturulurken her bir Calisan örneğini oluşturduğumuz sırada thread::spawn fonksiyonuna çalıştıracağı bir kapanış vermemiz gerekiyor.

Az önce oluşturduğumuz Calisan struct’larının, çalıştırılacak kodu IsParcacigiHavuzu içinde tutulan bir kuyruktan almasını ve çalıştırılmak üzere kendi thread’ine göndermesini istiyoruz.

Bölüm 16’da öğrendiğimiz mesajlaşma kanalları (mpsc kanalları) —iki thread arasında iletişim kurmanın basit bir yolu— bu kullanım senaryosu için biçilmiş kaftandır. Görev kuyruğu işlevi görmesi için bir kanal kullanacağız ve calistir metodu IsParcacigiHavuzu’ndan Calisan örneklerine bir görev gönderecek; Calisan örnekleri de bu görevi kendi thread’ine iletecektir. Planımız şöyledir:

1. IsParcacigiHavuzu bir kanal oluşturacak ve gönderici ucunu (gonderici) elinde tutacak.
2. Her Calisan alıcı ucunu (alici) elinde tutacak.
3. Kanal üzerinden göndermek istediğimiz kapanışları tutacak yeni bir Gorev struct’ı oluşturacağız.
4. calistir metodu, çalıştırmak istediği görevi gönderici aracılığıyla gönderecek.
5. Calisan, kendi thread’i içinde alıcısı üzerinde döngü kuracak ve aldığı tüm görevlerin kapanışlarını çalıştıracak.

Liste 21-16’da gösterildiği gibi IsParcacigiHavuzu::new içinde bir kanal oluşturarak ve göndericiyi IsParcacigiHavuzu örneğinde tutarak işe başlayalım. Gorev struct’ı şimdilik hiçbir şey tutmuyor, ancak kanal üzerinden göndereceğimiz öğenin türü olacaktır.

use std::{sync::mpsc, thread};

pub struct IsParcacigiHavuzu {
    calisanlar: Vec<Calisan>,
    gonderici: mpsc::Sender<Gorev>,
}

struct Gorev;

impl IsParcacigiHavuzu {
    // --snip--
    /// Yeni bir IsParcacigiHavuzu oluşturur.
    ///
    /// Boyut, havuzdaki thread sayısıdır.
    ///
    /// # Panics
    ///
    /// `new` fonksiyonu, boyut sıfırsa panikler.
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        assert!(boyut > 0);

        let (gonderici, alici) = mpsc::channel();

        let mut calisanlar = Vec::with_capacity(boyut);

        for kimlik in 0..boyut {
            calisanlar.push(Calisan::new(kimlik));
        }

        IsParcacigiHavuzu { calisanlar, gonderici }
    }
    // --snip--

    pub fn calistir<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

struct Calisan {
    kimlik: usize,
    thread: thread::JoinHandle<()>,
}

impl Calisan {
    fn new(kimlik: usize) -> Calisan {
        let thread = thread::spawn(|| {});

        Calisan { kimlik, thread }
    }
}

IsParcacigiHavuzu::new içinde yeni kanalımızı oluşturuyor ve havuzun gonderici ucunu tutmasını sağlıyoruz. Bu kod başarıyla derlenecektir.

Thread havuzu kanalı oluştururken kanalın alıcı ucunu her bir Calisan örneğine aktarmayı deneyelim. Alıcıyı Calisan örneklerinin başlattığı thread içinde kullanmak istediğimizi bildiğimizden, kapanış içinde alici parametresine başvuracağız. Liste 21-17’deki kod henüz tam olarak derlenmeyecektir.

use std::{sync::mpsc, thread};

pub struct IsParcacigiHavuzu {
    calisanlar: Vec<Calisan>,
    gonderici: mpsc::Sender<Gorev>,
}

struct Gorev;

impl IsParcacigiHavuzu {
    // --snip--
    /// Yeni bir IsParcacigiHavuzu oluşturur.
    ///
    /// Boyut, havuzdaki thread sayısıdır.
    ///
    /// # Panics
    ///
    /// `new` fonksiyonu, boyut sıfırsa panikler.
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        assert!(boyut > 0);

        let (gonderici, alici) = mpsc::channel();

        let mut calisanlar = Vec::with_capacity(boyut);

        for kimlik in 0..boyut {
            calisanlar.push(Calisan::new(kimlik, alici));
        }

        IsParcacigiHavuzu { calisanlar, gonderici }
    }
    // --snip--

    pub fn calistir<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

// --snip--


struct Calisan {
    kimlik: usize,
    thread: thread::JoinHandle<()>,
}

impl Calisan {
    fn new(kimlik: usize, alici: mpsc::Receiver<Gorev>) -> Calisan {
        let thread = thread::spawn(|| {
            alici;
        });

        Calisan { kimlik, thread }
    }
}

Küçük ve doğrudan birkaç değişiklik yaptık: Alıcıyı Calisan::new fonksiyonuna aktarıyor ve ardından kapanışın içinde kullanıyoruz.

Bu kodu kontrol etmeyi denediğimizde şu hatayı alırız:

$ cargo check
    Checking merhaba v0.1.0 (file:///projects/merhaba)
error[E0382]: use of moved value: `alici`
  --> src/lib.rs:26:50
   |
21 |         let (gonderici, alici) = mpsc::channel();
   |                         ----- move occurs because `alici` has type `std::sync::mpsc::Receiver<Gorev>`, which does not implement the `Copy` trait
...
25 |         for kimlik in 0..boyut {
   |         ---------------------- inside of this loop
26 |             calisanlar.push(Calisan::new(kimlik, alici));
   |                                                  ^^^^^ value moved here, in previous iteration of loop
   |
note: consider changing this parameter type in method `new` to borrow instead if owning the value isn't necessary
  --> src/lib.rs:47:34
   |
47 |     fn new(kimlik: usize, alici: mpsc::Receiver<Gorev>) -> Calisan {
   |        --- in this method        ^^^^^^^^^^^^^^^^^^^^^ this parameter takes ownership of the value

For more information about this error, try `rustc --explain E0382`.
error: could not compile `merhaba` (lib) due to 1 previous error

Derleyici use of moved value: alici (taşınmış değerin kullanımı: alici) hatası veriyor; çünkü kodumuz alici değerini döngü içinde birden fazla Calisan örneğine aktarmaya çalışıyor. Bölüm 16’dan hatırlayacağınız üzere bu işe yaramaz: Rust’ın sunduğu kanal uygulaması çok üreticili, tek tüketicilidir (multiple producer, single consumer — mpsc). Bu da kodu düzeltmek için kanalın tüketici ucunu öylece klonlayamayacağımız anlamına gelir. Zaten bir mesajı birden fazla tüketiciye birden çok kez göndermek de istemiyoruz; her mesajın yalnızca bir kez işleneceği biçimde, birden fazla Calisan örneğinin paylaştığı tek bir mesaj listesi istiyoruz.

Ek olarak, kanal kuyruğundan bir görev almak alici değerini değiştirmeyi (mutating) içerir; dolayısıyla thread’lerin alici değerini paylaşmak ve değiştirmek için güvenli bir yola ihtiyacı vardır; aksi takdirde yarış durumlarıyla (race conditions, Bölüm 16’da ele aldığımız gibi) karşılaşabiliriz.

Bölüm 16’da ele aldığımız thread açısından güvenli (thread-safe) akıllı işaretçileri hatırlayın: Sahipliği birden fazla thread arasında paylaşmak ve thread’lerin değeri değiştirmesine izin vermek için Arc<Mutex<T>> kullanmamız gerekir. Arc türü birden fazla Calisan örneğinin alıcıya sahip olmasına olanak tanırken, Mutex aynı anda yalnızca tek bir Calisan örneğinin alıcıdan görev almasını güvence altına alır. Liste 21-18 yapmamız gereken değişiklikleri göstermektedir.

use std::{
    sync::{Arc, Mutex, mpsc},
    thread,
};
// --snip--

pub struct IsParcacigiHavuzu {
    calisanlar: Vec<Calisan>,
    gonderici: mpsc::Sender<Gorev>,
}

struct Gorev;

impl IsParcacigiHavuzu {
    // --snip--
    /// Yeni bir IsParcacigiHavuzu oluşturur.
    ///
    /// Boyut, havuzdaki thread sayısıdır.
    ///
    /// # Panics
    ///
    /// `new` fonksiyonu, boyut sıfırsa panikler.
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        assert!(boyut > 0);

        let (gonderici, alici) = mpsc::channel();

        let alici = Arc::new(Mutex::new(alici));

        let mut calisanlar = Vec::with_capacity(boyut);

        for kimlik in 0..boyut {
            calisanlar.push(Calisan::new(kimlik, Arc::clone(&alici)));
        }

        IsParcacigiHavuzu { calisanlar, gonderici }
    }

    // --snip--

    pub fn calistir<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

// --snip--

struct Calisan {
    kimlik: usize,
    thread: thread::JoinHandle<()>,
}

impl Calisan {
    fn new(kimlik: usize, alici: Arc<Mutex<mpsc::Receiver<Gorev>>>) -> Calisan {
        // --snip--
        let thread = thread::spawn(|| {
            alici;
        });

        Calisan { kimlik, thread }
    }
}

IsParcacigiHavuzu::new içinde alıcıyı bir Arc ve bir Mutex içine koyuyoruz. Her yeni Calisan için, Calisan örneklerinin alıcının sahipliğini paylaşabilmesi amacıyla referans sayısını artırmak üzere Arc değerini klonluyoruz (Arc::clone(&alici)).

Bu değişikliklerle kod derleniyor! Hedefe yaklaşıyoruz!

calistir Metodunu Uygulamak

Sonunda IsParcacigiHavuzu üzerindeki calistir metodunu uygulayalım. Ayrıca Gorev tanımını bir struct olmaktan çıkarıp calistir metodunun aldığı kapanış türünü tutan bir trait nesnesi için tür takma adına (type alias) dönüştüreceğiz. Bölüm 20’deki “Tür Eşanlamlıları ve Tür Takma Adları” alt bölümünde ele aldığımız üzere, tür takma adları kullanım kolaylığı sağlamak amacıyla uzun türleri daha kısa hâle getirmemize olanak tanır. Liste 21-19’a bakın.

use std::{
    sync::{Arc, Mutex, mpsc},
    thread,
};

pub struct IsParcacigiHavuzu {
    calisanlar: Vec<Calisan>,
    gonderici: mpsc::Sender<Gorev>,
}

// --snip--

type Gorev = Box<dyn FnOnce() + Send + 'static>;

impl IsParcacigiHavuzu {
    // --snip--
    /// Yeni bir IsParcacigiHavuzu oluşturur.
    ///
    /// Boyut, havuzdaki thread sayısıdır.
    ///
    /// # Panics
    ///
    /// `new` fonksiyonu, boyut sıfırsa panikler.
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        assert!(boyut > 0);

        let (gonderici, alici) = mpsc::channel();

        let alici = Arc::new(Mutex::new(alici));

        let mut calisanlar = Vec::with_capacity(boyut);

        for kimlik in 0..boyut {
            calisanlar.push(Calisan::new(kimlik, Arc::clone(&alici)));
        }

        IsParcacigiHavuzu { calisanlar, gonderici }
    }

    pub fn calistir<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let gorev = Box::new(f);

        self.gonderici.send(gorev).unwrap();
    }
}

// --snip--

struct Calisan {
    kimlik: usize,
    thread: thread::JoinHandle<()>,
}

impl Calisan {
    fn new(kimlik: usize, alici: Arc<Mutex<mpsc::Receiver<Gorev>>>) -> Calisan {
        let thread = thread::spawn(|| {
            alici;
        });

        Calisan { kimlik, thread }
    }
}

calistir içinde aldığımız kapanışı kullanarak yeni bir Gorev örneği oluşturduktan sonra, bu görevi kanalın gönderici ucundan gönderiyoruz. Gönderme işleminin başarısız olma ihtimaline karşı send üzerinde unwrap çağırıyoruz. Bu durum, örneğin tüm thread’lerimizin çalışmasını durdurursak —yani alıcı uç yeni mesajlar almayı bırakırsa— meydana gelebilir. Şu an için thread’lerimizin çalışmasını durduramıyoruz: Havuz var olduğu sürece thread’lerimiz çalışmaya devam ediyor. unwrap kullanmamızın nedeni, hata durumunun gerçekleşmeyeceğini bizim bilmemiz, ancak derleyicinin bunu bilmemesidir.

Ancak işimiz henüz tam olarak bitmedi! Calisan içinde thread::spawn fonksiyonuna aktarılan kapanışımız hâlâ kanalın alıcı ucuna yalnızca başvuruda bulunuyor. Bunun yerine kapanışın sonsuza dek döngüde kalmasını, kanalın alıcı ucundan bir görev istemesini ve bir görev aldığında onu çalıştırmasını sağlamamız gerekiyor. Calisan::new üzerinde Liste 21-20’de gösterilen değişikliği yapalım.

use std::{
    sync::{Arc, Mutex, mpsc},
    thread,
};

pub struct IsParcacigiHavuzu {
    calisanlar: Vec<Calisan>,
    gonderici: mpsc::Sender<Gorev>,
}

type Gorev = Box<dyn FnOnce() + Send + 'static>;

impl IsParcacigiHavuzu {
    /// Yeni bir IsParcacigiHavuzu oluşturur.
    ///
    /// Boyut, havuzdaki thread sayısıdır.
    ///
    /// # Panics
    ///
    /// `new` fonksiyonu, boyut sıfırsa panikler.
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        assert!(boyut > 0);

        let (gonderici, alici) = mpsc::channel();

        let alici = Arc::new(Mutex::new(alici));

        let mut calisanlar = Vec::with_capacity(boyut);

        for kimlik in 0..boyut {
            calisanlar.push(Calisan::new(kimlik, Arc::clone(&alici)));
        }

        IsParcacigiHavuzu {
            calisanlar,
            gonderici,
        }
    }

    pub fn calistir<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let gorev = Box::new(f);

        self.gonderici.send(gorev).unwrap();
    }
}

struct Calisan {
    kimlik: usize,
    thread: thread::JoinHandle<()>,
}

// --snip--

impl Calisan {
    fn new(kimlik: usize, alici: Arc<Mutex<mpsc::Receiver<Gorev>>>) -> Calisan {
        let thread = thread::spawn(move || loop {
            let gorev = alici.lock().unwrap().recv().unwrap();

            println!("Çalışan {kimlik} bir görev aldı; çalıştırılıyor.");

            gorev();
        });

        Calisan { kimlik, thread }
    }
}

Burada önce mutex kilidini edinmek için alici üzerinde lock çağırıyor, ardından olası herhangi bir hatada paniklemek üzere unwrap çağırıyoruz. Mutex zehirlenmiş (poisoned) bir durumdaysa kilit edinme işlemi başarısız olabilir; bu durum, başka bir thread kilidi serbest bırakmak yerine kilidi elinde tutarken paniklediyse gerçekleşebilir. Böyle bir durumda bu thread’in de paniklemesi için unwrap çağırmak yapılacak doğru eylemdir. Bu unwrap çağrısını sizin için anlamlı bir hata mesajı içeren bir expect çağrısıyla değiştirmekte özgürsünüz.

Kilidi edinebilirsek kanaldan bir Gorev almak için recv çağırırız. Sondaki unwrap da buradaki olası hatalarda aynı şekilde programı durdurur; alıcı kapandığında send metodunun Err döndürmesine benzer biçimde, kanalın gönderici ucu düşürülmüşse (yani göndericiyi tutan taraf kapanmışsa) burada da hata oluşabilir.

recv çağrısı engelleyicidir (blocking); yani henüz bir görev yoksa mevcut thread bir görev hazır olana kadar bekleyecektir. Mutex<T>, aynı anda yalnızca bir Calisan thread’inin görev istemeye çalışmasını güvence altına alır.

Thread havuzumuz artık çalışır durumda! cargo run komutunu çalıştırın ve birkaç istek gönderin:

$ cargo run
   Compiling merhaba v0.1.0 (file:///projects/merhaba)
warning: field `calisanlar` is never read
 --> src/lib.rs:7:5
  |
6 | pub struct IsParcacigiHavuzu {
  |            ----------------- field in this struct
7 |     calisanlar: Vec<Calisan>,
  |     ^^^^^^^^^^
  |
  = note: `#[warn(dead_code)]` on by default

warning: fields `kimlik` and `thread` are never read
  --> src/lib.rs:51:5
   |
50 | struct Calisan {
   |        ------- fields in this struct
51 |     kimlik: usize,
   |     ^^^^^^
52 |     thread: thread::JoinHandle<()>,
   |     ^^^^^^

warning: `merhaba` (lib) generated 2 warnings
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 4.91s
     Running `target/debug/merhaba`
Çalışan 0 bir görev aldı; çalıştırılıyor.
Çalışan 2 bir görev aldı; çalıştırılıyor.
Çalışan 1 bir görev aldı; çalıştırılıyor.
Çalışan 3 bir görev aldı; çalıştırılıyor.
Çalışan 0 bir görev aldı; çalıştırılıyor.
Çalışan 2 bir görev aldı; çalıştırılıyor.
Çalışan 1 bir görev aldı; çalıştırılıyor.
Çalışan 3 bir görev aldı; çalıştırılıyor.
Çalışan 0 bir görev aldı; çalıştırılıyor.
Çalışan 2 bir görev aldı; çalıştırılıyor.

Başardık! Artık bağlantıları birbirini bekletmeden (eşzamansız olarak) yürüten bir thread havuzumuz var. Hiçbir zaman dörtten fazla thread oluşturulmadığından, sunucu çok sayıda istek alsa bile sistemimiz aşırı yüklenmeyecektir. /sleep yoluna bir istek yaparsak sunucu, diğer istekleri başka thread’lerde çalıştırarak yanıtlamaya devam edebilecektir.

Şimdi bir an durup şunu düşünmenin tam zamanı: Yapılacak iş için kapanış yerine future kullansaydık Liste 21-18, 21-19 ve 21-20’deki kod nasıl farklılaşırdı? Hangi türler değişirdi? Metot imzaları —eğer değişecekse— nasıl farklı olurdu? Kodun hangi kısımları aynı kalırdı?

Bölüm 17 ve Bölüm 19’da while let döngüsünü öğrendikten sonra, Calisan thread kodunu neden Liste 21-21’de gösterildiği gibi yazmadığımızı merak ediyor olabilirsiniz.

use std::{
    sync::{Arc, Mutex, mpsc},
    thread,
};

pub struct IsParcacigiHavuzu {
    calisanlar: Vec<Calisan>,
    gonderici: mpsc::Sender<Gorev>,
}

type Gorev = Box<dyn FnOnce() + Send + 'static>;

impl IsParcacigiHavuzu {
    /// Yeni bir IsParcacigiHavuzu oluşturur.
    ///
    /// Boyut, havuzdaki thread sayısıdır.
    ///
    /// # Panics
    ///
    /// `new` fonksiyonu, boyut sıfırsa panikler.
    pub fn new(boyut: usize) -> IsParcacigiHavuzu {
        assert!(boyut > 0);

        let (gonderici, alici) = mpsc::channel();

        let alici = Arc::new(Mutex::new(alici));

        let mut calisanlar = Vec::with_capacity(boyut);

        for kimlik in 0..boyut {
            calisanlar.push(Calisan::new(kimlik, Arc::clone(&alici)));
        }

        IsParcacigiHavuzu {
            calisanlar,
            gonderici,
        }
    }

    pub fn calistir<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let gorev = Box::new(f);

        self.gonderici.send(gorev).unwrap();
    }
}

struct Calisan {
    kimlik: usize,
    thread: thread::JoinHandle<()>,
}
// --snip--

impl Calisan {
    fn new(kimlik: usize, alici: Arc<Mutex<mpsc::Receiver<Gorev>>>) -> Calisan {
        let thread = thread::spawn(move || {
            while let Ok(gorev) = alici.lock().unwrap().recv() {
                println!("Çalışan {kimlik} bir görev aldı; çalıştırılıyor.");

                gorev();
            }
        });

        Calisan { kimlik, thread }
    }
}

Bu kod derlenir ve çalışır, ancak istediğimiz çoklu thread davranışını üretmez: Yavaş bir istek yine diğer isteklerin işlenmek için beklemesine neden olur. Bunun nedeni biraz inceliklidir: Mutex struct’ının açık (public) bir unlock metodu yoktur; çünkü kilidin sahipliği, lock metodunun döndürdüğü LockResult<MutexGuard<T>> içindeki MutexGuard<T> değerinin ömrüne (lifetime) dayanır. Böylece ödünç alma denetleyicisi (borrow checker), Mutex tarafından korunan bir kaynağa kilidi elimizde tutmadığımız sürece erişilemeyeceği kuralını derleme zamanında zorunlu kılabilir. Ne var ki MutexGuard<T> ömrüne dikkat etmezsek, bu tasarım kilidin amaçlanandan daha uzun süre tutulmasıyla da sonuçlanabilir.

Liste 21-20’de let gorev = alici.lock().unwrap().recv().unwrap(); kullanan kod sorunsuz çalışır; çünkü let ile, eşittir işaretinin sağ tarafındaki ifadede kullanılan tüm geçici değerler (temporary values), let deyimi sona erdiği anda derhal düşürülür (drop). Buna karşılık while let (ve if let ile match), geçici değerleri ilişkili bloğun sonuna kadar düşürmez. Liste 21-21’de kilit, gorev() çağrısı süresince tutulmaya devam eder; bu da diğer Calisan örneklerinin görev alamaması anlamına gelir.