Concurrency dan Async Programming di Rust: Panduan Lengkap

Di artikel sebelumnya kita sudah belajar error handling, traits, dan generic. Sekarang saatnya memasuki dunia concurrency dan async programming di Rust.

Rust adalah bahasa yang sangat unik dalam hal concurrency. Berkat ownership system dan type system yang ketat, Rust bisa menjamin keamanan concurrency di compile time. Tidak ada data races, tidak ada deadlocks yang tersembunyi, tidak ada race condition yang baru muncul di production.

Concurrency vs Parallelism

Sebelum mulai, mari bedakan dua konsep ini:

• Concurrency: Menjalankan beberapa tugas secara bergantian (interleaving). Satu CPU bisa melakukan concurrency.
• Parallelism: Menjalankan beberapa tugas secara bersamaan. Butuh multiple CPU core.

Rust mendukung keduanya dengan sangat baik. Kamu bisa pilih thread untuk parallelism, atau async/await untuk concurrency yang ringan.

Mengapa Concurrency Sulit?

Di bahasa lain, concurrency sering menjadi sumber bug yang paling sulit di-debug:

• Data races: Dua thread mengakses data yang sama, minimal satu menulis
• Deadlocks: Dua thread saling menunggu satu sama lain selamanya
• Race conditions: Hasil program berbeda-beda tergantung urutan eksekusi

Rust mencegah data races dan banyak race conditions di compile time. Jika kode kamu compile, kamu sudah terhindar dari seluruh kategori bug ini.

Thread

Thread adalah unit eksekusi yang berjalan secara independen. Rust menggunakan OS thread, artinya setiap thread Rust adalah thread asli dari sistem operasi.

Membuat Thread

use std::thread;
use std::time::Duration;

fn main() {
    let handle = thread::spawn(|| {
        for i in 1..=5 {
            println!("Thread: {}", i);
            thread::sleep(Duration::from_millis(100));
        }
    });

    for i in 1..=5 {
        println!("Main: {}", i);
        thread::sleep(Duration::from_millis(100));
    }

    handle.join().unwrap(); // Tunggu thread selesai
}

Output akan bercampur karena dua thread berjalan bersamaan. Urutannya berbeda setiap kali program dijalankan.

Join Handle

join() memblokir thread saat ini sampai thread lain selesai:

use std::thread;

fn main() {
    let handle = thread::spawn(|| {
        let mut sum = 0;
        for i in 1..=1000 {
            sum += i;
        }
        sum // Return value dari thread
    });

    let result = handle.join().unwrap();
    println!("Sum: {}", result); // 500500
}

Move Closure

Saat kamu butuh mengirim data ke thread, gunakan move:

use std::thread;

fn main() {
    let data = vec![1, 2, 3, 4, 5];

    let handle = thread::spawn(move || {
        println!("Data di thread: {:?}", data);
        let sum: i32 = data.iter().sum();
        sum
    });

    // println!("{:?}", data); // ERROR! data sudah di-move ke thread
    let result = handle.join().unwrap();
    println!("Sum: {}", result);
}

Tanpa move, compiler akan error karena closure meminjam data yang mungkin tidak hidup cukup lama. Dengan move, ownership dipindahkan ke thread.

Multiple Threads

use std::thread;

fn main() {
    let mut handles = vec![];

    for i in 0..5 {
        let handle = thread::spawn(move || {
            let result = i * i;
            println!("Thread {}: {} squared = {}", i, i, result);
            result
        });
        handles.push(handle);
    }

    let results: Vec<i32> = handles
        .into_iter()
        .map(|h| h.join().unwrap())
        .collect();

    println!("Results: {:?}", results);
}

Setiap thread menghitung kuadrat secara paralel. Hasil dikumpulkan setelah semua thread selesai.

Shared State dengan Mutex

Mutex (Mutual Exclusion) memungkinkan beberapa thread mengakses data yang sama secara bergantian. Hanya satu thread yang bisa mengakses data pada satu waktu.

Menggunakan Mutex

use std::sync::Mutex;

fn main() {
    let counter = Mutex::new(0);

    let mut handles = vec![];

    for _ in 0..10 {
        let handle = std::thread::spawn(move || {
            let mut num = counter.lock().unwrap();
            *num += 1;
        }); // lock di-release otomatis saat guard drop
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Result: {}", *counter.lock().unwrap()); // 10
}

Tapi kode di atas tidak compile! Kenapa? Karena counter di-move ke satu thread, thread lain tidak bisa mengaksesnya.

Arc (Atomic Reference Counting)

Untuk berbagi ownership antar thread, gunakan Arc:

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

fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();
            *num += 1;
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Result: {}", *counter.lock().unwrap()); // 10
}

Penjelasan:

• Arc::clone() tidak menyalin data, hanya menambah reference count
• Setiap thread punya Arc sendiri yang point ke data yang sama
• Mutex::lock() memastikan hanya satu thread yang bisa akses data
• Lock otomatis di-release saat guard keluar dari scope

Rc vs Arc

• Rc: Single-threaded reference counting. Lebih cepat tapi tidak thread-safe.
• Arc: Atomic reference counting. Thread-safe tapi sedikit lebih lambat.

Gunakan Rc untuk single-threaded, Arc untuk multi-threaded.

Deadlock

Hati-hati dengan deadlock:

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

// JANGAN lakukan ini!
// let a = Mutex::new(1);
// let b = Mutex::new(2);
// Thread 1: lock a, lalu lock b
// Thread 2: lock b, lalu lock a
// DEADLOCK!

Cegah deadlock dengan:

1. Selalu lock dalam urutan yang sama
2. Hindari nested lock
3. Gunakan try_lock() yang mengembalikan Result

Channel

Channel adalah cara aman untuk mengirim data antar thread. Mirip seperti pipa: satu thread mengirim, thread lain menerima.

Basic Channel

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

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("Hello dari thread!");
        tx.send(message).unwrap();
        // println!("{}", message); // ERROR! message sudah di-send
    });

    let received = rx.recv().unwrap();
    println!("Dapat: {}", received);
}

mpsc artinya Multiple Producer, Single Consumer. Banyak thread bisa mengirim, tapi hanya satu yang menerima.

Multiple Messages

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

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let messages = vec!["Halo", "dari", "thread", "lain"];
        for msg in messages {
            tx.send(String::from(msg)).unwrap();
            thread::sleep(Duration::from_millis(200));
        }
    });

    for received in rx {
        println!("Dapat: {}", received);
    }
}

rx bisa di-iterate seperti iterator. Loop berakhir saat semua sender di-drop.

Multiple Producers

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

fn main() {
    let (tx, rx) = mpsc::channel();

    for i in 0..3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let msg = format!("Pesan dari thread {}", i);
            tx.send(msg).unwrap();
        });
    }

    drop(tx); // Drop original sender

    for received in rx {
        println!("{}", received);
    }
}

Tiga thread mengirim pesan ke satu receiver. Channel otomatis ditutup saat semua sender di-drop.

Channel vs Mutex

Fitur	Channel	Mutex
Pola komunikasi	Message passing	Shared state
Thread safety	Otomatis	Manual lock/unlock
Deadlock risk	Rendah	Lebih tinggi
Cocok untuk	Kirim data antar thread	Akses data yang sama

Umumnya, channel lebih disarankan karena lebih mudah di-reasoning dan lebih aman.

Send dan Sync Traits

Rust menggunakan dua trait untuk menjamin keamanan concurrency:

• Send: Tipe yang aman dipindahkan ke thread lain
• Sync: Tipe yang aman diakses dari beberapa thread sekaligus

Hampir semua tipe di Rust adalah Send dan Sync. Pengecualian:

• Rc bukan Send karena reference counting-nya tidak atomic
• Cell dan RefCell bukan Sync karena tidak thread-safe
• Raw pointer bukan Send maupun Sync

Compiler secara otomatis mengecek trait ini. Jika kamu coba mengirim tipe yang bukan Send ke thread lain, compiler akan error:

use std::rc::Rc;
use std::thread;

// fn main() {
//     let rc = Rc::new(5);
//     thread::spawn(move || {
//         println!("{}", rc); // ERROR! Rc is not Send
//     });
// }

Inilah kekuatan Rust: compiler mencegah bug concurrency sebelum program berjalan.

Async/Await

Thread bagus untuk CPU-bound work (komputasi berat). Tapi untuk I/O-bound work (network, file, database), async/await lebih efisien karena tidak memakan resource thread.

Apa Itu Async?

Async memungkinkan program menunggu operasi I/O tanpa memblokir thread. Saat menunggu response dari server, thread bisa mengerjakan tugas lain.

Runtime Async

Rust tidak punya runtime async bawaan. Kamu perlu library seperti:

• tokio: Runtime async paling populer
• async-std: Mirip std library tapi async
• smol: Runtime async yang sangat kecil

Kita akan pakai tokio yang paling umum digunakan.

Setup Tokoi

Tambahkan di Cargo.toml:

[dependencies]
tokio = { version = "1", features = ["full"] }

Async Function

use tokio::time::{sleep, Duration};

async fn fetch_data(id: u32) -> String {
    println!("Fetching data {}...", id);
    sleep(Duration::from_millis(100)).await; // .await disini
    format!("Data untuk ID {}", id)
}

#[tokio::main]
async fn main() {
    let result = fetch_data(1).await;
    println!("{}", result);
}

async fn mengembalikan Future, bukan nilai langsung. .await menjalankan Future sampai selesai.

Concurrent dengan tokio::join!

use tokio::time::{sleep, Duration};

async fn task_a() -> String {
    sleep(Duration::from_millis(100)).await;
    String::from("Task A selesai")
}

async fn task_b() -> String {
    sleep(Duration::from_millis(100)).await;
    String::from("Task B selesai")
}

#[tokio::main]
async fn main() {
    // Jalankan keduanya secara concurrent
    let (a, b) = tokio::join!(task_a(), task_b());
    println!("{}", a);
    println!("{}", b);
    // Total waktu: ~100ms, bukan ~200ms
}

Dengan tokio::join!, kedua task berjalan secara concurrent. Waktu total adalah yang terlama, bukan jumlah keduanya.

Spawning Tasks

Untuk menjalankan banyak task secara concurrent:

use tokio::task;

async fn process(id: u32) -> u32 {
    // Simulasi kerja
    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
    id * 2
}

#[tokio::main]
async fn main() {
    let mut handles = vec![];

    for i in 0..5 {
        let handle = task::spawn(process(i));
        handles.push(handle);
    }

    let mut results = vec![];
    for handle in handles {
        results.push(handle.await.unwrap());
    }

    println!("Results: {:?}", results);
}

task::spawn() menjalankan Future di background. Mirip thread::spawn() tapi jauh lebih ringan.

Async vs Thread

Fitur	Async	Thread
Overhead	Sangat rendah (~bytes)	Cukup tinggi (~MB per thread)
Context switch	Cooperative	Preemptive
Cocok untuk	I/O-bound	CPU-bound
Jumlah concurrent task	Ratusan ribu	Ratusan
Kompleksitas	Butuh runtime	Langsung dari OS

Gunakan async untuk: HTTP request, database query, file I/O, WebSocket.

Gunakan thread untuk: komputasi berat, image processing, encoding video.

Contoh Praktis: Web Scraper Concurrent

Berikut contoh sederhana yang menggabungkan semua konsep:

use tokio::task;
use tokio::time::{sleep, Duration};

async fn fetch_url(url: &str) -> Result<String, String> {
    println!("Fetching: {}", url);
    sleep(Duration::from_millis(100)).await; // Simulasi HTTP request
    Ok(format!("Content dari {}", url))
}

#[tokio::main]
async fn main() {
    let urls = vec![
        "https://erlkim.web.id",
        "https://github.com/erlk1m",
        "https://docs.rust-lang.org",
    ];

    let mut handles = vec![];

    for url in urls {
        let url = url.to_string();
        let handle = task::spawn(async move {
            match fetch_url(&url).await {
                Ok(content) => println!("Berhasil: {}", content),
                Err(e) => println!("Gagal: {}", e),
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.await.unwrap();
    }

    println!("Semua selesai!");
}

Tiga URL di-fetch secara concurrent. Total waktu adalah yang terlama, bukan jumlah ketiganya.

Contoh: Producer-Consumer Pattern

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

async fn producer(tx: mpsc::Sender<String>, id: u32) {
    for i in 0..3 {
        let msg = format!("Producer {} - Message {}", id, i);
        tx.send(msg).await.unwrap();
        sleep(Duration::from_millis(50)).await;
    }
}

async fn consumer(mut rx: mpsc::Receiver<String>) {
    while let Some(msg) = rx.recv().await {
        println!("Consumed: {}", msg);
    }
    println!("Consumer selesai, semua producer sudah drop.");
}

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(32); // Buffer size 32

    let tx1 = tx.clone();
    let tx2 = tx.clone();
    drop(tx); // Drop original sender

    tokio::spawn(producer(tx1, 1));
    tokio::spawn(producer(tx2, 2));
    tokio::spawn(consumer(rx));

    sleep(Duration::from_secs(1)).await;
    println!("Program selesai.");
}

Pattern ini sangat umum di aplikasi production:

• Producer: Menghasilkan data (misal: membaca dari queue)
• Consumer: Memproses data (misal: menyimpan ke database)
• Channel: Menghubungkan keduanya dengan buffer

Best Practices Concurrency

1. Prefer Channel dari pada Mutex

Channel lebih mudah di-reasoning dan lebih aman:

// Lebih baik
let (tx, rx) = mpsc::channel();
tx.send(data).await;

// Kurang baik
let data = Arc::new(Mutex::new(vec![]));
data.lock().unwrap().push(item);

2. Hindari Nested Lock

// JANGAN
let a = lock_a.lock().unwrap();
let b = lock_b.lock().unwrap(); // Potensi deadlock

// LEBIH BAIK
let a = lock_a.lock().unwrap();
drop(a); // Release dulu
let b = lock_b.lock().unwrap();

3. Gunakan Arc untuk Shared Data

// Tidak bisa
// let data = Rc::new(Mutex::new(0)); // Rc is not Send

// Bisa
let data = Arc::new(Mutex::new(0)); // Arc is Send + Sync

4. Jangan Block di Async

// JANGAN: blocking call di async
async fn bad() {
    std::thread::sleep(Duration::from_secs(1)); // Block thread!
}

// BAIK: async sleep
async fn good() {
    tokio::time::sleep(Duration::from_secs(1)).await; // Non-blocking
}

5. Batasi Jumlah Concurrent Task

use tokio::sync::Semaphore;
use std::sync::Arc;

#[tokio::main]
async fn main() {
    let semaphore = Arc::new(Semaphore::new(10)); // Max 10 concurrent

    for i in 0..100 {
        let permit = semaphore.clone().acquire_owned().await.unwrap();
        tokio::spawn(async move {
            // Work here
            println!("Task {}", i);
            drop(permit); // Release slot
        });
    }
}

Semaphore membatasi jumlah task yang berjalan bersamaan. Penting untuk mencegah resource exhaustion.

Kesimpulan

Concurrency dan async programming di Rust sangat powerful berkat ownership system:

• Thread untuk CPU-bound work dengan keamanan compile time
• Mutex + Arc untuk shared state yang aman
• Channel untuk message passing yang lebih disarankan
• Async/Await untuk I/O-bound work yang sangat efisien
• Send + Sync traits menjamin keamanan di compile time

Rust membuktikan bahwa concurrency yang aman dan cepat bisa dicapai tanpa garbage collector. Di artikel selanjutnya, kita akan belajar tentang collections, iterator, dan closures di Rust.