Il Linguaggio di Programmazione Rust

Restituire il Controllo al Runtime

Ricorda da “Il Nostro Primo Programma Async” che ad ogni punto di attesa, Rust dà a un runtime la possibilità di mettere in pausa il compito e passare a un altro se la future in attesa non è pronta. Anche l’inverso è vero: Rust mette in pausa solo i blocchi async e restituisce il controllo a un runtime in un punto di attesa. Tutto ciò che si trova tra i punti di attesa è sincrono.

Questo significa che se fai un sacco di lavoro in un blocco async senza un punto di attesa, quella future bloccherà qualsiasi altra future dal fare progressi. A volte potresti sentire menzionato questo comportamento come ad una future che affama (starving) altre future. In alcuni casi, potrebbe non essere un grosso problema. Tuttavia, se stai facendo qualche tipo di elaborazione dispendiosa o lavoro a lungo termine, o se hai una future che continuerà a fare un particolare compito indefinitamente, dovrai pensare a quando e dove restituire il controllo al runtime.

Simuliamo un’operazione a lungo termine per illustrare il problema dell’affamamento e come risolverlo. Il Listato 17-14 introduce una funzione lenta.

extern crate trpl; // necessario per test mdbook

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

fn main() {
    trpl::block_on(async {
        // Più tardi chiameremo `lenta` da qui
    });
}

fn lenta(nome: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{nome}' eseguita per {ms}ms");
}

Questo codice utilizza std::thread::sleep invece di trpl::sleep in modo che chiamare lenta blocchi il thread corrente per un certo numero di millisecondi. Possiamo usare lenta per rappresentare operazioni del mondo reale che sono sia a lungo termine che bloccanti.

Nel Listato 17-15, utilizziamo lenta per emulare questo tipo di lavoro legato alla CPU in un paio di future.

extern crate trpl; // necessario per test mdbook

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

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

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

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

fn lenta(nome: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{nome}' eseguita per {ms}ms");
}

Ogni future restituisce il controllo al runtime solo dopo aver eseguito alcune operazioni lente. Se esegui questo codice, vedrai questo output:

'a' iniziata.
'a' eseguita per 30ms
'a' eseguita per 10ms
'a' eseguita per 20ms
'b' iniziata.
'b' eseguita per 75ms
'b' eseguita per 10ms
'b' eseguita per 15ms
'b' eseguita per 350ms
'a' finita.

Come per il Listato 17-5 dove abbiamo usato trpl::select per mettere a gara due future che elaboravano un URL, select termina non appena a è completata. Non c’è “intreccio” tra le due future, però. La future a fa tutto il suo lavoro fino a quando la chiamata a trpl::sleep viene attesa con await, poi la future b fa tutto il suo lavoro fino a quando la sua chiamata a trpl::sleep viene attesa, e infine la future a finisce. Per consentire a entrambe le future lente di fare progressi, abbiamo bisogno di punti di attesa in modo da poter restituire il controllo al runtime di tanto in tanto per consentire anche all’altra di proseguire!

Possiamo già vedere questo tipo di passaggio avvenire nel Listato 17-15: se rimuovessimo trpl::sleep alla fine della future a, essa completerebbe la propria esecuzione senza che la future b nemmeno cominciasse. Proviamo a utilizzare la funzione trpl::sleep come punto di partenza per consentire alle operazioni di alternarsi nel fare progressi, come mostrato nel Listato 17-16.

extern crate trpl; // necessario per test mdbook

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

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

        let a = async {
            println!("'a' iniziata.");
            lenta("a", 30);
            trpl::sleep(un_ms).await;
            lenta("a", 10);
            trpl::sleep(un_ms).await;
            lenta("a", 20);
            trpl::sleep(un_ms).await;
            println!("'a' finita.");
        };

        let b = async {
            println!("'b' iniziata.");
            lenta("b", 75);
            trpl::sleep(un_ms).await;
            lenta("b", 10);
            trpl::sleep(un_ms).await;
            lenta("b", 15);
            trpl::sleep(un_ms).await;
            lenta("b", 350);
            trpl::sleep(un_ms).await;
            println!("'b' finita.");
        };

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

fn lenta(nome: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{nome}' eseguita per {ms}ms");
}

Abbiamo aggiunto chiamate a trpl::sleep con punti di attesa tra ogni chiamata a lenta. Ora il lavoro delle due future è intervallato:

'a' iniziata.
'a' eseguita per 30ms
'b' iniziata.
'b' eseguita per 75ms
'a' eseguita per 10ms
'b' eseguita per 10ms
'a' eseguita per 20ms
'b' eseguita per 15ms
'a' finita.

La future a continua a lavorare per un po’ prima di restituire il controllo a b, perché chiama lenta prima di chiamare trpl::sleep, ma dopo ciò le future si alternano ogni volta che una di esse incontra un punto di attesa. In questo caso, abbiamo fatto ciò dopo ogni chiamata a lenta, ma potremmo suddividere il lavoro in qualsiasi modo abbia più senso per noi.

Ma non vogliamo davvero dormire qui, però: vogliamo eseguire le nostre operazioni il più velocemente possibile e restituire il controllo al runtime quando possibile. Possiamo farlo direttamente, utilizzando la funzione trpl::yield_now. Nel Listato 17-17, sostituiamo tutte quelle chiamate a trpl::sleep con trpl::yield_now.

extern crate trpl; // necessario per test mdbook

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

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

        let b = async {
            println!("'b' iniziata.");
            lenta("b", 75);
            trpl::yield_now().await;
            lenta("b", 10);
            trpl::yield_now().await;
            lenta("b", 15);
            trpl::yield_now().await;
            lenta("b", 350);
            trpl::yield_now().await;
            println!("'b' finita.");
        };

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

fn lenta(nome: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{nome}' eseguita per {ms}ms");
}

Questo codice è sia più chiaro riguardo all’intento reale sia può essere significativamente più veloce rispetto all’uso di sleep, perché i timer come quello usato da sleep hanno spesso limiti su quanto possono essere granulari. La versione di sleep che stiamo usando, ad esempio, dormirà sempre per almeno un millisecondo, anche se le passiamo una Duration di un nanosecondo. Ancora una volta, i computer moderni sono veloci: possono fare molto in un millisecondo!

Questo significa che l’async può essere utile anche per compiti legati al calcolo, a seconda di cosa sta facendo il tuo programma, perché fornisce uno strumento utile per strutturare le relazioni tra le diverse parti del programma (ma con un costo prestazionale per la macchina a stati async). Questa è una forma di multitasking cooperativo, in cui ogni future ha il potere di determinare quando restituisce il controllo tramite i punti di attesa. Ogni future ha quindi anche la responsabilità di evitare di bloccarsi troppo a lungo. In alcuni sistemi operativi embedded basati su Rust, questo è l’unico tipo di multi-tasking!

Nel codice reale, di solito non lavorerai direttamente alternando chiamate di funzione con punti di attesa su ogni singola riga, ovviamente. Anche se restituire il controllo in questo modo è relativamente poco costoso, non è gratuito. In molti casi, cercare di suddividere un compito legato al calcolo potrebbe renderlo significativamente più lento, quindi a volte è meglio per le prestazioni complessive lasciare che un’operazione si blocchi brevemente. Misura sempre per vedere quali sono i veri colli di bottiglia delle prestazioni del tuo codice. Tuttavia, la dinamica sottostante è importante da tenere a mente, se stai vedendo molto lavoro avvenire in serie che ti aspettavi avvenisse in parallelo!

Costruire le Nostre Astrazioni Async

Possiamo anche comporre le future insieme per creare nuovi modelli. Ad esempio, possiamo costruire una funzione timeout con i blocchi async che abbiamo già. Quando abbiamo finito, il risultato sarà un altro blocco di costruzione che potremmo usare per creare ancora più astrazioni async.

Il Listato 17-18 mostra come ci aspettiamo che funzioni questo timeout con una future lento.

extern crate trpl; // necessario per test mdbook

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let lento = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finalmente finito"
        };

        match timeout(lento, Duration::from_secs(2)).await {
            Ok(messaggio) => println!("Completato con '{messaggio}'"),
            Err(durata) => {
                println!("Fallito dopo {} secondi", durata.as_secs())
            }
        }
    });
}

Implementiamolo! Per cominciare, pensiamo all’API per timeout:

• Deve essere essa stessa una funzione async in modo da poterla attendere.
• Il suo primo parametro dovrebbe essere una future da eseguire. Possiamo renderla generica per consentirle di funzionare con qualsiasi future.
• Il suo secondo parametro sarà il tempo massimo da attendere. Se usiamo una Duration, sarà facile passarla a trpl::sleep.
• Dovrebbe restituire un Result. Se la future completa con successo, il Result sarà Ok con il valore prodotto dalla future. Se il timeout scade prima, il Result sarà Err con la durata che il timeout ha atteso.

Il Listato 17-19 mostra questa dichiarazione.

extern crate trpl; // necessario per test mdbook

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let lento = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finalmente finito"
        };

        match timeout(lento, Duration::from_secs(2)).await {
            Ok(messaggio) => println!("Completato con '{messaggio}'"),
            Err(durata) => {
                println!("Fallito dopo {} secondi", durata.as_secs())
            }
        }
    });
}

async fn timeout<F: Future>(
    future_da_testare: F,
    tempo_massimo: Duration,
) -> Result<F::Output, Duration> {
    // Qui è dove metteremo l'implementazione!
}

Questo soddisfa i nostri obiettivi per i type. Ora pensiamo al comportamento di cui abbiamo bisogno: vogliamo far competere la future passata contro la durata fornita. Possiamo usare trpl::sleep per creare una future che duri quanto richiesto e usare trpl::select per eseguirla contro la future che il chiamante passa.

Nel Listato 17-20, implementiamo timeout facendo il match sul risultato dell’attesa di trpl::select.

extern crate trpl; // necessario per test mdbook

use std::time::Duration;

use trpl::Either;

// --taglio--

fn main() {
    trpl::block_on(async {
        let lento = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finalmente finito"
        };

        match timeout(lento, Duration::from_secs(2)).await {
            Ok(messaggio) => println!("Completato con '{messaggio}'"),
            Err(durata) => {
                println!("Fallito dopo {} secondi", durata.as_secs())
            }
        }
    });
}

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

L’implementazione di trpl::select non è equa: processa gli argomenti sempre nell’ordine in cui sono passati (altre implementazioni di select scelgono a caso quale argomento processare per primo). Pertanto, passiamo future_da_testare a select per prima in modo che abbia la possibilità di completare anche se tempo_massimo è una durata molto breve. Se future_da_testare finisce prima, select restituirà Left con l’output da future_da_testare. Se il timer finisce prima, select restituirà Right con l’output del timer di ()

Se future_da_testare ha successo e otteniamo un Left(output), restituiamo Ok(output). Se invece il timer finisce prima e otteniamo un Right(()), ignoriamo il () con _ e restituiamo Err(tempo_massimo).

Con questo, abbiamo un timeout funzionante combinando più blocchi async. Se eseguiamo il nostro codice, stamperà la modalità di errore dopo il timeout:

Fallito dopo 2 secondi

Poiché le future si compongono con altre future, puoi costruire strumenti davvero potenti utilizzando blocchi di costruzione async più piccoli. Ad esempio, puoi utilizzare questo stesso approccio per combinare timeout con ripetizioni, e a loro volta usarli con operazioni come chiamate di rete (come quelli nel Listato 17-5).

Nella pratica, di solito lavorerai direttamente con async e await, e secondariamente con funzioni come select e macro come join! per controllare come vengano eseguite le varie future.

Abbiamo ora visto diversi modi per lavorare con più future contemporaneamente. Prossimamente, vedremo come possiamo lavorare con più future in una sequenza nel tempo con gli stream.