Rust Programlama Dili

Mesaj İletimiyle Thread’ler Arası Veri Transferi

Güvenli eşzamanlılığı sağlamanın giderek popülerleşen yaklaşımlarından biri, thread’lerin veya aktörlerin birbirlerine veri içeren mesajlar göndererek iletişim kurduğu mesaj iletimi (message passing) yöntemidir. Go dili belgelerindeki bir slogan bu fikri şöyle özetler: “Belleği paylaşarak iletişim kurmayın; bunun yerine iletişim kurarak belleği paylaşın.”

Mesaj göndermeye dayalı eşzamanlılığı gerçekleştirmek için Rust’ın standart kütüphanesi bir kanal uygulaması sunar. Kanal (channel), verinin bir thread’den diğerine gönderilmesini sağlayan genel bir programlama kavramıdır.

Programlamadaki bir kanalı, bir dere veya nehir gibi tek yönlü bir su kanalı olarak hayal edebilirsiniz. Bir nehre lastik ördek gibi bir şey bırakırsanız, akıntı yönünde ilerleyerek su yolunun sonuna kadar ulaşacaktır.

Bir kanalın iki yarısı vardır: bir gönderici (transmitter) ve bir alıcı (receiver). Gönderici yarısı, lastik ördeği nehre bıraktığınız akıntının üst kısmındaki konumdur; alıcı yarısı ise lastik ördeğin akıntı aşağısında ulaştığı noktadır. Kodunuzun bir bölümü göndermek istediğiniz verilerle gönderici üzerindeki metotları çağırır; diğer bölümü ise gelen mesajlar için alıcı ucu kontrol eder. Gönderici veya alıcı yarılarından herhangi biri bellekten düşürüldüğünde kanalın kapandığı (closed) söylenir.

Burada, değerler üretip bunları bir kanal üzerinden gönderen bir thread’e ve bu değerleri alıp ekrana yazdıran başka bir thread’e sahip bir programı adım adım oluşturacağız. Bu özelliği göstermek amacıyla bir kanal kullanarak thread’ler arasında basit değerler göndereceğiz. Bu tekniğe aşina olduktan sonra kanalları, birbirleriyle iletişim kurması gereken her türlü thread yapısında kullanabilirsiniz: örneğin bir sohbet sisteminde ya da pek çok thread’in bir hesaplamanın parçalarını gerçekleştirip sonuçları birleştiren tek bir thread’e gönderdiği bir sistemde.

İlk olarak Liste 16-6’da bir kanal oluşturacağız ancak onunla henüz hiçbir şey yapmayacağız. Rust, kanal üzerinden ne tür değerler göndermek istediğimizi henüz anlayamadığı için bu kodun şimdilik derlenmeyeceğine dikkat edin.

use std::sync::mpsc;

fn main() {
    let (gonderici, alici) = mpsc::channel();
}

mpsc::channel fonksiyonunu kullanarak yeni bir kanal oluşturuyoruz; mpsc kısaltması çoklu üretici, tek tüketici (multiple producer, single consumer) anlamına gelir. Kısacası Rust’ın standart kütüphanesinin kanalları uygulama biçimi, bir kanalın değer üreten birden fazla gönderme ucuna sahip olabileceği, ancak bu değerleri tüketen yalnızca tek bir alma ucuna sahip olabileceği anlamına gelir. Birden çok derenin birleşerek tek bir büyük nehre aktığını hayal edin: derelerin herhangi birinden aşağı gönderilen her şey, sonunda tek bir nehirde toplanacaktır. Şimdilik tek bir üreticiyle başlayacağız, ancak bu örneği çalışır hâle getirdiğimizde birden fazla üretici ekleyeceğiz.

mpsc::channel fonksiyonu bir demet (tuple) döndürür; bu demetin ilk elemanı gönderme ucu —yani gönderici—, ikinci elemanı ise alma ucu —yani alıcı— olur. Pek çok alanda transmitter (gönderici) ve receiver (alıcı) terimleri için geleneksel olarak sırasıyla tx ve rx kısaltmaları kullanılır; biz ise kodumuzda her bir ucu açıkça belirtmek üzere değişkenlerimizi gonderici ve alici olarak adlandırıyoruz. Burada demeti parçalarına ayıran (destructuring) bir desenle birlikte bir let deyimi kullanıyoruz; let deyimlerinde desen kullanımını ve parçalarına ayırma işlemini Bölüm 19’da ele alacağız. Şimdilik bir let deyimini bu şekilde kullanmanın, mpsc::channel tarafından döndürülen demetin parçalarını ayıklamak için elverişli bir yaklaşım olduğunu bilmeniz yeterlidir.

Liste 16-7’de gösterildiği gibi, gönderme ucunu oluşturulan bir thread’e taşıyalım ve oluşturulan thread’in ana thread ile iletişim kurması için bir string göndermesini sağlayalım. Bu, nehrin üst kısmından suya bir lastik ördek bırakmaya veya bir thread’den diğerine bir sohbet mesajı göndermeye benzer.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (gonderici, alici) = mpsc::channel();

    thread::spawn(move || {
        let deger = String::from("merhaba");
        gonderici.send(deger).unwrap();
    });
}

Yine yeni bir thread oluşturmak için thread::spawn kullanıyor ve ardından oluşturulan thread’in gonderici değerine sahip olması için move kullanarak gonderici’yi kapanışın içine taşıyoruz. Oluşturulan thread’in kanal üzerinden mesaj gönderebilmesi için göndericiye sahip olması gerekir (ayrıca send çağrısıyla bir String gönderdiğimiz için Rust artık kanalın türünü (Sender<String>, Receiver<String>) olarak otomatik çıkarsayabilir).

Göndericinin, göndermek istediğimiz değeri alan bir send metodu vardır. send metodu bir Result<T, E> türü döndürür; dolayısıyla alıcı daha önce bellekten düşürülmüşse ve değeri gönderecek bir yer kalmamışsa, gönderme işlemi bir hata döndürür. Bu örnekte, bir hata durumunda paniklemek için unwrap çağırıyoruz. Ancak gerçek bir uygulamada bu hatayı uygun şekilde ele alırız: uygun hata yönetimi stratejilerini gözden geçirmek için Bölüm 9’a dönebilirsiniz.

Liste 16-8’de, ana thread’de alıcıdan değeri alacağız. Bu, nehrin sonunda lastik ördeği sudan çıkarmaya veya bir sohbet mesajını almaya benzer.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (gonderici, alici) = mpsc::channel();

    thread::spawn(move || {
        let deger = String::from("merhaba");
        gonderici.send(deger).unwrap();
    });

    let alinan = alici.recv().unwrap();
    println!("Alındı: {alinan}");
}

Alıcının iki kullanışlı metodu vardır: recv ve try_recv. Burada, receive (almak) kelimesinin kısaltması olan recv metodunu kullanıyoruz; bu metot ana thread’in yürütmesini bloklar ve kanal üzerinden bir değer gönderilene kadar bekler. Bir değer gönderildiğinde recv, onu bir Result<T, E> içinde döndürür. Gönderici kapandığında ise recv, artık başka değer gelmeyeceğini bildirmek üzere bir hata döndürür.

try_recv metodu ise bloklamaz; bunun yerine hemen bir Result<T, E> döndürür: mevcut bir mesaj varsa onu tutan bir Ok değeri, o sırada hiç mesaj yoksa bir Err değeri döner. Bu thread’in mesajları beklerken yapacağı başka işler varsa try_recv kullanmak faydalıdır: belirli aralıklarla try_recv çağıran, mevcut bir mesaj varsa onu işleyen, aksi takdirde tekrar kontrol edene kadar bir süre başka işler yapan bir döngü yazabiliriz.

Bu örnekte sadelik adına recv kullandık; ana thread’in mesajları beklemek dışında yapacağı başka bir iş olmadığından ana thread’i bloklamak uygundur.

Liste 16-8’deki kodu çalıştırdığımızda, ana thread’den yazdırılan değeri görürüz:

Alındı: merhaba

Mükemmel!

Kanallar Üzerinden Sahiplik Aktarma

Sahiplik kuralları mesaj gönderiminde hayati bir rol oynar; çünkü güvenli ve eşzamanlı kod yazmanıza yardımcı olur. Eşzamanlı programlamada hataları önlemek, Rust programlarınız boyunca sahipliği hesaba katmanın sağladığı avantajdır. Kanalların ve sahipliğin sorunları önlemek üzere birlikte nasıl çalıştığını göstermek için bir deney yapalım: oluşturulan thread’de deger değişkenini kanaldan gönderdikten sonra kullanmaya çalışacağız. Bu koda neden izin verilmediğini görmek için Liste 16-9’daki kodu derlemeyi deneyin.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (gonderici, alici) = mpsc::channel();

    thread::spawn(move || {
        let deger = String::from("merhaba");
        gonderici.send(deger).unwrap();
        println!("değer şu: {deger}");
    });

    let alinan = alici.recv().unwrap();
    println!("Alındı: {alinan}");
}

Burada, gonderici.send aracılığıyla kanaldan gönderdikten sonra deger değişkenini yazdırmaya çalışıyoruz. Buna izin vermek kötü bir fikir olurdu: değer başka bir thread’e gönderildikten sonra, biz değeri tekrar kullanmaya çalışmadan önce o thread onu değiştirebilir veya bellekten düşürebilirdi. Diğer thread’in yaptığı değişiklikler, tutarsız veya var olmayan veriler yüzünden potansiyel olarak hatalara ya da beklenmedik sonuçlara yol açabilirdi. Ancak Liste 16-9’daki kodu derlemeye çalıştığımızda Rust bize bir hata verir:

$ cargo run
   Compiling mesaj-iletimi v0.1.0 (file:///projects/mesaj-iletimi)
error[E0382]: borrow of moved value: `deger`
  --> src/main.rs:10:30
   |
 8 |         let deger = String::from("merhaba");
   |             ----- move occurs because `deger` has type `String`, which does not implement the `Copy` trait
 9 |         gonderici.send(deger).unwrap();
   |                        ----- value moved here
10 |         println!("değer şu: {deger}");
   |                              ^^^^^ value borrowed here after move

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

Eşzamanlılık hatamız derleme zamanı hatasına neden oldu. send fonksiyonu parametresinin sahipliğini alır ve değer taşındığında alıcı onun sahipliğini devralır. Bu durum, değeri gönderdikten sonra yanlışlıkla tekrar kullanmamızı engeller; sahiplik sistemi her şeyin yolunda olduğunu denetler.

Birden Fazla Değer Gönderme

Liste 16-8’deki kod derlenip çalıştı, ancak iki ayrı thread’in kanal üzerinden birbiriyle konuştuğunu bize açıkça göstermedi.

Liste 16-10’da, Liste 16-8’deki kodun eşzamanlı olarak çalıştığını kanıtlayacak bazı değişiklikler yaptık: oluşturulan thread artık birden fazla mesaj gönderecek ve her bir mesaj arasında bir saniye duraklayacak.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (gonderici, alici) = mpsc::channel();

    thread::spawn(move || {
        let degerler = vec![
            String::from("merhaba"),
            String::from("oluşturulan"),
            String::from("iş"),
            String::from("parçacığından"),
        ];

        for deger in degerler {
            gonderici.send(deger).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for alinan in alici {
        println!("Alındı: {alinan}");
    }
}

Bu kez oluşturulan thread’de, ana thread’e göndermek istediğimiz string’lerden oluşan bir vektör bulunuyor. Bunların üzerinde ilerliyor, her birini tek tek gönderiyor ve bir saniyelik Duration değeriyle thread::sleep fonksiyonunu çağırarak her gönderim arasında duraklıyoruz.

Ana thread’de ise artık recv fonksiyonunu açıkça çağırmıyoruz: bunun yerine alici değerini bir iteratör olarak ele alıyoruz. Alınan her değeri ekrana yazdırıyoruz. Kanal kapandığında iterasyon da sona erecektir.

Liste 16-10’daki kodu çalıştırdığınızda, her satır arasında bir saniyelik duraklamayla aşağıdaki çıktıyı görmelisiniz:

Alındı: merhaba
Alındı: oluşturulan
Alındı: iş
Alındı: parçacığından

Ana thread’deki for döngüsünde duraklayan veya gecikme yaratan herhangi bir kod bulunmadığından, ana thread’in oluşturulan thread’den değerleri almak için beklediğini anlayabiliriz.

Birden Fazla Üretici Oluşturma

Daha önce mpsc kısaltmasının çoklu üretici, tek tüketici anlamına geldiğini belirtmiştik. Şimdi mpsc’yi amacına uygun biçimde kullanalım ve Liste 16-10’daki kodu, tamamı aynı alıcıya değerler gönderen birden fazla thread oluşturacak şekilde genişletelim. Bunu Liste 16-11’de gösterildiği gibi göndericiyi klonlayarak yapabiliriz.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    // --snip--

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

    let gonderici1 = gonderici.clone();
    thread::spawn(move || {
        let degerler = vec![
            String::from("merhaba"),
            String::from("oluşturulan"),
            String::from("iş"),
            String::from("parçacığından"),
        ];

        for deger in degerler {
            gonderici1.send(deger).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

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

        for deger in degerler {
            gonderici.send(deger).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for alinan in alici {
        println!("Alındı: {alinan}");
    }

    // --snip--
}

Bu kez, ilk thread’i oluşturmadan önce gönderici üzerinde clone metodunu çağırıyoruz. Bu bize oluşturulan ilk thread’e aktarabileceğimiz yeni bir gönderici verir. Orijinal göndericiyi ise oluşturulan ikinci bir thread’e aktarıyoruz. Böylece elimizde, her biri tek bir alıcıya farklı mesajlar gönderen iki thread olur.

Kodu çalıştırdığınızda çıktınız aşağıdakine benzer görünmelidir:

Alındı: merhaba
Alındı: daha
Alındı: oluşturulan
Alındı: fazla
Alındı: mesaj
Alındı: iş
Alındı: sana
Alındı: parçacığından

Sisteminize bağlı olarak değerleri farklı bir sırada görebilirsiniz. Eşzamanlılığı hem ilginç hem de zor kılan şey tam olarak budur. Farklı thread’lerde thread::sleep fonksiyonuna çeşitli değerler vererek denemeler yaparsanız, her çalıştırma daha belirlenimsiz (nondeterministic) olacak ve her seferinde farklı bir çıktı üretecektir.

Kanalların nasıl çalıştığını incelediğimize göre, şimdi farklı bir eşzamanlılık yöntemine bakalım.