Async Rust en Tokio

In les 1 schreef je async fn en .await zonder er veel over na te denken. Voor een webservice is het belangrijk te snappen wat er gebeurt, want één verkeerde regel kan je hele service laten haperen. In deze les experimenteer je in een apart voorbeeld en pas je het daarna toe op kassa-api.

Futures zijn lui

async fn zeg(tekst: &str) { println!("{tekst}"); }

#[tokio::main]
async fn main() {
    let f = zeg("hallo");   // er gebeurt nog niets
    println!("eerst dit");
    f.await;                // nu pas wordt "hallo" geprint
}

Een async fn geeft een future terug: een beschrijving van werk. Pas .await (of een runtime die hem uitvoert) laat het werk gebeuren. Bij elke .await waar nog niets klaar is, geeft de future de beurt terug aan de runtime, die dan iets anders doet.

De runtime

Tokio heeft een paar worker-threads (standaard één per CPU-kern) en verdeelt daarover duizenden taken. Elk binnenkomend verzoek is een taak. Wacht een taak op het netwerk of de database, dan draait de thread een andere taak. Zo bedient een kleine service veel gelijktijdige verzoeken.

Gelijktijdig wachten

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

async fn haal(naam: &str, ms: u64) -> String { sleep(Duration::from_millis(ms)).await; naam.to_string() }

#[tokio::main]
async fn main() {
    let t = Instant::now();
    let a = haal("voorraad", 300).await;
    let b = haal("prijzen", 300).await;
    println!("na elkaar: {:?}", t.elapsed());

    let t = Instant::now();
    let (a2, b2) = tokio::join!(haal("voorraad", 300), haal("prijzen", 300));
    println!("tegelijk: {:?}", t.elapsed());
}

Voorspel eerst: welke tijden zie je ongeveer?

Taken starten

let handle = tokio::spawn(async move {
    // draait als losse taak, mogelijk op een andere thread
    tel_dagomzet().await
});
let omzet = handle.await.unwrap();   // JoinHandle; Err als de taak panickte

Omdat een gespawnde taak naar een andere thread kan gaan, moet alles wat erin zit Send + 'static zijn: gebruik move en geef eigendom mee (of een Arc, les 6).

Nooit blokkeren

Blokkerend werk (zware berekening, std::fs, een synchrone bibliotheek, std::thread::sleep) houdt een worker-thread bezet. Met 8 threads en 8 trage verzoeken staat je hele service stil. Verplaats het:

let hash = tokio::task::spawn_blocking(move || bcrypt_achtige_hash(&wachtwoord)).await.unwrap();

spawn_blocking draait de closure op een aparte pool voor blokkerend werk. Voor bestanden heeft Tokio tokio::fs.

Timeouts en select!

use tokio::time::{timeout, Duration};
match timeout(Duration::from_secs(2), haal_koers()).await {
    Ok(koers) => println!("{koers}"),
    Err(_) => println!("te traag, gebruik de vorige koers"),
}

tokio::select! {
    r = haal_van_server_a() => println!("A was eerst: {r}"),
    r = haal_van_server_b() => println!("B was eerst: {r}"),
}

select! wacht op de eerste die klaar is en annuleert de andere (de future wordt gedropt).

Toepassen op kassa-api

Voeg een endpoint toe dat twee "trage bronnen" tegelijk raadpleegt:

async fn dashboard() -> Json<serde_json::Value> {
    let (open, omzet) = tokio::join!(open_bestellingen(), dagomzet());
    Json(serde_json::json!({ "open": open, "omzet_centen": omzet }))
}

Typische fouten

• future cannot be sent between threads safely: je houdt een niet-Send waarde (zoals Rc of een std::sync::MutexGuard) vast over een .await.
• .await vergeten: de compiler waarschuwt unused implementer of Future; het werk gebeurt dan nooit.
• Een std::sync::Mutex vasthouden over .await: kan vastlopen. Laat de guard los vóór de .await (les 6).

Oefenen

let mut handles = Vec::new();
for i in 0..10 {
    handles.push(tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(50 * (10 - i))).await;
        i
    }));
}
let mut uit = Vec::new();
for h in handles { uit.push(h.await.unwrap()); }
println!("{uit:?}"); // [0, 1, ..., 9]: volgorde van awaiten, niet van klaar zijn

async fn dashboard() -> Result<Json<serde_json::Value>, StatusCode> {
    let (open, omzet) = timeout(Duration::from_secs(1), async { tokio::join!(open_bestellingen(), dagomzet()) })
        .await.map_err(|_| StatusCode::GATEWAY_TIMEOUT)?;
    Ok(Json(serde_json::json!({ "open": open, "omzet_centen": omzet })))
}

let som = tokio::task::spawn_blocking(|| (1..=200_000_000u64).sum::<u64>()).await.unwrap();

Lokaal oefenen

Geen live lab. Maak voor de experimenten een los project (cargo new async-proef) met tokio (features full).

Samenvatting

• Futures zijn lui; .await voert ze uit en geeft de beurt terug als er gewacht moet worden.
• Tokio verdeelt taken over een paar threads; join! wacht tegelijk, spawn maakt losse taken.
• Nooit blokkeren in async code: gebruik spawn_blocking of async-varianten.
• timeout en select! begrenzen en kiezen.
• Volgende les: routing en handlers.