Async Lebih Dalam di Rust: select!, Timeout, Stream, dan Pin

Di halaman Async/Await, kamu sudah belajar dasarnya: async fn mengembalikan future yang lazy (baru jalan kalau ditunggu), .await menjalankannya, #[tokio::main] menyalakan runtime, dan tokio::join! serta tokio::spawn menjalankan beberapa hal sekaligus. Kalau semua itu masih terasa asing, baca halaman itu dulu. Halaman ini melanjutkan dari sana.

Program async di dunia nyata butuh beberapa alat tambahan. Bayangkan lagi dapur restoran yang sibuk. Kadang kamu menunggu semua hidangan selesai (join!). Kadang kamu cuma peduli pada hal pertama yang terjadi: makanannya sudah siap, atau pelanggannya keburu pergi (select!). Kadang kamu menyerah setelah lima menit (timeout). Dan kadang pesanan terus berdatangan, satu per satu, sepanjang malam (stream). Halaman ini membahas semuanya, ditambah Pin, kata aneh yang cepat atau lambat akan kamu lihat di pesan error. Tenang, kita jelaskan dengan bahasa sederhana.

Semua contoh memakai tokio dengan features = ["full"], sama seperti di halaman dasar. Tombol “Coba sendiri” berfungsi, karena Rust Playground sudah punya tokio bawaan.

Menjalankan banyak hal sekaligus

Kamu sudah kenal tokio::join!: ia menunggu semua future-nya dan mengembalikan hasilnya sebagai tuple. Saudaranya, tokio::try_join!, dipakai untuk future yang mengembalikan Result. Hasilnya Ok berisi semua nilai, atau Err pertama yang muncul. Begitu ada satu yang gagal, ia berhenti menunggu yang lain:

use std::time::Duration;
use tokio::time::sleep;

async fn load_profile(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(20)).await;
    Ok(format!("profil #{id}"))
}

async fn load_settings(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(10)).await;
    if id == 2 {
        return Err(format!("tidak ada pengaturan untuk user {id}"));
    }
    Ok(String::from("mode gelap"))
}

#[tokio::main]
async fn main() {
    for id in [1, 2] {
        match tokio::try_join!(load_profile(id), load_settings(id)) {
            Ok((profile, settings)) => println!("user {id}: {profile}, {settings}"),
            Err(e) => println!("user {id} gagal: {e}"),
        }
    }
}

Output:

user 1: profil #1, mode gelap
user 2 gagal: tidak ada pengaturan untuk user 2

▶ Coba sendiri

join! dan try_join! menjalankan future-future itu di dalam task yang sedang berjalan, secara bergantian. Itu pas untuk beberapa future yang sudah kamu ketahui dari awal. Tapi kalau kamu punya daftar pekerjaan yang jumlahnya baru diketahui saat program berjalan, jalankan masing-masing sebagai task terpisah. Kamu bisa memasukkan setiap JoinHandle ke dalam Vec, seperti di halaman dasar. Tapi tokio punya alat yang lebih enak untuk ini: JoinSet, sebuah “kantong” berisi task yang memberimu setiap hasil begitu task itu selesai:

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::sleep;

async fn check_site(site: &'static str, ms: u64) -> (&'static str, u64) {
    sleep(Duration::from_millis(ms)).await; // anggap saja ini panggilan jaringan
    (site, ms)
}

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    for (site, ms) in [("rust-lang.org", 30), ("docs.rs", 10), ("crates.io", 20)] {
        set.spawn(check_site(site, ms));
    }

    let mut results = Vec::new();
    while let Some(res) = set.join_next().await {
        results.push(res.unwrap()); // Err hanya kalau task-nya panic
    }

    results.sort(); // urutan selesai bisa berubah setiap kali dijalankan
    for (site, ms) in results {
        println!("{site} menjawab dalam {ms}ms");
    }
}

Output:

crates.io menjawab dalam 20ms
docs.rs menjawab dalam 10ms
rust-lang.org menjawab dalam 30ms

▶ Coba sendiri

join_next() mengembalikan None kalau kantongnya sudah kosong, jadi perulangan while let berhenti dengan sendirinya. Hasilnya datang sesuai urutan selesai, bukan urutan saat di-spawn. Kalau waktu kerja task-task itu mirip, urutannya bisa berubah dari satu run ke run berikutnya, jadi kita urutkan dulu sebelum dicetak. Satu hal lagi yang menyenangkan: kalau kamu men-drop sebuah JoinSet, semua task yang masih ada di dalamnya ikut dibatalkan. Tidak ada yang tertinggal berjalan tanpa sengaja.

Balapan future dengan select!

join! menunggu semuanya. tokio::select! menunggu yang pertama. Setiap cabang punya pattern, future, dan blok kode: pattern = future => kode. Future mana pun yang selesai duluan, kodenya yang dijalankan, dan cabang-cabang lainnya dibatalkan.

Apa arti “dibatalkan” di Rust? Future yang kalah cukup di-drop, sama seperti nilai lain yang keluar dari cakupan (scope). Ia tidak pernah berjalan lagi. Variabel lokalnya juga ikut di-drop, jadi kode pembersihnya tetap dijalankan. Contoh ini memakai tipe Drop kecil supaya hal itu kelihatan:

use std::time::Duration;
use tokio::time::sleep;

struct Download(&'static str);

impl Drop for Download {
    fn drop(&mut self) {
        println!("membersihkan unduhan {}", self.0);
    }
}

async fn download(mirror: &'static str, ms: u64) -> &'static str {
    let _download = Download(mirror);
    sleep(Duration::from_millis(ms)).await;
    println!("unduhan {mirror} selesai");
    mirror
}

#[tokio::main]
async fn main() {
    tokio::select! {
        winner = download("Eropa", 50) => println!("pemenang: {winner}"),
        winner = download("Asia", 1000) => println!("pemenang: {winner}"),
    }
    println!("main jalan terus");
}

Output:

unduhan Eropa selesai
membersihkan unduhan Eropa
membersihkan unduhan Asia
pemenang: Eropa
main jalan terus

▶ Coba sendiri

Perhatikan apa yang tidak muncul: “unduhan Asia selesai” tidak pernah dicetak. Future Asia dihentikan di .await-nya lalu di-drop, dan nilai Download-nya langsung dibersihkan. Begitulah cara pembatalan bekerja di seluruh async Rust: berhenti mem-poll sebuah future, lalu drop. Tidak perlu panggilan “cancel” khusus.

Pemakaian klasik select! adalah timeout: adu pekerjaanmu dengan sebuah timer.

tokio::select! {
    report = build_report() => println!("{report}"),
    _ = tokio::time::sleep(Duration::from_secs(5)) => println!("terlalu lama, menyerah"),
}

Cara ini berfungsi dengan baik, tapi ada jalan pintasnya, yaitu tokio::time::timeout, yang dibahas di bagian berikutnya.

Bekerja sampai ada sinyal berhenti

select! benar-benar bersinar di dalam loop. Di sini sebuah worker mengambil pekerjaan dari sebuah channel, tapi juga mendengarkan pesan “tolong berhenti” di channel kedua. Channel oneshot membawa tepat satu pesan, jadi sangat cocok untuk sinyal berhenti (shutdown):

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

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<String>(10);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    let worker = tokio::spawn(async move {
        let mut done = 0;
        loop {
            tokio::select! {
                Some(job) = job_rx.recv() => {
                    println!("mengerjakan: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("sinyal berhenti diterima");
                    break;
                }
            }
        }
        done
    });

    for job in ["ubah ukuran foto", "kirim email", "perbarui indeks pencarian"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    sleep(Duration::from_millis(100)).await; // beri worker waktu untuk menyelesaikan
    stop_tx.send(()).unwrap();

    let done = worker.await.unwrap();
    println!("worker menyelesaikan {done} pekerjaan");
}

Output:

mengerjakan: ubah ukuran foto
mengerjakan: kirim email
mengerjakan: perbarui indeks pencarian
sinyal berhenti diterima
worker menyelesaikan 3 pekerjaan

▶ Coba sendiri

Beberapa detail yang perlu kamu tahu:

• Kita menulis &mut stop_rx, bukan stop_rx. Setiap putaran loop, select! butuh receiver yang sama lagi, jadi kita meminjamkannya, bukan memindahkannya (move) ke putaran pertama.
• Setiap putaran loop, job_rx.recv() membuat future baru, dan future yang kalah di-drop. Itu tidak masalah, karena recv() itu cancel-safe (aman dibatalkan): men-drop-nya tidak pernah membuat pesan hilang.
• Kalau pattern-nya tidak cocok (di sini, recv() mengembalikan None karena semua pengirim sudah hilang), select! mematikan cabang itu dan tetap menunggu cabang lainnya.
• Kalau beberapa cabang siap di saat yang sama, select! memilih salah satunya secara acak, supaya tidak ada cabang yang memonopoli loop. Tulis biased; sebagai baris pertama di dalam select! kalau kamu ingin cabang-cabangnya diperiksa dari atas ke bawah.

Cancellation safety (keamanan pembatalan), dalam satu paragraf. Karena cabang yang kalah di-drop, pekerjaan yang baru setengah jalan di dalamnya ikut dibuang. Untuk recv() pada channel atau sleep(), itu tidak berbahaya: belum ada yang diambil. Tapi bayangkan sebuah future yang membaca 100 byte dari socket jaringan, dan sudah membaca 60 byte saat ia di-drop. Ke-60 byte itu hilang. Future seperti itu tidak cancel-safe. Dokumentasi tokio menyebutkan untuk setiap method apakah ia cancel-safe atau tidak. Saat kamu memakai select! di dalam loop, periksa method yang kamu panggil, atau buat future-nya sekali saja di luar loop dan terus poll future yang sama itu (caranya kamu lihat di bagian Pin).

Timeout dan timer

tokio::time::timeout(duration, future) membungkus future apa saja. Kalau future-nya selesai tepat waktu, kamu dapat Ok(value). Kalau waktunya habis duluan, future-nya di-drop (dibatalkan, persis seperti cabang select! yang kalah) dan kamu dapat Err(Elapsed):

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

async fn fetch_price(server_delay_ms: u64) -> u32 {
    sleep(Duration::from_millis(server_delay_ms)).await;
    42
}

#[tokio::main]
async fn main() {
    let limit = Duration::from_millis(200);

    for (server, delay) in [("server cepat", 20), ("server lambat", 2000)] {
        match timeout(limit, fetch_price(delay)).await {
            Ok(price) => println!("{server}: harganya {price}"),
            Err(e) => println!("{server}: {e}"),
        }
    }
}

Output:

server cepat: harganya 42
server lambat: deadline has elapsed

▶ Coba sendiri

“deadline has elapsed” adalah pesan bawaan tokio untuk error Elapsed, artinya “batas waktunya sudah lewat”. Pasang timeout pada apa pun yang berkomunikasi dengan dunia luar: panggilan HTTP, query database, membaca dari socket. Server yang tidak pernah menjawab seharusnya tidak membuat programmu macet selamanya.

sleep vs interval. sleep menunggu satu kali. interval memberimu irama yang teratur, seperti metronom. Kamu memanggil .tick().await di dalam loop, dan setiap tick datang sesuai jadwal. Tick pertama langsung terjadi:

use std::time::Duration;
use tokio::time::interval;

#[tokio::main]
async fn main() {
    let mut heartbeat = interval(Duration::from_millis(50));

    for beat in 1..=3 {
        heartbeat.tick().await; // tick pertama langsung selesai
        println!("detak {beat}");
    }
}

Output:

detak 1
detak 2
detak 3

▶ Coba sendiri

Kenapa tidak sleep 50ms saja di dalam loop? Karena dengan begitu setiap putaran memakan waktu “pekerjaanmu + 50ms”, dan jadwalnya pelan-pelan bergeser. interval mengukur dari waktu mulai yang direncanakan untuk setiap tick, jadi kalau pekerjaanmu butuh 10ms, ia hanya menunggu sisa 40ms. Pakai interval untuk heartbeat, polling, dan pekerjaan “simpan setiap menit”. Pakai sleep untuk jeda sekali saja.

Stream: iterator versi async

Sebuah iterator memberimu nilai satu per satu lewat next(). Stream adalah versi async-nya: next().await memberimu nilai berikutnya saat nilainya sudah siap, dan None kalau sudah tidak ada lagi. Bayangkan papan harga saham yang terus berganti, baris-baris data yang datang dari database, atau pesan dari server chat.

Sebenarnya kamu sudah pernah memakai sesuatu yang mirip stream: receiver dari channel tokio. rx.recv().await menunggu pesan berikutnya dan mengembalikan None kalau semua pengirimnya sudah hilang:

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);

    tokio::spawn(async move {
        for temp in [21, 22, 24, 23] {
            tx.send(temp).await.unwrap();
            sleep(Duration::from_millis(10)).await; // satu pembacaan setiap 10ms
        }
        // tx di-drop di sini, dan itu menutup channel-nya
    });

    while let Some(temp) = rx.recv().await {
        println!("suhu: {temp}°C");
    }
    println!("sensor berhenti");
}

Output:

suhu: 21°C
suhu: 22°C
suhu: 24°C
suhu: 23°C
sensor berhenti

▶ Coba sendiri

Standard library belum punya trait Stream (desainnya masih dikerjakan). Ekosistem Rust memakai trait dari crate futures, dan crate pembantu tokio, tokio-stream, menambahkan trait StreamExt dengan method yang sudah kamu kenal dari iterator: next(), map(), filter(), take(), dan lainnya. Tambahkan ke proyekmu:

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

use tokio_stream::{self as stream, StreamExt};

#[tokio::main]
async fn main() {
    let prices = stream::iter(vec![100, 102, 98, 105, 110]);

    let mut doubled = prices.map(|p| p * 2).take(3);

    while let Some(p) = doubled.next().await {
        println!("{p}");
    }
}

Program ini mencetak 200, 204, dan 196. Kodenya terbaca persis seperti kode iterator, kecuali .await setelah next(). Untuk mengubah receiver channel tokio menjadi stream sungguhan, bungkus saja: tokio_stream::wrappers::ReceiverStream::new(rx). (Crate futures punya StreamExt sendiri dengan ide yang sama. Pilih salah satu saja per file, karena kedua trait itu punya method next() dan keduanya akan bentrok.)

Pin, dengan bahasa sederhana

Cepat atau lambat kamu akan melihat Pin di sebuah tipe atau pesan error. Berikut idenya, tanpa teori yang terlalu dalam.

Saat Rust mengompilasi sebuah async fn, ia mengubahnya menjadi struct tersembunyi, sebuah state machine (mesin keadaan). Setiap variabel lokal yang masih dibutuhkan setelah .await menjadi field dari struct itu. Sekarang bayangkan ada satu variabel lokal yang meminjam variabel lain, seperti let name = String::from("Ana"); let r = &name; dengan .await di antaranya. Artinya struct itu menyimpan pointer ke field-nya sendiri. Kalau struct itu dipindahkan ke tempat lain di memori, pointer tadi akan menunjuk ke tempat lama yang sudah kosong.

Pin adalah solusinya. Nilai yang di-pin adalah sebuah janji: “nilai ini tidak akan pindah lagi di memori.” Begitu sebuah future di-pin, ia aman untuk di-poll. Biasanya kamu bahkan tidak sadar, karena .await sudah mem-pin future itu untukmu.

Kamu akan bertemu Pin dalam tiga situasi:

1. Mem-poll future yang sama berkali-kali, misalnya sebuah deadline di dalam loop select!.
2. Menyimpan future di dalam Vec atau field struct, biasanya sebagai Pin<Box<dyn Future<Output = T>>>.
3. Rekursi di dalam async fn.

Untuk situasi pertama, inilah yang terjadi kalau kamu mencoba memakai ulang sebuah sleep di dalam loop tanpa mem-pin-nya:

let deadline = sleep(Duration::from_millis(300));

loop {
    tokio::select! {
        _ = &mut deadline => break,
        Some(n) = rx.recv() => println!("{n}"),
    }
}

error[E0277]: `PhantomPinned` cannot be unpinned
   |
   = note: consider using the `pin!` macro
           consider using `Box::pin` if you need to access the pinned value outside of the current scope
   = note: required for `Sleep` to implement `Unpin`

Compiler bilang Sleep tidak bisa “di-unpin”, dan ia bahkan langsung menyarankan solusinya: pakai makro pin!, atau Box::pin. std::pin::pin! (atau tokio::pin!) mem-pin future itu di tempatnya, di stack. Sekarang deadline yang sama dipakai bersama oleh setiap putaran loop:

use std::pin::pin;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);
    tokio::spawn(async move {
        for order in 1..=3 {
            tx.send(order).await.unwrap();
            sleep(Duration::from_millis(10)).await;
        }
        sleep(Duration::from_secs(5)).await; // lalu tokonya jadi sepi
    });

    let mut deadline = pin!(sleep(Duration::from_millis(300)));

    loop {
        tokio::select! {
            _ = &mut deadline => {
                println!("waktunya tutup");
                break;
            }
            Some(order) = rx.recv() => println!("pesanan masuk #{order}"),
        }
    }
}

Output:

pesanan masuk #1
pesanan masuk #2
pesanan masuk #3
waktunya tutup

▶ Coba sendiri

Kalau kita menulis sleep(...) di dalam select!, setiap pesanan baru akan memulai timer 300ms yang baru, dan toko yang ramai tidak akan pernah tutup. Mem-pin satu deadline di luar loop menyelesaikan masalah itu.

Untuk rekursi, compiler perlu tahu seberapa besar ukuran future-nya, dan future yang berisi dirinya sendiri akan berukuran tak terhingga. Memanggil countdown(n - 1).await di dalam async fn countdown menghasilkan:

error[E0733]: recursion in an async fn requires boxing
  |
  = note: a recursive `async fn` call must introduce indirection such as `Box::pin` to avoid an infinitely sized future

Artinya: rekursi di async fn butuh boxing, misalnya dengan Box::pin, supaya ukuran future-nya tidak tak terhingga. Box::pin memindahkan future bagian dalam ke heap dan mem-pin-nya di sana. Future bagian luar sekarang hanya menyimpan sebuah pointer, yang ukurannya tetap:

async fn countdown(n: u32) {
    if n == 0 {
        println!("meluncur!");
        return;
    }
    println!("{n}...");
    Box::pin(countdown(n - 1)).await;
}

#[tokio::main]
async fn main() {
    countdown(3).await;
}

Output:

3...
2...
1...
meluncur!

▶ Coba sendiri

Jujur saja, hanya itu yang dibutuhkan kebanyakan orang: pin! untuk future yang kamu poll berkali-kali, dan Box::pin untuk menyimpan future atau untuk rekursi. Kalau kamu ingin cerita lengkapnya, bab Rust Book di daftar Bacaan lebih lanjut menjelaskannya langkah demi langkah.

async fn di trait dan async closure

Sejak Rust 1.75, kamu bisa menulis async fn langsung di dalam sebuah trait. Sejak Rust 1.85, kamu juga punya async closure, yang ditulis async |x| ..., lengkap dengan trait pasangannya AsyncFn, AsyncFnMut, dan AsyncFnOnce (versi async dari trait closure). Berikut keduanya dalam satu contoh:

use std::time::Duration;
use tokio::time::sleep;

trait Notifier {
    async fn notify(&self, user: &str) -> String;
}

struct Email;

impl Notifier for Email {
    async fn notify(&self, user: &str) -> String {
        sleep(Duration::from_millis(10)).await; // anggap saja sedang menghubungi server email
        format!("email terkirim ke {user}")
    }
}

async fn for_each_user(users: &[&str], job: impl AsyncFn(&str) -> String) {
    for user in users {
        println!("{}", job(user).await);
    }
}

#[tokio::main]
async fn main() {
    let email = Email;
    let users = ["ana", "ben"];

    for_each_user(&users, async |user: &str| email.notify(user).await).await;
}

Output:

email terkirim ke ana
email terkirim ke ben

▶ Coba sendiri

Ada satu jebakan pada async fn di trait. Trait itu tidak menjanjikan bahwa future yang dikembalikan bersifat Send, padahal tokio::spawn butuh future yang Send (karena future itu bisa berpindah antar-thread). Jadi men-spawn pemanggilan lewat generik N: Notifier akan gagal:

fn send_in_background<N: Notifier + Send + Sync + 'static>(notifier: N) {
    tokio::spawn(async move {
        let msg = notifier.notify("ana").await;
        println!("{msg}");
    });
}

error: future cannot be sent between threads safely
  |
7 |         let msg = notifier.notify("ana").await;
  |                   ^^^^^^^^^^^^^^^^^^^^^^ await occurs here on type `impl Future<Output = String>`, which is not `Send`
  |
help: `Send` can be made part of the associated future's guarantees for all implementations of `Notifier::notify`
  |
2 -     async fn notify(&self, user: &str) -> String;
2 +     fn notify(&self, user: &str) -> impl std::future::Future<Output = String> + Send;

Error ini bilang future-nya tidak bisa dikirim antar-thread dengan aman, dan bagian help menyarankan agar Send dijadikan bagian dari jaminan method itu. Saran compiler tersebut memang solusi yang biasa dipakai: tulis method trait-nya sebagai fn ... -> impl Future<Output = String> + Send. Yang mengimplementasikan trait itu tetap boleh menulis async fn biasa. Perlu dicatat juga, trait yang punya async fn belum bisa dipakai sebagai trait object dyn Notifier. Kalau kamu butuh itu, crate populer async-trait akan memasukkan future-future itu ke dalam box untukmu.

Kesalahan umum

Memblokir runtime. std::thread::sleep, perhitungan berat, atau pembacaan std::fs yang besar di dalam task async akan membekukan seluruh worker thread, dan semua task lain di thread itu ikut menunggu. Halaman dasar sudah menunjukkan versi sleep-nya. Untuk pekerjaan yang berat di CPU atau yang memblokir, serahkan ke tokio::task::spawn_blocking, yang menjalankannya di kumpulan thread terpisah yang memang dibuat untuk itu:

fn count_primes(limit: u64) -> usize {
    (2..limit)
        .filter(|&n| (2..).take_while(|d| d * d <= n).all(|d| n % d != 0))
        .count()
}

#[tokio::main]
async fn main() {
    let handle = tokio::task::spawn_blocking(|| count_primes(50_000));

    println!("tetap melayani request selagi kita menghitung...");

    let primes = handle.await.unwrap();
    println!("ada {primes} bilangan prima di bawah 50000");
}

Output:

tetap melayani request selagi kita menghitung...
ada 5133 bilangan prima di bawah 50000

▶ Coba sendiri

Memegang guard std::sync::Mutex melewati .await. Kode ini kelihatan tidak berbahaya:

let handle = tokio::spawn(async move {
    let mut count = visits.lock().unwrap();
    *count += 1;
    save_to_disk(*count).await;
});

error: future cannot be sent between threads safely
   |
   = help: within `{async block@src/main.rs:13:31: 13:41}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, u32>`
note: future is not `Send` as this value is used across an await
   |
14 |         let mut count = visits.lock().unwrap();
   |             --------- has type `std::sync::MutexGuard<'_, u32>` which is not `Send`
15 |         *count += 1;
16 |         save_to_disk(*count).await;
   |                              ^^^^^ await occurs here, with `mut count` maybe used later

Compiler sedang melindungimu. Pesan itu bilang MutexGuard bukan Send, dan guard itu masih hidup saat .await. Selama task ini tertidur di .await, ia masih memegang lock-nya, dan task lain yang ingin lock itu akan memblokir seluruh thread-nya. Solusinya adalah men-drop guard sebelum .await, dengan menaruh bagian penguncian di bloknya sendiri:

let count = {
    let mut count = visits.lock().unwrap();
    *count += 1;
    *count
}; // guard di-drop di sini
save_to_disk(count).await;

Kalau kamu benar-benar perlu memegang lock sambil menunggu, pakai tokio::sync::Mutex, yang lock().await-nya menunggu tanpa memblokir thread. (Untuk penjelasan lebih lanjut tentang lock, lihat Shared State.)

Lupa .await. Future itu lazy, jadi save_file(); tanpa .await sama sekali tidak melakukan apa-apa. Compiler memperingatkanmu dengan “futures do nothing unless you .await or poll them” (future tidak melakukan apa-apa kecuali kamu .await atau poll). Lihat Async/Await untuk cerita lengkapnya, dan jangan abaikan peringatan itu.

Men-spawn task lalu tidak pernah menunggunya. Saat main selesai, runtime ikut mati dan setiap task yang belum selesai di-drop:

use std::time::Duration;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        println!("laporan tersimpan"); // tidak pernah dicetak!
    });

    println!("main sudah selesai");
}

Output:

main sudah selesai

▶ Coba sendiri

Laporannya tidak pernah tersimpan. Kalau pekerjaan itu penting, simpan JoinHandle-nya dan .await sebelum main berakhir (atau masukkan task-task itu ke JoinSet lalu tunggu sampai kosong).

Latihan

1. Cek harga dengan batas waktu

Empat toko menjawab permintaan harga setelah 20, 40, 300, dan 60 milidetik. Tanyakan ke keempatnya sekaligus dengan JoinSet, tapi beri setiap toko waktu hanya 150ms. Cetak harga yang berhasil datang (diurutkan menurut nama toko), lalu berapa toko yang kehabisan waktu. Petunjuk: spawn timeout(limit, ask_shop(name, delay)) ke dalam set. Setiap task yang selesai lalu memberimu Ok atau Err(Elapsed), jadi kembalikan nama toko bersama hasil itu.

Lihat solusi

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{sleep, timeout};

async fn ask_shop(delay_ms: u64) -> u32 {
    sleep(Duration::from_millis(delay_ms)).await;
    100 + delay_ms as u32 // harga karangan
}

#[tokio::main]
async fn main() {
    let shops = [("Corner Shop", 20), ("MegaMart", 40), ("SlowCo", 300), ("Bits & Bobs", 60)];
    let limit = Duration::from_millis(150);

    let mut set = JoinSet::new();
    for (name, delay) in shops {
        set.spawn(async move { (name, timeout(limit, ask_shop(delay)).await) });
    }

    let mut prices = Vec::new();
    let mut timed_out = 0;
    while let Some(res) = set.join_next().await {
        match res.unwrap() {
            (name, Ok(price)) => prices.push((name, price)),
            (_, Err(_)) => timed_out += 1,
        }
    }

    prices.sort();
    for (name, price) in prices {
        println!("{name}: {price}");
    }
    println!("{timed_out} toko kehabisan waktu");
}

Output:

Bits & Bobs: 160
Corner Shop: 120
MegaMart: 140
1 toko kehabisan waktu

▶ Coba sendiri

Semua toko menunggu di saat yang sama, jadi seluruh pengecekan hanya butuh sekitar 150ms, selambat apa pun SlowCo. Saat timeout-nya habis, future SlowCo di-drop, jadi ia tidak bisa terus berjalan diam-diam di belakang.

2. Ukuran folder dengan rekursi async

Sebuah Folder punya nama, daftar ukuran file dalam KB, dan daftar sub-folder. Tulis async fn total_size(folder: &Folder) -> u64 yang menjumlahkan file-file miliknya sendiri ditambah ukuran setiap sub-folder, dengan memanggil dirinya sendiri untuk masing-masing sub-folder. Anggap setiap folder butuh “pembacaan disk” async yang singkat lebih dulu, dengan tokio::task::yield_now().await. Petunjuk: pemanggilan rekursifnya perlu dibungkus Box::pin(...), kalau tidak kamu akan mendapat error E0733.

Lihat solusi

struct Folder {
    name: &'static str,
    files_kb: Vec<u64>,
    subfolders: Vec<Folder>,
}

async fn total_size(folder: &Folder) -> u64 {
    tokio::task::yield_now().await; // anggap saja sedang membaca disk

    let mut total: u64 = folder.files_kb.iter().sum();
    for sub in &folder.subfolders {
        total += Box::pin(total_size(sub)).await;
    }
    println!("{}: {} KB", folder.name, total);
    total
}

#[tokio::main]
async fn main() {
    let home = Folder {
        name: "home",
        files_kb: vec![4],
        subfolders: vec![
            Folder { name: "foto", files_kb: vec![2048, 1024], subfolders: vec![] },
            Folder {
                name: "kode",
                files_kb: vec![12, 8],
                subfolders: vec![Folder { name: "target", files_kb: vec![500], subfolders: vec![] }],
            },
        ],
    };

    let total = total_size(&home).await;
    println!("total: {total} KB");
}

Output:

foto: 3072 KB
target: 500 KB
kode: 520 KB
home: 3596 KB
total: 3596 KB

▶ Coba sendiri

Setiap folder baru dicetak setelah semua sub-folder-nya, karena ia harus menunggu ukuran sub-folder itu lebih dulu. Box::pin menaruh future dari setiap pemanggilan bersarang di heap, jadi ukuran future bagian luar tetap.

3. Habiskan antrean, lalu berhenti

Ubah contoh sinyal berhenti tadi supaya tidak butuh sleep(100ms) sebelum mengirim sinyal berhenti. Kirim ketiga pekerjaan dan sinyal berhentinya langsung, tapi pastikan worker tetap menyelesaikan semua pekerjaan di antrean sebelum berhenti. Petunjuk: tanpa bantuan, select! memilih cabang secara acak kalau keduanya siap. Tambahkan biased; dan taruh cabang pekerjaan di urutan pertama.

Lihat solusi

use tokio::sync::{mpsc, oneshot};

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<String>(10);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    for job in ["ubah ukuran foto", "kirim email", "perbarui indeks pencarian"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    stop_tx.send(()).unwrap();

    let worker = tokio::spawn(async move {
        let mut done = 0;
        loop {
            tokio::select! {
                biased;
                Some(job) = job_rx.recv() => {
                    println!("mengerjakan: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("antrean kosong, berhenti");
                    break;
                }
            }
        }
        done
    });

    let done = worker.await.unwrap();
    println!("worker menyelesaikan {done} pekerjaan");
}

Output:

mengerjakan: ubah ukuran foto
mengerjakan: kirim email
mengerjakan: perbarui indeks pencarian
antrean kosong, berhenti
worker menyelesaikan 3 pekerjaan

▶ Coba sendiri

Dengan biased;, select! selalu mencoba cabang pekerjaan lebih dulu, jadi ia baru melihat sinyal berhenti saat recv() sudah tidak punya apa-apa yang siap. Tanpa itu, worker bisa saja berhenti setelah satu atau dua pekerjaan, dan bug seperti ini cuma muncul sesekali, jadi sulit dilacak.

Ringkasan

• join! menunggu semua future, try_join! berhenti di Err pertama, dan JoinSet menjalankan banyak task yang di-spawn lalu mengembalikan hasilnya begitu masing-masing selesai (urutkan kalau urutan itu penting).
• select! menunggu future pertama yang selesai. Yang lain di-drop, dan begitulah cara pembatalan bekerja di async Rust. Di dalam loop, pakai future yang cancel-safe, dan biased; kalau urutan itu penting.
• timeout menyerah pada future yang terlalu lambat. interval berdetak dengan jadwal yang teratur, sedangkan sleep menunggu satu kali.
• Stream adalah iterator versi async. Receiver channel bekerja seperti stream, dan tokio-stream menambahkan map, take, dan kawan-kawannya.
• Pin menjanjikan bahwa sebuah future tidak akan pindah di memori. Pakai pin! untuk mem-poll satu future berkali-kali, dan Box::pin untuk future yang disimpan dan untuk rekursi async.
• Jangan memblokir runtime (pakai spawn_blocking), jangan memegang lock std melewati .await, dan tunggu task yang kamu spawn.

Kuis singkat

Kuis singkat

1. Di tokio::select!, apa yang terjadi pada cabang-cabang yang tidak selesai duluan?

Mereka terus berjalan di belakang layar Mereka di-drop, jadi berhenti dan dibersihkan Mereka dijalankan lagi di putaran loop berikutnya

2. Apa yang dikembalikan tokio::time::timeout saat waktunya habis?

Ia panic Ok dengan nilai default Err(Elapsed), dan future di dalamnya di-drop

3. Kenapa kamu butuh Box::pin saat sebuah async fn memanggil dirinya sendiri?

Future yang berisi dirinya sendiri akan berukuran tak terhingga Rekursi sama sekali tidak diizinkan di kode async Supaya fungsinya berjalan di thread lain

4. Kamu harus menjalankan pekerjaan berat di CPU selama 2 detik di dalam program tokio. Alat apa yang paling tepat?

std::thread::sleep di sekitarnya tokio::select! JoinSet dengan tokio::spawn tokio::task::spawn_blocking

Skor: 0 / 4

Bacaan lebih lanjut

• Select — the Tokio tutorial (bahasa Inggris): select! secara detail: cara kerja cabang-cabangnya, arti pembatalan, dan cara memakainya di dalam loop.
• Streams — the Tokio tutorial (bahasa Inggris): cara memakai, mengadaptasi, dan membuat stream dengan tokio-stream.
• A Closer Look at the Traits for Async — The Rust Book (bahasa Inggris): pembahasan yang ramah dan bertahap tentang Future, Pin, Unpin, dan Stream.
• std::pin — Rust standard library (bahasa Inggris): penjelasan resmi tentang pinning, Pin, Unpin, dan makro pin!.
• tokio::time di docs.rs (bahasa Inggris): sleep, timeout, interval, dan cara menangani tick yang terlewat.