Rust Programlama Dili

Async ile Eşzamanlılık Uygulama

Bu bölümde async yaklaşımını, Bölüm 16’da thread’lerle ele aldığımız eşzamanlılık zorluklarının bazılarına uygulayacağız. Temel fikirlerin pek çoğunu orada zaten tartıştığımız için, bu bölümde thread’ler ile future’lar arasındaki farklara odaklanacağız.

Pek çok durumda async kullanarak eşzamanlılıkla çalışmaya yönelik API’ler, thread kullanmaya yönelik API’lere çok benzer. Bazı durumlarda ise oldukça farklı olurlar. API’ler thread’ler ile async arasında birbirine benzer göründüğünde bile çoğu zaman farklı davranışlar sergiler — ve neredeyse her zaman farklı performans özelliklerine sahiptir.

spawn_task ile Yeni Bir Görev Oluşturma

Bölüm 16’daki “spawn ile Yeni Bir Thread Oluşturma” başlığında ele aldığımız ilk işlem, iki ayrı thread üzerinde sayıları artırarak saymaktı. Gelin aynısını async kullanarak yapalım. trpl crate’i, thread::spawn API’sine çok benzeyen bir spawn_task fonksiyonu ve thread::sleep API’sinin asenkron sürümü olan bir sleep fonksiyonu sunar. Liste 17-6’da gösterildiği gibi, sayma örneğini uygulamak için bunları bir arada kullanabiliriz.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        trpl::spawn_task(async {
            for i in 1..10 {
                println!("birinci görevden merhaba sayı {i}!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        });

        for i in 1..5 {
            println!("ikinci görevden merhaba sayı {i}!");
            trpl::sleep(Duration::from_millis(500)).await;
        }
    });
}

Başlangıç noktası olarak, en üst düzey fonksiyonumuzun asenkron olabilmesi için main fonksiyonumuzu trpl::block_on ile kuruyoruz.

Ardından bu bloğun içinde, her biri bir sonraki mesajı yazdırmadan önce yarım saniye (500 milisaniye) bekleyen bir trpl::sleep çağrısı içeren iki döngü yazıyoruz. Döngülerden birini trpl::spawn_task gövdesine koyuyor, diğerini ise en üst düzeyde bir for döngüsü olarak bırakıyoruz. Ayrıca sleep çağrılarının sonuna birer await ekliyoruz.

Bu kod, kendi terminalinizde çalıştırdığınızda mesajların farklı bir sırayla görünebilmesi de dahil olmak üzere thread tabanlı uygulamaya benzer biçimde davranır:

ikinci görevden merhaba sayı 1!
birinci görevden merhaba sayı 1!
birinci görevden merhaba sayı 2!
ikinci görevden merhaba sayı 2!
birinci görevden merhaba sayı 3!
ikinci görevden merhaba sayı 3!
birinci görevden merhaba sayı 4!
ikinci görevden merhaba sayı 4!
birinci görevden merhaba sayı 5!

Bu sürüm, ana async bloğun gövdesindeki for döngüsü biter bitmez durur; çünkü spawn_task tarafından başlatılan görev, main fonksiyonu sona erdiğinde kapatılır. Görevin sonuna kadar çalışıp tamamlanmasını istiyorsanız, birinci görevin bitmesini beklemek için bir birleştirme tutamacı (join handle) kullanmanız gerekir. Thread’lerle çalışırken, thread çalışmasını bitirene kadar “bloklamak” için join metodunu kullanmıştık. Liste 17-7’de ise aynı şeyi yapmak için await kullanabiliriz; çünkü görev tutamacının (tutamac) kendisi de bir future’dır. Output türü bir Result olduğu için, onu bekledikten sonra üzerinde unwrap da çağırıyoruz.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let tutamac = trpl::spawn_task(async {
            for i in 1..10 {
                println!("birinci görevden merhaba sayı {i}!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        });

        for i in 1..5 {
            println!("ikinci görevden merhaba sayı {i}!");
            trpl::sleep(Duration::from_millis(500)).await;
        }

        tutamac.await.unwrap();
    });
}

Güncellenen bu sürüm her iki döngü de bitene kadar çalışır:

ikinci görevden merhaba sayı 1!
birinci görevden merhaba sayı 1!
birinci görevden merhaba sayı 2!
ikinci görevden merhaba sayı 2!
birinci görevden merhaba sayı 3!
ikinci görevden merhaba sayı 3!
birinci görevden merhaba sayı 4!
ikinci görevden merhaba sayı 4!
birinci görevden merhaba sayı 5!
birinci görevden merhaba sayı 6!
birinci görevden merhaba sayı 7!
birinci görevden merhaba sayı 8!
birinci görevden merhaba sayı 9!

Şimdiye kadar async ve thread’ler bize yalnızca farklı bir sözdizimiyle benzer sonuçlar veriyor gibi görünüyor: birleştirme tutamacı üzerinde join çağırmak yerine await kullanmak ve sleep çağrılarını beklemek.

Asıl büyük fark, bunu yapmak için başka bir işletim sistemi thread’i başlatmamıza gerek kalmamasıdır. Hatta burada yeni bir görev başlatmamıza bile gerek yoktur. Async bloklar anonim future’lara derlendiği için, her döngüyü bir async bloğa koyabilir ve trpl::join fonksiyonunu kullanarak çalışma zamanının her ikisini de tamamlanana kadar çalıştırmasını sağlayabiliriz.

Bölüm 16’daki “Tüm Thread’lerin Bitmesini Bekleme” başlığında, std::thread::spawn çağırdığınızda döndürülen JoinHandle türü üzerinde join metodunun nasıl kullanılacağını göstermiştik. trpl::join fonksiyonu da buna benzer, ancak future’lar içindir. Ona iki future verdiğinizde, her ikisi de tamamlandığında geçirdiğiniz her bir future’ın çıktısını içeren bir demet (tuple) üreten tek bir yeni future oluşturur. Böylece Liste 17-8’de, hem fut1 hem de fut2 değerlerinin bitmesini beklemek için trpl::join kullanıyoruz. fut1 ve fut2 değerlerini ayrı ayrı değil, bunun yerine trpl::join tarafından üretilen yeni future’ı bekliyoruz. Çıktıyı ise yalnızca iki birim değer (()) içeren bir demet olduğu için yoksayıyoruz.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let fut1 = async {
            for i in 1..10 {
                println!("birinci görevden merhaba sayı {i}!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let fut2 = async {
            for i in 1..5 {
                println!("ikinci görevden merhaba sayı {i}!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        trpl::join(fut1, fut2).await;
    });
}

Bunu çalıştırdığımızda her iki future’ın da tamamlanana kadar çalıştığını görürüz:

birinci görevden merhaba sayı 1!
ikinci görevden merhaba sayı 1!
birinci görevden merhaba sayı 2!
ikinci görevden merhaba sayı 2!
birinci görevden merhaba sayı 3!
ikinci görevden merhaba sayı 3!
birinci görevden merhaba sayı 4!
ikinci görevden merhaba sayı 4!
birinci görevden merhaba sayı 5!
birinci görevden merhaba sayı 6!
birinci görevden merhaba sayı 7!
birinci görevden merhaba sayı 8!
birinci görevden merhaba sayı 9!

Artık her seferinde birebir aynı sırayı görürsünüz; bu durum thread’lerle ve Liste 17-7’deki trpl::spawn_task ile gördüğümüzden çok farklıdır. Bunun nedeni trpl::join fonksiyonunun adil (fair) olmasıdır; yani her bir future’ı eşit sıklıkta, aralarında sırayla geçiş yaparak kontrol eder ve diğeri hazırken birinin hızla öne geçip fark açmasına asla izin vermez. Thread’lerde hangi thread’in kontrol edileceğine ve ne kadar süre çalışmasına izin verileceğine işletim sistemi karar verir. Async Rust’ta ise hangi görevin kontrol edileceğine çalışma zamanı karar verir. (Pratikte ayrıntılar karmaşıklaşır; çünkü bir asenkron çalışma zamanı eşzamanlılığı yönetme biçiminin bir parçası olarak perde arkasında işletim sistemi thread’lerini kullanabilir, dolayısıyla adilliği garanti etmek bir çalışma zamanı için daha fazla iş anlamına gelebilir — ama yine de mümkündür!) Çalışma zamanları herhangi bir işlem için adilliği garanti etmek zorunda değildir ve adillik isteyip istemediğinizi seçebilmeniz için çoğu zaman farklı API’ler sunarlar.

Future’ları beklemeye yönelik şu çeşitlemelerden bazılarını deneyip ne yaptıklarını görün:

• Döngülerden birinin veya her ikisinin etrafındaki async bloğu kaldırın.
• Her bir async bloğu tanımladıktan hemen sonra await ile bekleyin.
• Yalnızca ilk döngüyü bir async bloğa sarın ve ortaya çıkan future’ı ikinci döngünün gövdesinden sonra bekleyin.

Ek bir meydan okuma olarak, kodu çalıştırmadan önce her bir durumda çıktının ne olacağını tahmin edip edemeyeceğinize bakın!

Mesaj İletimiyle İki Görev Arasında Veri Gönderme

Future’lar arasında veri paylaşmak da tanıdık gelecektir: yine mesaj iletimini kullanacağız, ancak bu kez türlerin ve fonksiyonların asenkron sürümleriyle. Thread tabanlı ve future tabanlı eşzamanlılık arasındaki bazı temel farkları göstermek için Bölüm 16’daki “Mesaj İletimiyle Thread’ler Arası Veri Transferi” başlığında izlediğimizden biraz farklı bir yol izleyeceğiz. Orada ayrı bir thread başlatmıştık; Liste 17-9’da ise ayrı bir görev bile başlatmadan, yalnızca tek bir async blokla başlayacağız.

extern crate trpl; // required for mdbook test

fn main() {
    trpl::block_on(async {
        let (gonderici, mut alici) = trpl::channel();

        let deger = String::from("merhaba");
        gonderici.send(deger).unwrap();

        let alinan = alici.recv().await.unwrap();
        println!("alındı '{alinan}'");
    });
}

Burada, Bölüm 16’da thread’lerle kullandığımız çoklu üretici, tek tüketici kanal API’sinin asenkron bir sürümü olan trpl::channel fonksiyonunu kullanıyoruz. API’nin asenkron sürümü thread tabanlı sürümden yalnızca biraz farklıdır: değiştirilemez yerine değiştirilebilir bir alici (mut alici) kullanır ve recv metodu değeri doğrudan üretmek yerine beklememiz gereken bir future üretir. Artık göndericiden alıcıya mesaj gönderebiliriz. Ayrı bir thread veya hatta ayrı bir görev başlatmamız gerekmediğine dikkat edin; yalnızca alici.recv çağrısını beklememiz yeterlidir.

std::sync::mpsc modülündeki senkron Receiver::recv metodu bir mesaj alana kadar bloklar. trpl::Receiver::recv metodu ise asenkron olduğu için bloklamaz. Bloklamak yerine, bir mesaj alınana ya da kanalın gönderici tarafı kapanana kadar denetimi çalışma zamanına geri verir. Buna karşılık send çağrısını beklemeyiz, çünkü bloklamaz. Bloklamasına da gerek yoktur; çünkü içine gönderim yaptığımız kanal sınırsızdır (unbounded).

Bu örnekle ilgili iki şeye dikkat edin. Birincisi, mesaj anında ulaşacaktır. İkincisi, burada bir future kullanmamıza rağmen henüz eşzamanlılık yoktur. Listedeki her şey, tıpkı işin içinde hiç future yokmuş gibi sırayla gerçekleşir.

İlk kısmı, Liste 17-10’da gösterildiği gibi bir dizi mesaj gönderip aralarında uyuyarak ele alalım.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (gonderici, mut alici) = trpl::channel();

        let degerler = vec![
            String::from("merhaba"),
            String::from("gelecekten"),
            String::from("selam"),
            String::from("var"),
        ];

        for deger in degerler {
            gonderici.send(deger).unwrap();
            trpl::sleep(Duration::from_millis(500)).await;
        }

        while let Some(alinan) = alici.recv().await {
            println!("alındı '{alinan}'");
        }
    });
}

Mesajları göndermenin yanı sıra onları almamız da gerekir. Bu örnekte kaç mesaj geleceğini bildiğimiz için, alici.recv().await çağrısını dört kez yazarak bunu elle yapabilirdik. Ancak gerçek dünyada genellikle bilinmeyen sayıda mesajı bekleriz; bu nedenle daha fazla mesaj kalmadığını belirleyene kadar beklemeye devam etmemiz gerekir.

Liste 16-10’da, senkron bir kanaldan alınan tüm öğeleri işlemek için bir for döngüsü kullanmıştık. Ne var ki Rust’ta henüz asenkron olarak üretilen bir öğe serisiyle for döngüsü kullanmanın bir yolu yoktur; bu yüzden daha önce görmediğimiz bir döngü kullanmamız gerekir: while let koşullu döngüsü. Bu yapı, Bölüm 6’daki “if let ve let...else ile Özlü Kontrol Akışı” başlığında gördüğümüz if let yapısının döngü sürümüdür. Döngü, belirttiği desen değerle eşleşmeye devam ettiği sürece çalışmayı sürdürür.

alici.recv çağrısı bir future üretir ve biz de onu bekleriz. Çalışma zamanı, hazır olana dek future’ı duraklatacaktır. Future, her mesaj ulaştığında Some(alinan) olarak çözülecektir. Kanal kapandığında ise, herhangi bir mesaj ulaşıp ulaşmadığına bakılmaksızın, artık başka değer kalmadığını ve dolayısıyla yoklamayı —yani beklemeyi— durdurmamız gerektiğini belirtmek üzere future bunun yerine None olarak çözülecektir.

while let döngüsü tüm bunları bir araya getirir. alici.recv().await çağrısının sonucu Some(alinan) ise mesaja erişim sağlarız ve tıpkı if let ile yapabildiğimiz gibi onu döngü gövdesinde kullanabiliriz. Sonuç None ise döngü sona erer. Döngü her tamamlandığında yeniden bekleme noktasına ulaşır; böylece çalışma zamanı bir başka mesaj gelene kadar onu tekrar duraklatır.

Kod artık tüm mesajları başarıyla gönderip alır. Ne yazık ki hâlâ birkaç sorun vardır. Birincisi, mesajlar yarım saniyelik aralıklarla gelmez. Programı başlattıktan 2 saniye (2.000 milisaniye) sonra hepsi bir anda gelir. İkincisi, bu program hiçbir zaman sonlanmaz! Bunun yerine sonsuza dek yeni mesajlar bekler. Programı ctrl-C kullanarak kapatmanız gerekecektir.

Tek Bir Async Blok İçindeki Kod Doğrusal Çalışır

Mesajların neden her birinin arasında gecikmelerle gelmek yerine, toplam gecikmenin sonunda hep birlikte bir anda geldiğini inceleyerek başlayalım. Belirli bir async blok içinde, await anahtar kelimelerinin kodda göründüğü sıra, aynı zamanda program çalışırken yürütüldükleri sıradır.

Liste 17-10’da yalnızca tek bir async blok vardır; dolayısıyla içindeki her şey doğrusal olarak çalışır. Hâlâ eşzamanlılık yoktur. Tüm gonderici.send çağrıları, aralarına serpiştirilmiş tüm trpl::sleep çağrıları ve onlara bağlı bekleme noktalarıyla birlikte sırayla gerçekleşir. while let döngüsü ancak ondan sonra recv çağrıları üzerindeki bekleme noktalarından herhangi birinden geçebilir.

Uyku gecikmesinin her mesaj arasında gerçekleştiği istediğimiz davranışı elde etmek için, Liste 17-11’de gösterildiği gibi gonderici ve alici işlemlerini kendi async bloklarına koymamız gerekir. Böylece çalışma zamanı, tıpkı Liste 17-8’de olduğu gibi trpl::join kullanarak bunların her birini ayrı ayrı yürütebilir. Bir kez daha, tekil future’ları değil, trpl::join çağrısının sonucunu bekliyoruz. Tekil future’ları sırayla bekleseydik, yine sıralı bir akışa geri dönerdik — ki bu tam olarak yapmamaya çalıştığımız şeydir.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (gonderici, mut alici) = trpl::channel();

        let gonderici_gelecegi = async {
            let degerler = vec![
                String::from("merhaba"),
                String::from("gelecekten"),
                String::from("selam"),
                String::from("var"),
            ];

            for deger in degerler {
                gonderici.send(deger).unwrap();
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let alici_gelecegi = async {
            while let Some(alinan) = alici.recv().await {
                println!("alındı '{alinan}'");
            }
        };

        trpl::join(gonderici_gelecegi, alici_gelecegi).await;
    });
}

Liste 17-11’deki güncellenmiş kodla birlikte mesajlar, 2 saniye sonra bir çırpıda topluca gelmek yerine 500 milisaniyelik aralıklarla yazdırılır.

Sahipliği Bir Async Bloğa Taşıma

Yine de program, while let döngüsünün trpl::join ile etkileşime girme biçimi yüzünden hâlâ hiçbir zaman sonlanmaz:

• trpl::join çağrısından dönen future, ancak kendisine geçirilen her iki future da tamamlandığında tamamlanır.
• gonderici_gelecegi future’ı, degerler içindeki son mesajı gönderdikten sonraki uykusunu bitirince tamamlanır.
• alici_gelecegi future’ı, while let döngüsü sona erene kadar tamamlanmaz.
• while let döngüsü, alici.recv çağrısını beklemek None üretene kadar sona ermez.
• alici.recv çağrısını beklemek, yalnızca kanalın diğer ucu kapandığında None döndürür.
• Kanal yalnızca alici.close çağırırsak ya da gönderici taraf olan gonderici düşürüldüğünde kapanır.
• Hiçbir yerde alici.close çağırmıyoruz ve gonderici, trpl::block_on fonksiyonuna geçirilen en dıştaki async blok sona erene kadar düşürülmeyecektir.
• En dıştaki blok sona eremez; çünkü trpl::join çağrısının tamamlanmasını bekleyerek bloklanmıştır — bu da bizi bu listenin en başına geri götürür.

Şu anda mesajları gönderdiğimiz async blok, mesaj göndermek sahiplik gerektirmediği için gonderici değişkenini yalnızca ödünç alır; ancak gonderici değişkenini bu async bloğun içine taşıyabilseydik, söz konusu blok sona erdiğinde düşürülürdü. Bölüm 13’teki “Referansları Yakalamak veya Sahipliği Taşımak” başlığında kapanışlarla move anahtar kelimesinin nasıl kullanılacağını öğrenmiştiniz ve Bölüm 16’daki “Thread’lerle move Kapanışlarını Kullanma” başlığında ele alındığı üzere, thread’lerle çalışırken verileri çoğu zaman kapanışların içine taşımamız gerekir. Aynı temel dinamikler async bloklar için de geçerlidir; dolayısıyla move anahtar kelimesi async bloklarla da tıpkı kapanışlarda olduğu gibi çalışır.

Liste 17-12’de, mesaj göndermek için kullanılan bloğu async yerine async move olarak değiştiriyoruz.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (gonderici, mut alici) = trpl::channel();

        let gonderici_gelecegi = async move {
            // --snip--
            let degerler = vec![
                String::from("merhaba"),
                String::from("gelecekten"),
                String::from("selam"),
                String::from("var"),
            ];

            for deger in degerler {
                gonderici.send(deger).unwrap();
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let alici_gelecegi = async {
            while let Some(alinan) = alici.recv().await {
                println!("alındı '{alinan}'");
            }
        };

        trpl::join(gonderici_gelecegi, alici_gelecegi).await;
    });
}

Kodun bu sürümünü çalıştırdığımızda, son mesaj gönderilip alındıktan sonra düzgün bir şekilde kapanır.

Şimdi, birden fazla future’dan veri göndermek için nelerin değişmesi gerekeceğine bakalım.

join! Makrosuyla Birden Fazla Future’ı Birleştirme

Bu asenkron kanal aynı zamanda çoklu üreticili bir kanal olduğundan, Liste 17-13’te gösterildiği gibi birden fazla future’dan mesaj göndermek istiyorsak gonderici üzerinde clone çağırabiliriz.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (gonderici, mut alici) = trpl::channel();

        let gonderici1 = gonderici.clone();
        let gonderici1_gelecegi = async move {
            let degerler = vec![
                String::from("merhaba"),
                String::from("gelecekten"),
                String::from("selam"),
                String::from("var"),
            ];

            for deger in degerler {
                gonderici1.send(deger).unwrap();
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let alici_gelecegi = async {
            while let Some(alinan) = alici.recv().await {
                println!("alındı '{alinan}'");
            }
        };

        let gonderici_gelecegi = async move {
            let degerler = vec![
                String::from("daha"),
                String::from("fazla"),
                String::from("mesaj"),
                String::from("sana"),
            ];

            for deger in degerler {
                gonderici.send(deger).unwrap();
                trpl::sleep(Duration::from_millis(1500)).await;
            }
        };

        trpl::join!(gonderici1_gelecegi, gonderici_gelecegi, alici_gelecegi);
    });
}

İlk olarak, birinci async bloğun dışında gonderici değişkenini klonlayarak gonderici1 oluşturuyoruz. Tıpkı daha önce gonderici ile yaptığımız gibi gonderici1 değerini bu bloğun içine taşıyoruz. Ardından orijinal gonderici değerini yeni bir async bloğa taşıyor ve orada biraz daha yavaş bir gecikmeyle daha fazla mesaj gönderiyoruz. Bu yeni async bloğu tesadüfen mesajları alan async bloktan sonraya koyduk, ancak ondan önceye de koyabilirdik. Burada kilit nokta future’ların hangi sırayla oluşturulduğu değil, hangi sırayla beklendiğidir.

Mesaj gönderen async blokların her ikisinin de async move bloğu olması gerekir ki bu bloklar bittiğinde hem gonderici hem de gonderici1 düşürülsün. Aksi takdirde, başladığımız aynı sonsuz döngünün içine geri döneriz.

Son olarak, ek future’ı ele almak için trpl::join yerine trpl::join! makrosuna geçiyoruz: join! makrosu, future sayısını derleme zamanında bildiğimiz durumlarda keyfî sayıda future’ı bekler. Sayısı bilinmeyen bir future koleksiyonunu beklemeyi bu bölümün ilerleyen kısımlarında ele alacağız.

Artık her iki gönderici future’dan gelen tüm mesajları görürüz ve gönderici future’lar gönderim sonrasında biraz farklı gecikmeler kullandığından, mesajlar da bu farklı aralıklarla alınır:

alındı 'merhaba'
alındı 'daha'
alındı 'gelecekten'
alındı 'selam'
alındı 'fazla'
alındı 'var'
alındı 'mesaj'
alındı 'sana'

Future’lar arasında veri göndermek için mesaj iletiminin nasıl kullanılacağını, bir async blok içindeki kodun nasıl sıralı çalıştığını, sahipliğin bir async bloğa nasıl taşınacağını ve birden fazla future’ın nasıl birleştirileceğini inceledik. Şimdi, çalışma zamanına başka bir göreve geçebileceğini nasıl ve neden söyleyeceğimizi ele alalım.