Threads | Faisal Affan

Versi Bahasa Indonesia. Baca dalam Bahasa Inggris

Rust menyediakan dukungan bawaan untuk pemrograman konkuren melalui modul standard library std::thread.
Kita dapat membuat thread menggunakan std::thread::spawn, yang menerima sebuah closure dan menjalankannya di thread baru.
Dalam program multi-threaded, jika kita membutuhkan shared mutable state, kita dapat menggunakan:
- Arc<T>: Sebuah atomic reference-counted smart pointer untuk shared ownership di seluruh thread.
- Mutex<T>: Sebuah primitive sinkronisasi untuk mutual exclusion agar dapat mengakses data bersama dengan aman.
Rust menyediakan channels untuk komunikasi thread di modul std::sync::mpsc.
- mpsc::channel membuat transmitter (Sender) dan receiver (Receiver).
- Pesan dikirim melalui transmitter dan diterima melalui receiver.
- Sender dapat di-clone untuk send ke channel yang sama beberapa kali, tetapi hanya satu Receiver yang didukung.
Referensi:
- Using Threads to Run Code Simultaneously
- Using Message Passing to Transfer Data Between Threads

threads1.rs

// This program spawns multiple threads that each runs for at least 250ms, and
// each thread returns how much time it took to complete. The program should
// wait until all the spawned threads have finished and should collect their
// return values into a vector.

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

fn main() {
    let mut handles = Vec::new();
    for i in 0..10 {
        let handle = thread::spawn(move || {
            let start = Instant::now();
            thread::sleep(Duration::from_millis(250));
            println!("Thread {i} done");
            start.elapsed().as_millis()
        });
        handles.push(handle);
    }

    let mut results = Vec::new();
    for handle in handles {
        // TODO: Collect the results of all threads into the `results` vector.
        // Use the `JoinHandle` struct which is returned by `thread::spawn`.
        results.push(handle.join().unwrap());
    }

    if results.len() != 10 {
        panic!("Oh no! Some thread isn't done yet!");
    }

    println!();
    for (i, result) in results.into_iter().enumerate() {
        println!("Thread {i} took {result}ms");
    }
}

Di latihan ini kita hanya perlu menunggu thread yang dibuat selesai, mendapatkan hasilnya, dan push ke results.
Kita dapat menunggu thread selesai menggunakan method join.

pub fn join(self) -> Result<T>: Menunggu thread yang terkait selesai. Fungsi ini akan kembali segera jika thread yang terkait sudah selesai.
```
results.push(handle.join().unwrap());
```

threads2.rs

// Building on the last exercise, we want all of the threads to complete their
// work. But this time, the spawned threads need to be in charge of updating a
// shared value: `JobStatus.jobs_done`

use std::{
    sync::{Arc, Mutex},
    thread,
    time::Duration,
};

struct JobStatus {
    jobs_done: u32,
}

fn main() {
    // Use Arc and Mutex
    let status = Arc::new(Mutex::new(JobStatus { jobs_done: 0 }));

    let mut handles = Vec::new();
    for _ in 0..10 {
        let status_shared = Arc::clone(&status);
        let handle = thread::spawn(move || {
            thread::sleep(Duration::from_millis(250));

            // Lock and update jobs_done
            status_shared.lock().unwrap().jobs_done += 1
        });
        handles.push(handle);
    }

    // Waiting for all jobs to complete.
    for handle in handles {
        handle.join().unwrap();
    }

    // Print the value of `JobStatus.jobs_done`.
    println!("Jobs done: {}", status.lock().unwrap().jobs_done);
}

Di latihan ini menggunakan Arc saja tidak cukup karena kita juga membutuhkan mutability.
Jadi kita harus menambahkan Mutex, sebuah primitive mutual exclusion yang berguna untuk melindungi data bersama.

Mutex akan memblokir thread yang menunggu lock tersedia. Mutex dapat dibuat melalui constructor new. Setiap mutex memiliki parameter tipe yang mewakili data yang dilindunginya. Data hanya dapat diakses melalui RAII guards yang dikembalikan dari lock dan try_lock, yang menjamin bahwa data hanya dapat diakses ketika mutex di-lock.

Pertama kita tambahkan mutex seperti ini:

let status = Arc::new(Mutex::new(JobStatus { jobs_done: 0 }));

Kemudian di dalam block thread yang dibuat kita mendapatkan lock dan memperbarui jobs_done seperti ini:
```
status_shared.lock().unwrap().jobs_done += 1
```
Untuk mendapatkan nilai untuk print kita dapat melakukan hal yang sama seperti di atas:
```
println!("Jobs done: {}", status.lock().unwrap().jobs_done);
```

threads3.rs

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

struct Queue {
    first_half: Vec<u32>,
    second_half: Vec<u32>,
}

impl Queue {
    fn new() -> Self {
        Self {
            first_half: vec![1, 2, 3, 4, 5],
            second_half: vec![6, 7, 8, 9, 10],
        }
    }
}

fn send_tx(q: Queue, tx: mpsc::Sender<u32>) {
    // Clone the sender
    let tx1 = tx.clone();
    thread::spawn(move || {
        for val in q.first_half {
            println!("Sending {val:?}");
            tx1.send(val).unwrap();
            thread::sleep(Duration::from_millis(250));
        }
    });

    thread::spawn(move || {
        for val in q.second_half {
            println!("Sending {val:?}");
            tx.send(val).unwrap();
            thread::sleep(Duration::from_millis(250));
        }
    });
}

fn main() {
    // You can optionally experiment here.
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn threads3() {
        let (tx, rx) = mpsc::channel();
        let queue = Queue::new();

        send_tx(queue, tx);

        let mut received = Vec::with_capacity(10);
        for value in rx {
            received.push(value);
        }

        received.sort();
        assert_eq!(received, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
    }
}

Di latihan ini kita belajar cara menggunakan channel.

Karena kita akan menggunakan dua thread untuk mengirim data, kita perlu meng-clone Sender dan menggunakannya di thread pertama.

Sender dapat di-clone untuk send ke channel yang sama beberapa kali, tetapi hanya satu Receiver yang didukung.

let tx1 = tx.clone();
thread::spawn(move || {
    for val in q.first_half {
        println!("Sending {val:?}");
        tx1.send(val).unwrap();
        thread::sleep(Duration::from_millis(250));
    }
});