Bahasa Pemrograman Rust

Menyerahkan Kontrol (Yielding) ke Runtime

Ingat kembali dari bagian “Program Asinkron Pertama Kita” kalau di tiap titik await, Rust ngasih kesempatan ke sebuah runtime buat me-pause task dan beralih ke task lain kalau future yang lagi di-await ternyata belum siap. Kebalikannya juga benar: Rust hanya me-pause blok asinkron dan menyerahkan kontrol kembali ke runtime pada saat titik await. Semua hal di antara titik-titik await itu sifatnya sinkron.

Itu artinya kalau kita melakukan banyak pekerjaan di dalam sebuah blok asinkron tanpa adanya titik await, future tersebut bakal memblokir futures lainnya supaya tidak bisa bikin progress. Kita mungkin kadang-kadang mendengar hal ini disebut sebagai satu future yang me-starving (membuat lapar/menghambat) futures lainnya. Di beberapa kasus, itu mungkin bukan masalah besar. Tapi, kalau kita lagi melakukan semacam setup yang berat atau pekerjaan yang makan waktu lama, atau kalau kita punya sebuah future yang bakal terus-menerus melakukan suatu tugas tertentu selamanya, kita perlu memikirkan kapan dan di mana harus menyerahkan kontrol kembali ke runtime.

Mari kita simulasikan sebuah operasi yang memakan waktu lama buat mengilustrasikan masalah starvation ini, lalu mengeksplorasi gimana cara menyelesaikannya. Listing 17-14 memperkenalkan sebuah fungsi slow.

extern crate trpl; // required for mdbook test

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

fn main() {
    trpl::block_on(async {
        // We will call `slow` here later
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

Kode ini memakai std::thread::sleep bukannya trpl::sleep supaya pemanggilan slow bakal memblokir thread saat ini selama beberapa milidetik. Kita bisa memakai slow sebagai pengganti buat operasi di dunia nyata yang sifatnya makan waktu lama sekaligus memblokir.

Di Listing 17-15, kita memakai slow buat meniru pengerjaan tugas semacam CPU-bound ini di dalam sepasang futures.

extern crate trpl; // required for mdbook test

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

fn main() {
    trpl::block_on(async {
        let a = async {
            println!("'a' started.");
            slow("a", 30);
            slow("a", 10);
            slow("a", 20);
            trpl::sleep(Duration::from_millis(50)).await;
            println!("'a' finished.");
        };

        let b = async {
            println!("'b' started.");
            slow("b", 75);
            slow("b", 10);
            slow("b", 15);
            slow("b", 350);
            trpl::sleep(Duration::from_millis(50)).await;
            println!("'b' finished.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

Tiap future menyerahkan kontrol kembali ke runtime cuma setelah melaksanakan serangkaian operasi lambat. Kalau kita menjalankan kode ini, kita bakal melihat output ini:

'a' started.
'a' ran for 30ms
'a' ran for 10ms
'a' ran for 20ms
'b' started.
'b' ran for 75ms
'b' ran for 10ms
'b' ran for 15ms
'b' ran for 350ms
'a' finished.

Sama kayak di Listing 17-5 di mana kita memakai trpl::select buat mengadu futures yang mengambil dua URL, select tetap selesai begitu a beres. Tapi nggak ada proses selang-seling (interleaving) di antara pemanggilan ke slow di kedua futures tersebut. Future a melakukan semua pekerjaannya sampai pemanggilan trpl::sleep di-await, baru setelah itu future b melakukan semua pekerjaannya sampai pemanggilan trpl::sleep-nya sendiri di-await, dan akhirnya future a selesai. Buat membolehkan kedua futures membikin progress di sela-sela tugas lambat mereka, kita butuh titik await supaya kita bisa menyerahkan kontrol kembali ke runtime. Itu artinya kita butuh sesuatu yang bisa kita await!

Kita sudah bisa melihat penyerahan kontrol semacam ini terjadi di Listing 17-15: kalau seandainya kita menghapus trpl::sleep di akhir future a, dia bakal selesai tanpa future b sempat berjalan sama sekali. Mari kita coba memakai fungsi trpl::sleep sebagai titik awal buat membiarkan operasi-operasinya bergantian membikin progress, kayak yang ditunjukkan di Listing 17-16.

extern crate trpl; // required for mdbook test

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

fn main() {
    trpl::block_on(async {
        let one_ms = Duration::from_millis(1);

        let a = async {
            println!("'a' started.");
            slow("a", 30);
            trpl::sleep(one_ms).await;
            slow("a", 10);
            trpl::sleep(one_ms).await;
            slow("a", 20);
            trpl::sleep(one_ms).await;
            println!("'a' finished.");
        };

        let b = async {
            println!("'b' started.");
            slow("b", 75);
            trpl::sleep(one_ms).await;
            slow("b", 10);
            trpl::sleep(one_ms).await;
            slow("b", 15);
            trpl::sleep(one_ms).await;
            slow("b", 350);
            trpl::sleep(one_ms).await;
            println!("'b' finished.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

Kita sudah nambahin pemanggilan trpl::sleep dengan titik-titik await di tiap sela pemanggilan ke slow. Sekarang pekerjaan kedua futures tersebut sudah diselang-seling:

'a' started.
'a' ran for 30ms
'b' started.
'b' ran for 75ms
'a' ran for 10ms
'b' ran for 10ms
'a' ran for 20ms
'b' ran for 15ms
'a' finished.

Future a tetap jalan sebentar sebelum menyerahkan kontrol ke b, karena dia memanggil slow sebelum pernah memanggil trpl::sleep, tapi setelah itu futures-nya bertukar peran bolak-balik setiap kali salah satu dari mereka mencapai titik await. Di kasus ini, kita sudah melakukan itu setelah tiap pemanggilan ke slow, tapi kita bisa membagi-bagi pekerjaannya pakai cara apa pun yang paling masuk akal buat kita.

Tapi sebenarnya kita nggak mau bener-bener “tidur” (sleep) di sini: kita mau bikin progress secepat yang kita bisa. Kita cuma perlu menyerahkan kembali kontrol ke runtime. Kita bisa melakukan itu secara langsung, menggunakan fungsi trpl::yield_now. Di Listing 17-17, kita mengganti semua pemanggilan trpl::sleep tadi dengan trpl::yield_now.

extern crate trpl; // required for mdbook test

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

fn main() {
    trpl::block_on(async {
        let a = async {
            println!("'a' started.");
            slow("a", 30);
            trpl::yield_now().await;
            slow("a", 10);
            trpl::yield_now().await;
            slow("a", 20);
            trpl::yield_now().await;
            println!("'a' finished.");
        };

        let b = async {
            println!("'b' started.");
            slow("b", 75);
            trpl::yield_now().await;
            slow("b", 10);
            trpl::yield_now().await;
            slow("b", 15);
            trpl::yield_now().await;
            slow("b", 350);
            trpl::yield_now().await;
            println!("'b' finished.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

Kode ini terasa lebih jelas soal niat aslinya dan bisa jadi jauh lebih kencang ketimbang memakai sleep, karena timers kayak yang dipakai sama sleep sering kali punya batas seberapa spesifik (granular) mereka bisa bekerja. Versi sleep yang kita pakai, misalnya, bakal selalu tidur setidaknya selama satu milidetik, biarpun kita memberinya Duration sebesar satu nanodetik. Sekali lagi, komputer modern itu cepat: mereka bisa melakukan banyak hal dalam satu milidetik!

Ini artinya asinkron bisa berguna bahkan buat tugas-tugas compute-bound, tergantung dari apa lagi yang lagi dilakukan sama program kita, karena dia menyediakan alat yang berguna buat menstrukturkan hubungan antara berbagai bagian program yang berbeda (tapi dengan biaya beban dari state machine asinkron tersebut). Ini adalah suatu bentuk cooperative multitasking, di mana tiap future punya kuasa buat menentukan kapan dia menyerahkan kontrol lewat titik-titik await. Oleh karena itu, tiap future juga punya tanggung jawab buat menghindari memblokir terlalu lama. Di beberapa sistem operasi embedded berbasis Rust, ini adalah satu-satunya jenis multitasking yang ada!

Di kode dunia nyata, tentu saja kita tidak bakal biasanya menyelingi pemanggilan fungsi dengan titik-titik await di tiap baris kodenya. Meskipun menyerahkan kontrol pakai cara ini terhitung murah biayanya, tetap saja ia tidak gratis. Di banyak kasus, mencoba memecah-mecah tugas compute-bound bisa jadi malah bikin dia jauh lebih lambat, jadi kadang-kadang lebih baik buat performa keseluruhan kalau kita membiarkan sebuah operasi memblokir sebentar. Selalu lakukan pengukuran (measure) buat melihat di mana sebenarnya letak bottlenecks performa kode kita. Dinamika dasarnya tetap penting buat diingat, terutama kalau kita melihat banyak pekerjaan yang terjadi secara serial padahal kita mengharapnya terjadi secara konkuren!

Membikin Abstraksi Asinkron Kita Sendiri

Kita juga bisa menggabungkan (compose) futures bersama-sama buat membikin pola baru. Misalnya, kita bisa membangun fungsi timeout dengan blok penyusun asinkron yang sudah kita punya. Pas kita sudah selesai, hasilnya bakal jadi blok penyusun lainnya yang bisa kita pakai buat membikin abstraksi asinkron yang lebih banyak lagi.

Listing 17-18 menunjukkan gimana ekspektasi kita soal cara kerja timeout ini bersama sebuah future yang lambat.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let slow = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finally finished"
        };

        match timeout(slow, Duration::from_secs(2)).await {
            Ok(message) => println!("Succeeded with '{message}'"),
            Err(duration) => {
                println!("Failed after {} seconds", duration.as_secs())
            }
        }
    });
}

Mari kita implementasikan ini! Sebagai permulaan, mari pikirkan soal API buat timeout:

• Ia sendiri harus berupa fungsi asinkron supaya kita bisa me-await-nya.
• Parameter pertamanya haruslah berupa sebuah future buat dijalankan. Kita bisa membikinnya generik biar dia bisa bekerja buat future apa saja.
• Parameter keduanya adalah waktu maksimal buat menunggu. Kalau kita memakai Duration, itu bakal gampang diteruskan ke trpl::sleep.
• Ia harus mengembalikan sebuah Result. Kalau future-nya sukses selesai, Result-nya bakal berupa Ok dengan nilai yang dihasilkan sama future tersebut. Kalau batas waktunya keburu habis duluan, Result-nya bakal berupa Err dengan durasi yang sudah dilewati sama si timeout.

Listing 17-19 menunjukkan deklarasi ini.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let slow = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finally finished"
        };

        match timeout(slow, Duration::from_secs(2)).await {
            Ok(message) => println!("Succeeded with '{message}'"),
            Err(duration) => {
                println!("Failed after {} seconds", duration.as_secs())
            }
        }
    });
}

async fn timeout<F: Future>(
    future_to_try: F,
    max_time: Duration,
) -> Result<F::Output, Duration> {
    // Here is where our implementation will go!
}

Itu sudah memenuhi tujuan kita buat tipe-tipenya. Sekarang mari kita pikirkan soal perilaku yang kita butuhkan: kita mau mengadu (race) future yang dimasukkan tadi melawan durasinya. Kita bisa memakai trpl::sleep buat membikin timer future dari durasinya, lalu memakai trpl::select buat menjalankan timer tersebut barengan sama future yang diberikan sama si pemanggil.

Di Listing 17-20, kita mengimplementasikan timeout dengan melakukan match pada hasil dari me-await trpl::select.

extern crate trpl; // required for mdbook test

use std::time::Duration;

use trpl::Either;

// --snip--

fn main() {
    trpl::block_on(async {
        let slow = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finally finished"
        };

        match timeout(slow, Duration::from_secs(2)).await {
            Ok(message) => println!("Succeeded with '{message}'"),
            Err(duration) => {
                println!("Failed after {} seconds", duration.as_secs())
            }
        }
    });
}

async fn timeout<F: Future>(
    future_to_try: F,
    max_time: Duration,
) -> Result<F::Output, Duration> {
    match trpl::select(future_to_try, trpl::sleep(max_time)).await {
        Either::Left(output) => Ok(output),
        Either::Right(_) => Err(max_time),
    }
}

Implementasi dari trpl::select itu tidak adil (not fair): ia selalu melakukan polling pada argumen-argumennya sesuai urutan saat mereka dioper (implementasi select lainnya biasanya bakal memilih argumen mana yang mau di-poll duluan secara acak). Maka dari itu, kita mengoper future_to_try ke select duluan biar dia dapet kesempatan buat selesai biarpun max_time itu durasinya sangat singkat. Kalau future_to_try beres duluan, select bakal mengembalikan Left yang isinya output dari future_to_try. Kalau timer beres duluan, select bakal mengembalikan Right berisi output dari timernya yaitu ().

Kalau future_to_try sukses dan kita dapet Left(output), kita mengembalikan Ok(output). Kalau sebaliknya si sleep timer yang habis waktunya dan kita dapet Right(()), kita mengabaikan () tersebut pakai _ lalu mengembalikan Err(max_time) sebagai gantinya.

Selesai deh, kita sudah punya fungsi timeout yang bisa jalan yang dibangun dari dua buah pembantu asinkron lainnya. Kalau kita jalankan kodenya, dia bakal mencetak mode kegagalan setelah timeout terjadi:

Failed after 2 seconds

Karena futures bisa digabung-gabungin bareng futures lain, kita bisa membangun alat-alat yang sangat kuat memakai blok penyusun asinkron yang kecil- kecil. Misalnya, kita bisa memakai pendekatan yang sama ini buat menggabungkan timeouts dengan retries, dan sebaliknya memakai itu semua bersama operasi lain kayak pemanggilan jaringan (seperti contoh yang ada di Listing 17-5).

Di praktiknya, biasanya kita bakal bekerja secara langsung dengan async dan await, dan secara sekunder memakai fungsi-fungsi kayak select dan macros kayak macro join! buat mengontrol gimana futures paling luarnya dieksekusi.

Kita sekarang sudah melihat sejumlah cara buat bekerja bareng banyak futures di saat yang bersamaan. Selanjutnya, kita bakal melihat gimana cara kita bekerja dengan banyak futures di dalam sebuah urutan seiring berjalannya waktu menggunakan streams.