Async in profondità in Rust: select!, timeout, stream e Pin

Nella pagina su async/await hai imparato le basi: una async fn restituisce un future pigro, .await lo esegue, #[tokio::main] avvia il runtime, e tokio::join! e tokio::spawn eseguono cose nello stesso momento. Se qualcosa di tutto questo ti suona nuovo, leggi prima quella pagina. Questa ci costruisce sopra.

I programmi async reali hanno bisogno di qualche strumento in più. Torniamo alla cucina affollata. A volte aspetti tutti i piatti (join!). A volte ti interessa solo la prima cosa che succede: il piatto è pronto, oppure il cliente se ne va (select!). A volte lasci perdere dopo cinque minuti (timeout). E a volte le ordinazioni continuano ad arrivare, una dopo l’altra, per tutta la sera (gli stream). Questa pagina li copre tutti, più Pin, la strana parola che prima o poi vedrai nei messaggi di errore, spiegata in parole semplici.

Tutti gli esempi usano tokio con features = ["full"], proprio come la pagina delle basi. I pulsanti «Provalo tu» funzionano, perché il Playground di Rust include già tokio.

Eseguire tante cose insieme

Conosci già tokio::join!: aspetta tutti i suoi future e ne restituisce i risultati in una tupla. Suo fratello tokio::try_join! è per i future che restituiscono un Result. Ti dà Ok con tutti i valori, oppure il primo Err, e smette di aspettare gli altri appena uno fallisce:

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!("profilo #{id}"))
}

async fn load_settings(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(10)).await;
    if id == 2 {
        return Err(format!("nessuna impostazione per l'utente {id}"));
    }
    Ok(String::from("tema scuro"))
}

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

Output:

utente 1: profilo #1, tema scuro
utente 2, errore: nessuna impostazione per l'utente 2

▶ Provalo tu

join! e try_join! eseguono i future dentro il task corrente, a turno. È perfetto per una manciata di future che conosci in anticipo. Quando hai un elenco di lavori la cui lunghezza conosci solo durante l’esecuzione, avviali come task separati. Potresti mettere ogni JoinHandle in un Vec, come nella pagina delle basi. Ma tokio ha uno strumento più comodo: JoinSet, un sacchetto di task che ti consegna ogni risultato appena è pronto:

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; // fingiamo che sia una chiamata di rete
    (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 solo se il task è andato in panic
    }

    results.sort(); // l'ordine di arrivo può cambiare da un'esecuzione all'altra
    for (site, ms) in results {
        println!("{site} ha risposto in {ms}ms");
    }
}

Output:

crates.io ha risposto in 20ms
docs.rs ha risposto in 10ms
rust-lang.org ha risposto in 30ms

▶ Provalo tu

join_next() restituisce None quando il set è vuoto, quindi il ciclo while let finisce da solo. I risultati arrivano in ordine di completamento, non di avvio. Quando i task richiedono tempi simili, quell’ordine può cambiare tra un’esecuzione e l’altra, perciò ordiniamo prima di stampare. Un’altra cosa comoda: se elimini un JoinSet, ogni task ancora al suo interno viene cancellato. Niente resta a girare per sbaglio.

Mettere in gara i future con select!

join! aspetta tutto. tokio::select! aspetta il primo. Ogni ramo ha un pattern, un future e un blocco di codice: pattern = future => codice. Il future che finisce per primo esegue il suo codice, e gli altri rami vengono cancellati.

Che cosa vuol dire «cancellato» in Rust? Il future perdente viene semplicemente eliminato (drop), come qualsiasi altro valore che esce dallo scope. Non viene più eseguito. Anche le sue variabili locali vengono eliminate, quindi il loro codice di pulizia parte. Questo esempio usa un piccolo tipo con Drop per rendere la cosa visibile:

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

struct Download(&'static str);

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

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

#[tokio::main]
async fn main() {
    tokio::select! {
        winner = download("Europa", 50) => println!("vincitore: {winner}"),
        winner = download("Asia", 1000) => println!("vincitore: {winner}"),
    }
    println!("main prosegue");
}

Output:

download Europa completato
pulizia del download Europa
pulizia del download Asia
vincitore: Europa
main prosegue

▶ Provalo tu

Guarda cosa manca: «download Asia completato» non viene mai stampato. Il future Asia è stato fermato al suo .await ed eliminato, e il suo valore Download è stato ripulito subito. È così che funziona la cancellazione ovunque nell’async di Rust: smetti di interrogare (poll) un future ed eliminalo. Non serve nessuna chiamata speciale per «cancellare».

Un uso classico di select! è il timeout: metti in gara il tuo lavoro contro un timer.

tokio::select! {
    report = build_report() => println!("{report}"),
    _ = tokio::time::sleep(Duration::from_secs(5)) => println!("troppo lento, lascio perdere"),
}

Funziona bene, ma c’è una scorciatoia apposta, tokio::time::timeout, che vedrai nella prossima sezione.

Lavorare fino a un segnale di arresto

select! dà il meglio dentro un loop. Qui un worker prende lavori da un canale, ma ascolta anche un messaggio «per favore, fermati» su un secondo canale. Un canale oneshot trasporta esattamente un messaggio, il che lo rende un segnale di arresto perfetto:

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!("al lavoro su: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("segnale di arresto ricevuto");
                    break;
                }
            }
        }
        done
    });

    for job in ["ridimensiona foto", "invia email", "aggiorna indice di ricerca"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    sleep(Duration::from_millis(100)).await; // dai al worker il tempo di finire
    stop_tx.send(()).unwrap();

    let done = worker.await.unwrap();
    println!("il worker ha completato {done} lavori");
}

Output:

al lavoro su: ridimensiona foto
al lavoro su: invia email
al lavoro su: aggiorna indice di ricerca
segnale di arresto ricevuto
il worker ha completato 3 lavori

▶ Provalo tu

Alcuni dettagli utili da sapere:

• Scriviamo &mut stop_rx, non stop_rx. A ogni giro del ciclo, select! ha bisogno di nuovo dello stesso destinatario, quindi glielo prestiamo invece di spostarlo nel primo giro.
• A ogni giro del ciclo, job_rx.recv() crea un future nuovo, e quello perdente viene eliminato. Va bene così, perché recv() è cancel-safe (sicuro da cancellare): eliminarlo non fa mai perdere un messaggio.
• Se il pattern non corrisponde (qui, recv() restituisce None perché tutti i mittenti sono spariti), select! disattiva quel ramo e continua ad aspettare gli altri.
• Se più rami sono pronti nello stesso momento, select! ne sceglie uno a caso, così nessun ramo può monopolizzare il ciclo. Scrivi biased; come prima riga dentro select! per controllare invece i rami dall’alto in basso.

La sicurezza della cancellazione, in un paragrafo. Visto che il ramo perdente viene eliminato, il lavoro che aveva fatto a metà va perso. Per recv() su un canale o per sleep() non è un problema: non era ancora stato preso niente. Ma immagina un future che legge 100 byte da un socket di rete e ne ha già letti 60 quando viene eliminato. Quei 60 byte sono persi. Un future del genere non è cancel-safe. La documentazione di tokio dice, per ogni metodo, se è cancel-safe. Quando usi select! in un ciclo, controlla i metodi che chiami, oppure crea il future una volta sola fuori dal ciclo e continua a interrogare quello stesso future (vedrai come nella sezione su Pin).

Timeout e timer

tokio::time::timeout(durata, future) avvolge qualsiasi future. Se il future finisce in tempo ottieni Ok(valore). Se il tempo scade prima, il future viene eliminato (cancellato, proprio come un ramo perdente di select!) e ottieni 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 veloce", 20), ("server lento", 2000)] {
        match timeout(limit, fetch_price(delay)).await {
            Ok(price) => println!("{server}: il prezzo è {price}"),
            Err(e) => println!("{server}: {e}"),
        }
    }
}

Output:

server veloce: il prezzo è 42
server lento: deadline has elapsed

▶ Provalo tu

«deadline has elapsed» è il messaggio dell’errore Elapsed di tokio, che resta in inglese: significa «il tempo limite è scaduto». Metti un timeout su tutto ciò che parla con il mondo esterno: chiamate HTTP, query al database, letture da un socket. Un server che non risponde mai non deve bloccare il tuo programma per sempre.

sleep o interval. sleep aspetta una volta. interval ti dà un ritmo costante, come un metronomo. Chiami .tick().await in un ciclo, e ogni tick arriva puntuale. Il primo tick scatta subito:

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; // il primo tick si completa subito
        println!("battito {beat}");
    }
}

Output:

battito 1
battito 2
battito 3

▶ Provalo tu

Perché non fare semplicemente sleep di 50 ms in un ciclo? Perché così ogni giro dura «il tuo lavoro + 50 ms», e la tabella di marcia slitta piano piano. Un interval misura dall’inizio previsto di ogni tick, quindi se il tuo lavoro richiede 10 ms, aspetta solo i 40 ms rimanenti. Usa interval per heartbeat, polling e lavori del tipo «salva ogni minuto». Usa sleep per una singola pausa.

Stream: iteratori async

Un iteratore ti dà i valori uno alla volta con next(). Uno stream è la versione async: next().await ti dà il valore successivo quando è pronto, e None quando non ce ne sono più. Pensa a un flusso di prezzi di borsa, a righe che arrivano da un database o ai messaggi di un server di chat.

In realtà hai già usato qualcosa di simile a uno stream: il destinatario di un canale tokio. rx.recv().await aspetta il messaggio successivo e restituisce None quando tutti i mittenti sono spariti:

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; // una lettura ogni 10 ms
        }
        // qui tx viene eliminato, e questo chiude il canale
    });

    while let Some(temp) = rx.recv().await {
        println!("temperatura: {temp}°C");
    }
    println!("sensore fermato");
}

Output:

temperatura: 21°C
temperatura: 22°C
temperatura: 24°C
temperatura: 23°C
sensore fermato

▶ Provalo tu

La libreria standard non ha ancora un trait Stream (è ancora in fase di progettazione). L’ecosistema usa quello del crate futures, e il crate di supporto di tokio, tokio-stream, aggiunge un trait StreamExt con i metodi che conosci dagli iteratori: next(), map(), filter(), take() e altri. Aggiungilo al tuo progetto:

[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}");
    }
}

Questo stampa 200, 204 e 196. Si legge proprio come il codice con gli iteratori, a parte il .await su next(). Per trasformare il destinatario di un canale tokio in un vero stream, avvolgilo: tokio_stream::wrappers::ReceiverStream::new(rx). (Il crate futures ha il suo StreamExt con la stessa idea. Scegline uno per file, perché entrambi i trait hanno un metodo next() e vanno in conflitto.)

Pin, in parole semplici

Prima o poi vedrai Pin in un tipo o in un messaggio di errore. Ecco l’idea, senza la teoria profonda.

Quando Rust compila una async fn, la trasforma in una struct nascosta, una macchina a stati. Ogni variabile locale che serve ancora dopo un .await diventa un campo di quella struct. Ora immagina che una variabile locale prenda in prestito un’altra, come let name = String::from("Ana"); let r = &name; con un .await nel mezzo. A quel punto la struct contiene un puntatore a un suo stesso campo. Se spostassi la struct in un altro punto della memoria, quel puntatore punterebbe al vecchio posto, ormai vuoto.

Pin è la soluzione. Un valore «pinnato» (fissato) è una promessa: «questo valore non si sposterà più in memoria.» Una volta fissato un future, è sicuro interrogarlo. Quasi sempre non te ne accorgi nemmeno, perché .await fissa il future al posto tuo.

Incontrerai Pin in tre situazioni:

1. Interrogare lo stesso future più volte, per esempio una scadenza in un ciclo con select!.
2. Conservare dei future in un Vec o nel campo di una struct, di solito come Pin<Box<dyn Future<Output = T>>>.
3. La ricorsione in una async fn.

Per il primo caso, ecco cosa succede se provi a riusare uno sleep in un ciclo senza fissarlo:

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`

Il compilatore ti suggerisce perfino la soluzione: usare la macro pin!, oppure Box::pin. std::pin::pin! (o tokio::pin!) fissa il future sul posto, nello stack. Ora la stessa scadenza è condivisa da ogni giro del ciclo:

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; // poi il negozio si fa silenzioso
    });

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

    loop {
        tokio::select! {
            _ = &mut deadline => {
                println!("orario di chiusura");
                break;
            }
            Some(order) = rx.recv() => println!("ricevuto l'ordine #{order}"),
        }
    }
}

Output:

ricevuto l'ordine #1
ricevuto l'ordine #2
ricevuto l'ordine #3
orario di chiusura

▶ Provalo tu

Se scrivessimo sleep(...) dentro il select!, ogni nuovo ordine farebbe partire un timer nuovo da 300 ms, e un negozio affollato non chiuderebbe mai. Fissare una sola scadenza fuori dal ciclo risolve il problema.

Per la ricorsione, il compilatore ha bisogno di sapere quanto è grande il future, e un future che contiene sé stesso sarebbe infinitamente grande. Chiamare countdown(n - 1).await dentro async fn countdown dà:

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

Il messaggio dice che la ricorsione in una async fn richiede un «box». Box::pin sposta il future interno nell’heap e lo fissa lì. Il future esterno ora contiene solo un puntatore, che ha una dimensione fissa:

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

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

Output:

3...
2...
1...
decollo!

▶ Provalo tu

Sinceramente, è tutto quello che serve alla maggior parte delle persone: pin! per un future che interroghi più volte, Box::pin per conservare future o per la ricorsione. Se vuoi la storia completa, il capitolo del Rust Book nell’elenco «Per approfondire» la spiega passo dopo passo.

async fn nei trait e closure async

Da Rust 1.75 puoi scrivere async fn direttamente in un trait. Da Rust 1.85 hai anche le closure async, scritte async |x| ..., con i relativi trait AsyncFn, AsyncFnMut e AsyncFnOnce (le versioni async dei trait delle closure). Eccoli entrambi in un solo esempio:

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; // fingiamo di parlare con un server di posta
        format!("email inviata a {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 inviata a ana
email inviata a ben

▶ Provalo tu

C’è un intoppo con async fn nei trait. Il trait non promette che il future restituito sia Send, e tokio::spawn ha bisogno di future Send (possono spostarsi tra thread). Quindi avviare una chiamata attraverso un generico N: Notifier fallisce:

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;

L’errore dice che il future non può essere mandato tra thread in sicurezza, e il suggerimento del compilatore è la soluzione abituale: scrivere il metodo del trait come fn ... -> impl Future<Output = String> + Send. Chi implementa il trait può comunque scrivere una normale async fn. Nota anche che i trait con async fn non si possono ancora usare come trait object dyn Notifier. Se ti serve, il popolare crate async-trait mette i future in un box al posto tuo.

Errori comuni

Bloccare il runtime. std::thread::sleep, calcoli pesanti o grosse letture con std::fs dentro un task async congelano l’intero thread worker, e ogni altro task su quel thread aspetta. La pagina delle basi ha mostrato la versione con sleep. Per il lavoro pesante sulla CPU o bloccante, affidalo a tokio::task::spawn_blocking, che lo esegue su un pool di thread separato fatto apposta:

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!("rispondo ancora alle richieste mentre contiamo...");

    let primes = handle.await.unwrap();
    println!("ci sono {primes} numeri primi sotto 50000");
}

Output:

rispondo ancora alle richieste mentre contiamo...
ci sono 5133 numeri primi sotto 50000

▶ Provalo tu

Tenere la guardia di un std::sync::Mutex attraverso un .await. Sembra innocuo:

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

Il compilatore ti sta proteggendo: la MutexGuard non è Send ed è ancora viva durante l’.await. Mentre questo task dorme all’.await, terrebbe ancora il lock, e qualsiasi altro task che vuole il lock bloccherebbe il suo intero thread. La soluzione è eliminare la guardia prima dell’.await, mettendo il lock in un blocco a parte:

let count = {
    let mut count = visits.lock().unwrap();
    *count += 1;
    *count
}; // qui la guardia viene eliminata
save_to_disk(count).await;

Se hai davvero bisogno di tenere un lock mentre aspetti, usa tokio::sync::Mutex, il cui lock().await aspetta senza bloccare il thread. (Per saperne di più sui lock, vedi Stato condiviso.)

Dimenticare .await. I future sono pigri, quindi save_file(); senza .await non fa proprio niente. Il compilatore ti avvisa con «futures do nothing unless you .await or poll them» (i future non fanno niente se non fai .await o non li interroghi). Vedi Async/await per tutta la storia, e non ignorare quel warning.

Avviare un task e non aspettarlo mai. Quando main ritorna, il runtime si spegne e ogni task non finito viene eliminato:

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

#[tokio::main]
async fn main() {
    tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        println!("report salvato"); // non viene mai stampato!
    });

    println!("main ha finito");
}

Output:

main ha finito

▶ Provalo tu

Il report non viene mai salvato. Se il lavoro è importante, tieni il JoinHandle e fai .await su di lui (oppure metti i task in un JoinSet e svuotalo) prima che main finisca.

Pratica

1. Confronto prezzi con una scadenza

Quattro negozi rispondono a una richiesta di prezzo dopo 20, 40, 300 e 60 millisecondi. Interrogali tutti e quattro nello stesso momento con un JoinSet, ma dai a ogni negozio solo 150 ms. Stampa i prezzi arrivati (ordinati per nome del negozio), poi quanti negozi non hanno risposto in tempo. Suggerimento: metti nel set timeout(limit, ask_shop(name, delay)). Ogni task finito ti dà allora un Ok o un Err(Elapsed), quindi restituisci il nome del negozio insieme a quel risultato.

Mostra la soluzione

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 // un prezzo inventato
}

#[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!("negozi fuori tempo: {timed_out}");
}

Output:

Bits & Bobs: 160
Corner Shop: 120
MegaMart: 140
negozi fuori tempo: 1

▶ Provalo tu

Tutti i negozi aspettano nello stesso momento, quindi l’intero controllo dura circa 150 ms, per quanto lento sia SlowCo. Quando scatta il suo timeout, il suo future viene eliminato, quindi non può continuare a girare in background.

2. Dimensione delle cartelle con la ricorsione async

Una Folder ha un nome, un elenco di dimensioni di file in KB e un elenco di sottocartelle. Scrivi async fn total_size(folder: &Folder) -> u64, che somma i suoi file più la dimensione di ogni sottocartella, chiamando sé stessa per ciascuna. Fai finta che ogni cartella abbia prima bisogno di una breve «lettura del disco» async, con tokio::task::yield_now().await. Suggerimento: la chiamata ricorsiva va avvolta in Box::pin(...), altrimenti ottieni l’errore E0733.

Mostra la soluzione

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

async fn total_size(folder: &Folder) -> u64 {
    tokio::task::yield_now().await; // fingiamo di leggere il disco

    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: "casa",
        files_kb: vec![4],
        subfolders: vec![
            Folder { name: "foto", files_kb: vec![2048, 1024], subfolders: vec![] },
            Folder {
                name: "codice",
                files_kb: vec![12, 8],
                subfolders: vec![Folder { name: "target", files_kb: vec![500], subfolders: vec![] }],
            },
        ],
    };

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

Output:

foto: 3072 KB
target: 500 KB
codice: 520 KB
casa: 3596 KB
totale: 3596 KB

▶ Provalo tu

Ogni cartella stampa dopo tutte le sue sottocartelle, perché prima deve aspettare le loro dimensioni. Box::pin mette nell’heap il future di ogni chiamata annidata, così il future esterno resta di dimensione fissa.

3. Svuota la coda, poi fermati

Modifica l’esempio dell’arresto in modo che non serva lo sleep(100ms) prima di mandare il segnale di stop. Manda subito i tre lavori e il segnale di stop, ma fai in modo che il worker finisca comunque tutti i lavori in coda prima di fermarsi. Suggerimento: senza aiuto, select! sceglie un ramo a caso quando sono pronti entrambi. Aggiungi biased; e metti per primo il ramo dei lavori.

Mostra la soluzione

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 ["ridimensiona foto", "invia email", "aggiorna indice di ricerca"] {
        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!("al lavoro su: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("coda vuota, mi fermo");
                    break;
                }
            }
        }
        done
    });

    let done = worker.await.unwrap();
    println!("il worker ha completato {done} lavori");
}

Output:

al lavoro su: ridimensiona foto
al lavoro su: invia email
al lavoro su: aggiorna indice di ricerca
coda vuota, mi fermo
il worker ha completato 3 lavori

▶ Provalo tu

Con biased;, select! prova sempre per primo il ramo dei lavori, quindi guarda il segnale di stop solo quando recv() non ha niente di pronto. Senza, il worker potrebbe fermarsi dopo uno o due lavori: un bug che si presenterebbe solo ogni tanto.

Riepilogo

• join! aspetta tutti i future, try_join! si ferma al primo Err, e JoinSet esegue tanti task avviati e ti consegna i risultati man mano che finiscono (ordinali se l’ordine conta).
• select! aspetta che finisca il primo future. Gli altri vengono eliminati: è così che funziona la cancellazione nell’async di Rust. Dentro un ciclo, usa future cancel-safe e biased; quando l’ordine conta.
• timeout rinuncia a un future lento. interval scandisce un ritmo costante, sleep aspetta una volta.
• Uno stream è un iteratore async. Il destinatario di un canale funziona come uno stream, e tokio-stream aggiunge map, take e compagnia.
• Pin promette che un future non si sposterà in memoria. Usa pin! per interrogare più volte lo stesso future e Box::pin per i future conservati e la ricorsione async.
• Non bloccare il runtime (spawn_blocking), non tenere un lock della std attraverso un .await, e aspetta i task che avvii.

Verifica veloce

Verifica veloce

1. In tokio::select!, che fine fanno i rami che non finiscono per primi?

Continuano a girare in background Vengono eliminati, quindi si fermano e fanno pulizia Ripartono al giro successivo del ciclo

2. Che cosa restituisce tokio::time::timeout quando il tempo scade?

Va in panic Ok con un valore predefinito Err(Elapsed), e il future interno viene eliminato

3. Perché serve Box::pin quando una async fn chiama sé stessa?

Un future che contiene sé stesso avrebbe una dimensione infinita La ricorsione non è proprio permessa nel codice async Per far girare la funzione su un altro thread

4. Devi fare 2 secondi di lavoro pesante sulla CPU dentro un programma tokio. Qual è lo strumento migliore?

std::thread::sleep intorno al lavoro tokio::select! Un JoinSet con tokio::spawn tokio::task::spawn_blocking

Punteggio: 0 / 4

Per approfondire

• Select — il tutorial di Tokio (in inglese): select! nel dettaglio: come funzionano i rami, cosa significa la cancellazione e come usarlo in un ciclo.
• Streams — il tutorial di Tokio (in inglese): come usare, adattare e costruire stream con tokio-stream.
• A Closer Look at the Traits for Async — The Rust Book (in inglese): uno sguardo graduale, passo dopo passo, a Future, Pin, Unpin e Stream.
• std::pin — libreria standard di Rust (in inglese): la spiegazione ufficiale del pinning, di Pin, Unpin e della macro pin!.
• tokio::time su docs.rs (in inglese): sleep, timeout, interval e come vengono gestiti i tick persi.