Channels

Topik ini membahas channel pada Rust sebagai sarana komunikasi antar thread dengan cara memindahkan kepemilikan pesan, mulai dari pasangan pengirim dan penerima hasil mpsc::channel, pengiriman lewat send, penerimaan lewat recv, sampai pola banyak pengirim dengan Sender::clone. Pembahasan juga mencakup filosofi message passing sebagai lawan dari state bersama serta penutupan channel secara otomatis ketika seluruh pengirim telah dijatuhkan. Setelah membaca topik ini, aliran data antar thread dapat dirancang sebagai rangkaian pesan tanpa satu pun nilai yang dimiliki dua thread sekaligus.

Filosofi message passing#

Message passing adalah gaya konkurensi yang mengomunikasikan data dengan mengirim kepemilikan nilai, bukan dengan membuka satu memori yang sama bagi banyak thread. Semboyan dari komunitas Go yang dikutip buku resmi Rust meringkasnya: komunikasi dilakukan bukan dengan berbagi memori, melainkan memori dibagikan melalui komunikasi. Pada gaya ini setiap nilai selalu memiliki satu pemilik, dan pemilik berganti hanya ketika pesan dikirim. Karena tidak pernah ada dua pegangan atas data yang sama, aturan pinjaman tidak perlu dijaga dengan kunci, dan pemeriksaan keamanan thread tetap berlangsung sepenuhnya saat kompilasi.

Channel menjadi wadah pemindahan tersebut. Pengirim menyerahkan nilai ke dalam antrean channel, dan penerima mengambil nilai itu sebagai pemilik baru, sehingga data berpindah seperti berpindah antar fungsi lewat nilai kembalian. Modul std::sync::mpsc menyediakan channel bawaan Rust dengan pola multi-producer single-consumer: banyak thread boleh mengirim ke satu channel, tetapi satu channel hanya dilayani satu penerima. Variasi lain seperti banyak penerima tidak disediakan pustaka standar dan tersedia pada crate eksternal seperti crossbeam-channel.

channel, send, dan recv#

Fungsi mpsc::channel() membuat satu channel dan mengembalikan pasangan (Sender, Receiver). Metode send pada pengirim memindahkan nilai ke antrean channel dan mengembalikan Result yang berisi Err apabila penerima sudah tidak ada. Metode recv pada penerima memblokir thread pemanggil hingga pesan tiba, dan nilai kembaliannya juga Result karena channel dapat tertutup sebelum pesan berikutnya ada. Penerimaan berulang dapat ditulis dengan perulangan for karena Receiver mengimplementasikan Iterator yang berhenti dengan sendirinya saat channel tertutup.

Program berikut mengirim tiga nilai suhu dari satu thread pengirim, lalu thread utama menerima semuanya. Urutan baris keluaran pasti karena satu pengirim mengirim secara berurutan dan channel menjaga urutan itu, sedangkan pencetakan dilakukan oleh thread utama selaku satu-satunya penerima.

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

fn main() {
    let (kirim, terima) = mpsc::channel();

    thread::spawn(move || {
        let suhu = [30, 31, 29];
        for nilai in suhu {
            kirim.send(nilai).unwrap();
        }
    });

    for nilai in terima {
        println!("suhu diterima: {}", nilai);
    }
}

suhu diterima: 30
suhu diterima: 31
suhu diterima: 29

Perulangan for nilai in terima berakhir bukan karena jumlah pesan diketahui, melainkan karena thread pengirim selesai dan kirim yang dipindahkan ke thread itu ikut dijatuhkan sehingga channel tertutup. Pemanggilan unwrap pada send aman pada contoh ini karena penerima masih hidup selama program berjalan. Kata kunci move tetap diperlukan pada closure pengirim agar kepemilikan kirim berpindah ke thread itu, sesuai syarat 'static yang berlaku pada setiap spawn.

Banyak pengirim dengan clone#

Sender mengimplementasikan Clone, sehingga beberapa thread dapat mengirim ke channel yang sama. Setiap salinan pengirim adalah pegangan terpisah, dan channel dinyatakan tertutup hanya setelah seluruh salinan itu dijatuhkan. Akibatnya, thread utama yang membuat channel perlu menjatuhkan pegangan asalnya dengan drop apabila seluruh pengirim sudah dipindahkan ke thread lain. Selama pegangan asal masih hidup, channel belum dianggap tertutup dan penerima terus menunggu.

Program berikut membagi tiga pengirim ke tiga thread, menerima enam pesan di thread utama, lalu mencetaknya setelah diurutkan. Pengurutan diperlukan karena urutan kedatangan pesan lintas pengirim tidak ditentukan, sedangkan keluaran contoh harus tetap sama pada setiap jalankan.

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

fn main() {
    let (kirim, terima) = mpsc::channel();

    let mut pegangan = Vec::new();
    for id in 1..=3 {
        let kirim = kirim.clone();
        pegangan.push(thread::spawn(move || {
            for langkah in 1..=2 {
                kirim.send(id * 10 + langkah).unwrap();
            }
        }));
    }
    drop(kirim);

    let mut hasil = Vec::new();
    for nilai in terima {
        hasil.push(nilai);
    }
    for utas in pegangan {
        utas.join().unwrap();
    }
    hasil.sort();

    println!("pesan diterima: {:?}", hasil);
}

pesan diterima: [11, 12, 21, 22, 31, 32]

Nilai 11 dan 12 dikirim thread pertama, 21 dan 22 thread kedua, dan 31 dan 32 thread ketiga. Pasangan dari satu thread pasti berurutan karena pengirimannya berurutan, tetapi susunan antar pasangan berubah mengikuti penjadwal, sehingga pengurutan memulihkan determinisme sebelum pencetakan. Pemanggilan join dilakukan setelah penerimaan selesai karena perulangan for atas terima sendiri sudah menunggu ketiga thread selesai mengirim. Diagram berikut merangkum arsitektur mpsc yang dipakai program ini.

Kepemilikan berpindah dalam saluran#

Metode send mengambil alih kepemilikan nilai, sehingga pengirim kehilangan akses begitu pesan masuk antrean. Sifat ini yang menjadikan channel bebas data race tanpa kunci: setelah dikirim, nilai hanya dapat dipegang oleh penerima, dan pengirim lama tidak memiliki pegangan apa pun untuk membaca atau menulisnya kembali. Kompilator menegakkan batas itu sebagai error biasa, sebagaimana tampak pada program berikut yang mencoba mencetak String yang sudah terkirim.

export PATH="$HOME/.cargo/bin:$PATH"
# nilai yang sudah dikirim tidak dapat dipakai lagi oleh pengirim
printf 'use std::sync::mpsc;\nuse std::thread;\n\nfn main() {\n    let (kirim, _terima) = mpsc::channel();\n\n    thread::spawn(move || {\n        let laporan = String::from("catatan sensor");\n        kirim.send(laporan).unwrap();\n        println!("{}", laporan);\n    })\n    .join()\n    .unwrap();\n}\n' > kirim-lagi.rs
# tampilkan pesan error kompilasi yang timbul
rustc --edition=2021 kirim-lagi.rs 2>&1 | sed -n '1,14p'

error[E0382]: borrow of moved value: `laporan`
  --> kirim-lagi.rs:10:24
   |
 8 |         let laporan = String::from("catatan sensor");
   |             ------- move occurs because `laporan` has type `String`, which does not implement the `Copy` trait
 9 |         kirim.send(laporan).unwrap();
   |                    ------- value moved here
10 |         println!("{}", laporan);
   |                        ^^^^^^^ value borrowed here after move

error: aborting due to 1 previous error

For more information about this error, try `rustc --explain E0382`.

Error E0382 menunjuk baris send sebagai titik perpindahan dan baris println! sebagai pemakaian setelah perpindahan. Apabila pengirim memang perlu membaca kembali isi pesan, nilai harus dikirim sebagai salinan, misalnya lewat clone sebelum send, atau hasil olahan dikirim kembali sebagai pesan baru dari penerima. Dibandingkan dengan membagikan nilai, kedua pilihan itu menjaga tiap jalur data tetap memiliki satu pemilik pada satu waktu.

Pemeriksaan lintas thread tetap berperan pada channel: nilai yang dikirim harus Send, dan penerimanya menjadi pemilik baru yang bebas memproses nilai itu tanpa kunci. Dengan dua jaminan tersebut, alur kerja antar thread dapat dibaca dari arah pesannya, karena data berpindah dari pengirim ke penerima dan hanya dipegang satu tahap pada satu waktu.

Kesalahan umum#

Kesalahan pertama adalah melupakan drop pada pengirim asal di thread utama. Perulangan penerimaan menunggu channel tertutup, sedangkan channel hanya tertutup setelah seluruh pengirim dijatuhkan, sehingga satu pegangan terlupakan membuat program berhenti menunggu tanpa pesan error. Gejalanya mudah dikenali: program tetap hidup tanpa keluaran error, tetapi pencetakan setelah perulangan penerimaan tidak pernah terjadi. Jatuhkan pengirim asal segera setelah seluruh salinan dipindahkan ke thread pengirim.

Kesalahan kedua adalah memanggil unwrap pada recv padahal kehilangan pengirim merupakan keadaan yang wajar. Penerimaan mengembalikan Err ketika seluruh pengirim sudah tidak ada, misalnya karena thread pengirim berhenti lebih awal, dan unwrap mengubah keadaan itu menjadi panic. Cocokkan Ok dan Err secara eksplisit, atau perlakukan penutupan channel sebagai sinyal selesai.

Kesalahan ketiga adalah mengharapkan urutan pesan tertentu lintas pengirim. Channel menjaga urutan pesan dari satu pengirim, tetapi tidak menentukan siapa yang mengirim lebih dahulu di antara beberapa pengirim. Urutkan hasil penerimaan bila urutan akhirnya memengaruhi keluaran program.

Ringkasan#

• mpsc::channel() mengembalikan pasangan Sender dan Receiver: send memindahkan kepemilikan pesan ke antrean, recv memblokir hingga pesan tiba, dan keduanya mengembalikan Result.
• Receiver mengimplementasikan Iterator sehingga perulangan for menerima pesan hingga channel tertutup, yaitu ketika seluruh Sender telah dijatuhkan.
• Sender dapat di-clone untuk banyak pengirim pada satu channel, dan pengirim asal di thread utama perlu dijatuhkan dengan drop agar penutupan channel tidak tertunda.
• Nilai yang sudah dikirim tidak dapat dipakai pengirimnya lagi, dan penegakannya berupa error E0382 seperti perpindahan kepemilikan biasa.

Referensi#

• Using Message Passing to Transfer Data between Threads — The Rust Programming Language — Bab resmi tentang channel mpsc, banyak pengirim, dan perpindahan kepemilikan pesan.
• std::sync::mpsc — Rust standard library — Dokumentasi modul mpsc beserta penjelasan arsitektur multi-producer single-consumer.
• mpsc::Receiver — Rust standard library — Dokumentasi penerima mengenai recv, try_recv, dan implementasi Iterator.