Rust Programlama Dili

Herhangi Bir Sayıda Future ile Çalışma

Denetimi Çalışma Zamanına Geri Vermek

“İlk Async Programımız” bölümünden hatırlayacağınız üzere Rust, her await noktasında beklenen future henüz hazır değilse çalışma zamanına görevi duraklatıp başka bir göreve geçme fırsatı verir. Bunun tersi de doğrudur: Rust, async blokları yalnızca bir await noktasında duraklatır ve denetimi çalışma zamanına geri verir. await noktaları arasındaki her şey senkrondur.

Yani bir async blok içinde herhangi bir await noktası olmadan çok sayıda iş yaparsanız, o future diğer tüm future’ların ilerleme kaydetmesini engeller. Bu durumun zaman zaman bir future’ın diğer future’ları aç bırakması (starving) olarak adlandırıldığını duyabilirsiniz. Bazı durumlarda bu önemli bir sorun olmayabilir. Ancak maliyetli bir kurulum ya da uzun süren bir iş yapıyorsanız veya belirli bir görevi süresiz olarak yürütmeye devam edecek bir future’ınız varsa, denetimi çalışma zamanına ne zaman ve nerede geri vereceğinizi düşünmeniz gerekir.

Gelin aç bırakma problemini göstermek ve ardından bu sorunun nasıl çözüleceğini incelemek üzere uzun süren bir işlemi taklit edelim. Liste 17-14, bir slow fonksiyonu tanıtır.

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        // Daha sonra burada `slow` fonksiyonunu çağıracağız
    });
}

fn slow(isim: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{isim}' {ms}ms çalıştı");
}

Bu kod, trpl::sleep yerine std::thread::sleep kullanır; böylece slow çağrısı mevcut thread’i belirtilen milisaniye süresince bloklar. Gerçek dünyadaki hem uzun süren hem de bloklayıcı işlemleri temsil etmesi için slow fonksiyonunu kullanabiliriz.

Liste 17-15’te, bir çift future içinde bu türden CPU bağımlı (CPU-bound) işler yapmayı taklit etmek için slow fonksiyonunu kullanıyoruz.

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        let a = async {
            println!("'a' başladı.");
            slow("a", 30);
            slow("a", 10);
            slow("a", 20);
            trpl::sleep(Duration::from_millis(50)).await;
            println!("'a' bitti.");
        };

        let b = async {
            println!("'b' başladı.");
            slow("b", 75);
            slow("b", 10);
            slow("b", 15);
            slow("b", 350);
            trpl::sleep(Duration::from_millis(50)).await;
            println!("'b' bitti.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(isim: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{isim}' {ms}ms çalıştı");
}

Her bir future, denetimi çalışma zamanına ancak bir dizi yavaş işlemi tamamladıktan sonra geri verir. Bu kodu çalıştırdığınızda şu çıktıyı görürsünüz:

'a' başladı.
'a' 30ms çalıştı
'a' 10ms çalıştı
'a' 20ms çalıştı
'b' başladı.
'b' 75ms çalıştı
'b' 10ms çalıştı
'b' 15ms çalıştı
'b' 350ms çalıştı
'a' bitti.

İki URL’yi getiren future’ları yarıştırmak için trpl::select kullandığımız Liste 17-5’te olduğu gibi, burada da select yine a biter bitmez sonlanır. Ne var ki iki future içindeki slow çağrıları arasında hiçbir nöbetleşe çalışma (interleaving) yoktur. a future’ı, trpl::sleep çağrısı beklenene (await) kadar tüm işini tek seferde yapar; ardından b future’ı kendi trpl::sleep çağrısı beklenene kadar tüm işini yapar ve son olarak a future’ı tamamlanır. Her iki future’ın da yavaş görevleri arasında ilerleme kaydedebilmesine olanak tanımak için denetimi çalışma zamanına geri verebileceğimiz await noktalarına ihtiyacımız vardır. Bu da await edebileceğimiz bir şeye ihtiyacımız olduğu anlamına gelir!

Bu tür bir denetim devrinin Liste 17-15’te şimdiden gerçekleştiğini görebiliriz: a future’ının sonundaki trpl::sleep çağrısını kaldırsaydık, b future’ı hiç çalışmadan a tamamlanırdı. Gelin Liste 17-16’da gösterildiği gibi, işlemlerin sırayla ilerleme kaydetmesini sağlamak üzere başlangıç noktası olarak trpl::sleep fonksiyonunu kullanmayı deneyelim.

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        let bir_ms = Duration::from_millis(1);

        let a = async {
            println!("'a' başladı.");
            slow("a", 30);
            trpl::sleep(bir_ms).await;
            slow("a", 10);
            trpl::sleep(bir_ms).await;
            slow("a", 20);
            trpl::sleep(bir_ms).await;
            println!("'a' bitti.");
        };

        let b = async {
            println!("'b' başladı.");
            slow("b", 75);
            trpl::sleep(bir_ms).await;
            slow("b", 10);
            trpl::sleep(bir_ms).await;
            slow("b", 15);
            trpl::sleep(bir_ms).await;
            slow("b", 350);
            trpl::sleep(bir_ms).await;
            println!("'b' bitti.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(isim: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{isim}' {ms}ms çalıştı");
}

Her slow çağrısının arasına await noktaları içeren trpl::sleep çağrıları ekledik. Artık iki future’ın yaptığı iş nöbetleşe biçimde iç içe geçer:

'a' başladı.
'a' 30ms çalıştı
'b' başladı.
'b' 75ms çalıştı
'a' 10ms çalıştı
'b' 10ms çalıştı
'a' 20ms çalıştı
'b' 15ms çalıştı
'a' bitti.

a future’ı, trpl::sleep çağırmadan önce slow fonksiyonunu çağırdığı için denetimi b’ye devretmeden önce yine bir süre çalışır; ancak bundan sonra future’lar, içlerinden biri her await noktasına ulaştığında sırayı birbirine devreder. Bu örnekte bunu her slow çağrısından sonra yaptık, ancak işi bize en mantıklı gelen herhangi bir biçimde de parçalara bölebilirdik.

Ancak burada gerçekte uyumak istemiyoruz: olabildiğince hızlı ilerleme kaydetmek istiyoruz. Tek ihtiyacımız denetimi çalışma zamanına geri vermektir. Bunu trpl::yield_now fonksiyonunu kullanarak doğrudan yapabiliriz. Liste 17-17’de, tüm trpl::sleep çağrılarını trpl::yield_now ile değiştiriyoruz.

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        let a = async {
            println!("'a' başladı.");
            slow("a", 30);
            trpl::yield_now().await;
            slow("a", 10);
            trpl::yield_now().await;
            slow("a", 20);
            trpl::yield_now().await;
            println!("'a' bitti.");
        };

        let b = async {
            println!("'b' başladı.");
            slow("b", 75);
            trpl::yield_now().await;
            slow("b", 10);
            trpl::yield_now().await;
            slow("b", 15);
            trpl::yield_now().await;
            slow("b", 350);
            trpl::yield_now().await;
            println!("'b' bitti.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(isim: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{isim}' {ms}ms çalıştı");
}

Bu kod hem gerçek niyetimizi daha açık biçimde ortaya koyar hem de sleep kullanmaktan belirgin ölçüde daha hızlı olabilir; çünkü sleep’in kullandığı türden zamanlayıcıların ne kadar hassas olabileceği konusunda çoğu zaman sınırlar vardır. Örneğin kullandığımız sleep sürümü, ona bir nanosaniyelik Duration geçirsek bile her zaman en az bir milisaniye uyur. Yine hatırlatalım, modern bilgisayarlar hızlıdır: bir milisaniye içinde pek çok iş yapabilirler!

Bu durum, programınızın başka neler yaptığına bağlı olarak async’in hesaplama bağımlı (compute-bound) görevler için bile faydalı olabileceği anlamına gelir; çünkü programın farklı bölümleri arasındaki ilişkileri yapılandırmak için kullanışlı bir araç sunar (ancak bunun karşılığında async durum makinesinin getirdiği ek yük maliyeti vardır). Bu bir işbirlikli çoklu görev (cooperative multitasking) biçimidir: her bir future, await noktaları aracılığıyla denetimi ne zaman devredeceğini belirleme gücüne sahiptir. Dolayısıyla her future, çok uzun süre bloklamaktan kaçınma sorumluluğunu da taşır. Rust tabanlı bazı gömülü işletim sistemlerinde bu, tek çoklu görev türüdür!

Gerçek dünya kodlarında elbette genellikle her satırda fonksiyon çağrıları ile await noktalarını ardışık olarak sıralamazsınız. Denetimi bu şekilde devretmek nispeten ucuz olsa da bedelsiz değildir. Pek çok durumda, hesaplama bağımlı bir görevi parçalara bölmeye çalışmak onu belirgin biçimde yavaşlatabilir; dolayısıyla bazen genel performans açısından bir işlemin kısa süreliğine bloklamasına izin vermek daha iyidir. Kodunuzun gerçek performans darboğazlarının neler olduğunu görmek için her zaman ölçüm yapın. Yine de eşzamanlı gerçekleşmesini beklediğiniz pek çok işin seri hâlde yürütüldüğünü görüyorsanız, arka plandaki bu dinamiği akılda tutmak son derece önemlidir!

Kendi Async Soyutlamalarımızı Kurmak

Yeni desenler oluşturmak üzere future’ları birbiriyle birleştirebiliriz de. Örneğin elimizde hâlihazırda bulunan async yapı taşlarıyla bir timeout (zaman aşımı) fonksiyonu inşa edebiliriz. İşimiz bittiğinde ortaya çıkan sonuç, daha da fazla async soyutlama oluşturmak için kullanabileceğimiz bir başka yapı taşı olacaktır.

Liste 17-18, bu timeout fonksiyonunun yavaş bir future ile nasıl çalışmasını beklediğimizi gösterir.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let yavas = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Sonunda bitti"
        };

        match timeout(yavas, Duration::from_secs(2)).await {
            Ok(mesaj) => println!("'{mesaj}' ile başarılı oldu"),
            Err(sure) => {
                println!("{} saniye sonra başarısız oldu", sure.as_secs())
            }
        }
    });
}

Gelin bunu uygulayalım! Başlamak için timeout fonksiyonunun API’sini düşünelim:

• Onu await edebilmemiz için kendisinin de bir async fonksiyon olması gerekir.
• İlk parametresi çalıştırılacak bir future olmalıdır. Herhangi bir future ile çalışmasına olanak tanımak için onu jenerik yapabiliriz.
• İkinci parametresi beklenecek azami süre olacaktır. Bir Duration kullanırsak, bu değeri trpl::sleep fonksiyonuna iletmek kolaylaşır.
• Bir Result döndürmelidir. Future başarıyla tamamlanırsa Result, future’ın ürettiği değeri içeren Ok olur. Zaman aşımı süresi daha önce dolarsa Result, zaman aşımının beklediği süreyi içeren Err olur.

Liste 17-19 bu bildirimi gösterir.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let yavas = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Sonunda bitti"
        };

        match timeout(yavas, Duration::from_secs(2)).await {
            Ok(mesaj) => println!("'{mesaj}' ile başarılı oldu"),
            Err(sure) => {
                println!("{} saniye sonra başarısız oldu", sure.as_secs())
            }
        }
    });
}

async fn timeout<F: Future>(
    denenecek_future: F,
    azami_sure: Duration,
) -> Result<F::Output, Duration> {
    // Uygulamamız buraya gelecek!
}

Bu imza, türlere ilişkin hedeflerimizi karşılar. Şimdi ihtiyacımız olan davranışı düşünelim: içeriye geçirilen future’ı süreye karşı yarıştırmak istiyoruz. Süreden bir zamanlayıcı future oluşturmak için trpl::sleep fonksiyonunu, bu zamanlayıcıyı çağıranın geçirdiği future ile birlikte çalıştırmak için de trpl::select fonksiyonunu kullanabiliriz.

Liste 17-20’de, trpl::select çağrısını beklemenin (await) sonucu üzerinde desen eşleştirerek timeout fonksiyonunu uyguluyoruz.

extern crate trpl; // required for mdbook test

use std::time::Duration;

use trpl::Either;

// --snip--

fn main() {
    trpl::block_on(async {
        let yavas = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Sonunda bitti"
        };

        match timeout(yavas, Duration::from_secs(2)).await {
            Ok(mesaj) => println!("'{mesaj}' ile başarılı oldu"),
            Err(sure) => {
                println!("{} saniye sonra başarısız oldu", sure.as_secs())
            }
        }
    });
}

async fn timeout<F: Future>(
    denenecek_future: F,
    azami_sure: Duration,
) -> Result<F::Output, Duration> {
    match trpl::select(denenecek_future, trpl::sleep(azami_sure)).await {
        Either::Left(cikti) => Ok(cikti),
        Either::Right(_) => Err(azami_sure),
    }
}

trpl::select uygulaması adil (fair) değildir: argümanları her zaman geçirildikleri sırayla yoklar (diğer select uygulamaları ilk olarak hangi argümanı yoklayacağını rastgele seçer). Bu nedenle select fonksiyonuna ilk olarak denenecek_future argümanını geçiriyoruz; böylece azami_sure çok kısa bir süre olsa bile denenecek_future tamamlanma şansı elde eder. denenecek_future önce biterse select, denenecek_future çıktısını içeren Left varyantını döndürür. Zamanlayıcı önce biterse select, zamanlayıcının çıktısı olan () değerini içeren Right varyantını döndürür.

denenecek_future başarılı olur ve bir Left(cikti) elde edersek Ok(cikti) döndürürüz. Uyku zamanlayıcısının süresi önce dolar ve bir Right(()) elde edersek, () değerini _ ile yok sayıp Err(azami_sure) döndürürüz.

Böylece diğer iki async yardımcı üzerine inşa edilmiş, çalışan bir timeout fonksiyonuna sahip olduk. Kodumuzu çalıştırdığımızda zaman aşımının ardından başarısızlık durumunu ekrana yazdırır:

2 saniye sonra başarısız oldu

Future’lar diğer future’larla birleştirilebildiğinden, daha küçük async yapı taşlarını kullanarak gerçekten güçlü araçlar inşa edebilirsiniz. Örneğin aynı yaklaşımı zaman aşımlarını yeniden denemelerle (retries) birleştirmek için kullanabilir ve bunları da ağ çağrıları gibi işlemlerle (Liste 17-5’tekiler gibi) birlikte çalıştırabilirsiniz.

Pratikte çoğunlukla doğrudan async ve await ile, ikincil olarak da en dıştaki future’ların nasıl yürütüleceğini denetlemek üzere select gibi fonksiyonlar ve join! makrosu gibi makrolarla çalışırsınız.

Şimdiye kadar aynı anda birden fazla future ile çalışmanın çeşitli yollarını gördük. Sırada, zaman içinde art arda gelen birden fazla future ile akışlar (streams) aracılığıyla nasıl çalışabileceğimize bakacağız.