Async en profondeur en Rust : select!, timeouts, streams et Pin

Sur la page Async/Await, vous avez appris les bases : une async fn renvoie un future paresseux, .await l’exécute, #[tokio::main] démarre le runtime, et tokio::join! et tokio::spawn font tourner plusieurs choses en même temps. Si tout cela vous semble nouveau, lisez d’abord cette page. Celle-ci s’appuie dessus.

Les vrais programmes async ont besoin de quelques outils de plus. Reprenons l’image de la cuisine animée. Parfois, vous attendez tous les plats (join!). Parfois, seul compte le premier événement : le plat est prêt, ou le client s’en va (select!). Parfois, vous abandonnez au bout de cinq minutes (timeout). Et parfois, les commandes arrivent une par une, toute la soirée (les streams). Cette page couvre tout cela, plus Pin, ce mot étrange que vous finirez par croiser dans les messages d’erreur, expliqué avec des mots simples.

Tous les exemples utilisent tokio avec features = ["full"], comme sur la page des bases. Les boutons « Essayez par vous-même » fonctionnent, car le Playground Rust intègre tokio.

Lancer plusieurs choses à la fois

Vous connaissez déjà tokio::join! : il attend tous ses futures et renvoie leurs résultats dans un tuple. Son cousin tokio::try_join! est fait pour les futures qui renvoient un Result. Il vous donne Ok avec toutes les valeurs, ou la première Err, et il arrête d’attendre les autres dès que l’un échoue :

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 n°{id}"))
}

async fn load_settings(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(10)).await;
    if id == 2 {
        return Err(format!("aucun réglage pour l'utilisateur {id}"));
    }
    Ok(String::from("mode sombre"))
}

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

Sortie :

utilisateur 1 : profil n°1, mode sombre
échec pour l'utilisateur 2 : aucun réglage pour l'utilisateur 2

▶ Essayez par vous-même

join! et try_join! exécutent les futures dans la tâche courante, chacun son tour. C’est parfait pour une poignée de futures connus à l’avance. Quand vous avez une liste de travaux dont la taille n’est connue qu’à l’exécution, lancez-les comme des tâches séparées. Vous pourriez ranger chaque JoinHandle dans un Vec, comme sur la page des bases. Mais tokio propose un outil plus agréable pour ça : JoinSet, un sac de tâches qui vous rend chaque résultat dès qu’il est terminé :

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; // imaginons que c'est un appel réseau
    (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 seulement si la tâche a paniqué
    }

    results.sort(); // l'ordre de fin peut changer d'une exécution à l'autre
    for (site, ms) in results {
        println!("{site} a répondu en {ms}ms");
    }
}

Sortie :

crates.io a répondu en 20ms
docs.rs a répondu en 10ms
rust-lang.org a répondu en 30ms

▶ Essayez par vous-même

join_next() renvoie None quand le set est vide, donc la boucle while let s’arrête toute seule. Les résultats arrivent dans l’ordre de fin, pas dans l’ordre de lancement. Quand les tâches durent à peu près le même temps, cet ordre peut changer d’une exécution à l’autre : c’est pourquoi on trie avant d’afficher. Autre point appréciable : si vous libérez (drop) un JoinSet, toutes les tâches encore dedans sont annulées. Rien ne reste en marche par accident.

Faire la course entre futures avec select!

join! attend tout le monde. tokio::select! attend le premier. Chaque branche a un pattern, un future et un bloc de code : pattern = future => code. Le premier future qui termine voit son code s’exécuter, et les autres branches sont annulées.

Que veut dire « annulé » en Rust ? Le future perdant est tout simplement libéré (dropped), comme n’importe quelle valeur qui sort de sa portée. Il ne s’exécute plus jamais. Ses variables locales sont libérées elles aussi, donc leur code de nettoyage s’exécute. Cet exemple utilise un petit type avec Drop pour le rendre visible :

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

struct Download(&'static str);

impl Drop for Download {
    fn drop(&mut self) {
        println!("nettoyage du téléchargement depuis {}", self.0);
    }
}

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

#[tokio::main]
async fn main() {
    tokio::select! {
        winner = download("Europe", 50) => println!("gagnant : {winner}"),
        winner = download("Asie", 1000) => println!("gagnant : {winner}"),
    }
    println!("main continue");
}

Sortie :

téléchargement depuis Europe terminé
nettoyage du téléchargement depuis Europe
nettoyage du téléchargement depuis Asie
gagnant : Europe
main continue

▶ Essayez par vous-même

Regardez ce qui manque : « téléchargement depuis Asie terminé » n’est jamais affiché. Le future Asie a été stoppé à son .await puis libéré, et sa valeur Download a été nettoyée tout de suite. C’est ainsi que fonctionne l’annulation partout en Rust async : on arrête de poller un future et on le libère. Pas besoin d’un appel spécial « annuler ».

Un usage classique de select! est le timeout (délai maximal) : vous faites la course entre votre travail et un minuteur.

tokio::select! {
    report = build_report() => println!("{report}"),
    _ = tokio::time::sleep(Duration::from_secs(5)) => println!("trop lent, on abandonne"),
}

Ça marche très bien, mais il existe un raccourci, tokio::time::timeout, présenté dans la section suivante.

Travailler jusqu’au signal d’arrêt

select! donne toute sa mesure dans une loop. Ici, un worker (une tâche de fond qui traite des travaux) prend des travaux dans un canal, mais écoute aussi un message « merci de t’arrêter » sur un second canal. Un canal oneshot transporte exactement un message, ce qui en fait un signal d’arrêt idéal :

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!("en cours : {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("signal d'arrêt reçu");
                    break;
                }
            }
        }
        done
    });

    for job in ["redimensionner la photo", "envoyer l'e-mail", "mettre à jour l'index de recherche"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    sleep(Duration::from_millis(100)).await; // laisser au worker le temps de finir
    stop_tx.send(()).unwrap();

    let done = worker.await.unwrap();
    println!("le worker a terminé {done} travaux");
}

Sortie :

en cours : redimensionner la photo
en cours : envoyer l'e-mail
en cours : mettre à jour l'index de recherche
signal d'arrêt reçu
le worker a terminé 3 travaux

▶ Essayez par vous-même

Quelques détails bons à savoir :

• On écrit &mut stop_rx, pas stop_rx. À chaque tour de boucle, select! a besoin du même récepteur, donc on le prête au lieu de le déplacer (move) dans le premier tour.
• À chaque tour de boucle, job_rx.recv() crée un nouveau future, et celui qui perd est libéré. Ce n’est pas grave, car recv() est cancel-safe (sûr face à l’annulation) : le libérer ne fait jamais perdre de message.
• Si le pattern ne correspond pas (ici, recv() renvoie None parce que tous les émetteurs ont disparu), select! désactive cette branche et continue d’attendre les autres.
• Si plusieurs branches sont prêtes au même moment, select! en choisit une au hasard, pour qu’aucune branche ne monopolise la boucle. Écrivez biased; en première ligne dans select! pour vérifier les branches de haut en bas à la place.

La cancel safety, en un paragraphe. Comme la branche perdante est libérée, tout travail qu’elle avait à moitié fait est jeté. Pour recv() sur un canal ou pour sleep(), c’est sans danger : rien n’avait encore été pris. Mais imaginez un future qui lit 100 octets depuis un socket réseau et en a déjà lu 60 au moment où il est libéré. Ces 60 octets sont perdus. Un tel future n’est pas cancel-safe. La documentation de tokio indique pour chaque méthode si elle est cancel-safe ou non. Quand vous utilisez select! dans une boucle, vérifiez les méthodes que vous appelez, ou créez le future une seule fois en dehors de la boucle et continuez à poller ce même future (vous verrez comment dans la section sur Pin).

Timeouts et minuteurs

tokio::time::timeout(duration, future) enveloppe n’importe quel future. Si le future termine à temps, vous obtenez Ok(value). Si le temps s’écoule avant, le future est libéré (annulé, exactement comme une branche perdante de select!) et vous obtenez 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 [("serveur rapide", 20), ("serveur lent", 2000)] {
        match timeout(limit, fetch_price(delay)).await {
            Ok(price) => println!("{server} : le prix est {price}"),
            Err(e) => println!("{server} : {e}"),
        }
    }
}

Sortie :

serveur rapide : le prix est 42
serveur lent : deadline has elapsed

▶ Essayez par vous-même

« deadline has elapsed » (« le délai est écoulé ») est le message que tokio lui-même affiche pour l’erreur Elapsed : il reste donc en anglais.

Mettez un timeout sur tout ce qui parle au monde extérieur : appels HTTP, requêtes de base de données, lectures sur un socket. Un serveur qui ne répond jamais ne devrait pas bloquer votre programme pour toujours.

sleep ou interval ? sleep attend une fois. interval vous donne un rythme régulier, comme un métronome. Vous appelez .tick().await dans une boucle, et chaque tick arrive à l’heure prévue. Le premier tick se déclenche immédiatement :

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; // le premier tick se termine immédiatement
        println!("battement {beat}");
    }
}

Sortie :

battement 1
battement 2
battement 3

▶ Essayez par vous-même

Pourquoi ne pas simplement faire un sleep de 50ms dans une boucle ? Parce qu’alors chaque tour dure « votre travail + 50ms », et le planning dérive petit à petit. Un interval mesure à partir du début prévu de chaque tick : si votre travail prend 10ms, il n’attend que les 40ms restantes. Utilisez interval pour les heartbeats (signaux de vie), le polling et les tâches du type « sauvegarder chaque minute ». Utilisez sleep pour une pause unique.

Streams : des itérateurs async

Un itérateur vous donne des valeurs une par une avec next(). Un stream (flux) en est la version async : next().await vous donne la valeur suivante quand elle est prête, et None quand il n’y en a plus. Pensez à un fil de cours de la bourse, à des lignes qui arrivent d’une base de données ou aux messages d’un serveur de chat.

En fait, vous avez déjà utilisé quelque chose qui ressemble à un stream : le récepteur d’un canal tokio. rx.recv().await attend le message suivant et renvoie None quand tous les émetteurs ont disparu :

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; // une mesure toutes les 10ms
        }
        // tx est libéré ici, ce qui ferme le canal
    });

    while let Some(temp) = rx.recv().await {
        println!("température : {temp} °C");
    }
    println!("capteur arrêté");
}

Sortie :

température : 21 °C
température : 22 °C
température : 24 °C
température : 23 °C
capteur arrêté

▶ Essayez par vous-même

La bibliothèque standard n’a pas encore de trait Stream (il est toujours en cours de conception). L’écosystème utilise celui de la crate futures, et la crate d’accompagnement de tokio, tokio-stream, ajoute un trait StreamExt avec les méthodes que vous connaissez déjà des itérateurs : next(), map(), filter(), take() et d’autres. Ajoutez-la à votre projet :

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

Ce programme affiche 200, 204 et 196. Il se lit exactement comme du code avec des itérateurs, à part le .await sur next(). Pour transformer le récepteur d’un canal tokio en vrai stream, enveloppez-le : tokio_stream::wrappers::ReceiverStream::new(rx). (La crate futures a son propre StreamExt, sur la même idée. Choisissez-en un par fichier, car les deux traits ont une méthode next() et ils entrent en conflit.)

Pin, en termes simples

Tôt ou tard, vous verrez Pin dans un type ou dans un message d’erreur. Voici l’idée, sans la théorie pointue.

Quand Rust compile une async fn, il la transforme en une struct cachée, une machine à états. Chaque variable locale encore nécessaire après un .await devient un champ de cette struct. Imaginez maintenant qu’une variable locale emprunte une autre variable, comme let name = String::from("Ana"); let r = &name; avec un .await entre les deux. La struct contient alors un pointeur vers son propre champ. Si vous déplaciez la struct ailleurs en mémoire, ce pointeur viserait l’ancien emplacement, désormais vide.

Pin est la solution. Une valeur épinglée (pinned) est une promesse : « cette valeur ne bougera plus en mémoire. » Une fois qu’un future est épinglé, on peut le poller sans danger. La plupart du temps, vous ne vous en rendez même pas compte, car .await épingle le future pour vous.

Vous rencontrerez Pin dans trois situations :

1. Poller le même future plusieurs fois, par exemple une échéance dans une boucle select!.
2. Stocker des futures dans un Vec ou dans un champ de struct, en général sous la forme Pin<Box<dyn Future<Output = T>>>.
3. La récursion dans une async fn.

Pour la première, voici ce qui se passe si vous essayez de réutiliser un sleep dans une boucle sans l’épingler :

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`

Le compilateur vous dit que Sleep ne peut pas être « désépinglé » (unpinned), et il vous donne même la solution : la macro pin!, ou Box::pin. std::pin::pin! (ou tokio::pin!) épingle le future sur place, sur la pile. Désormais, la même échéance est partagée par tous les tours de la boucle :

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; // puis la boutique devient calme
    });

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

    loop {
        tokio::select! {
            _ = &mut deadline => {
                println!("c'est l'heure de fermer");
                break;
            }
            Some(order) = rx.recv() => println!("commande n°{order} reçue"),
        }
    }
}

Sortie :

commande n°1 reçue
commande n°2 reçue
commande n°3 reçue
c'est l'heure de fermer

▶ Essayez par vous-même

Si nous écrivions sleep(...) à l’intérieur du select!, chaque nouvelle commande lancerait un nouveau minuteur de 300ms, et une boutique très fréquentée ne fermerait jamais. Épingler une seule échéance en dehors de la boucle règle le problème.

Pour la récursion, le compilateur doit connaître la taille du future, et un future qui se contient lui-même serait infiniment grand. Appeler countdown(n - 1).await à l’intérieur de async fn countdown donne :

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

Autrement dit : la récursion dans une async fn exige une indirection, comme Box::pin, pour éviter un future de taille infinie. Box::pin déplace le future intérieur sur le tas (heap) et l’y épingle. Le future extérieur ne contient plus qu’un pointeur, dont la taille est fixe :

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

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

Sortie :

3...
2...
1...
décollage !

▶ Essayez par vous-même

Honnêtement, c’est tout ce dont la plupart des gens ont besoin : pin! pour un future que vous pollez plusieurs fois, Box::pin pour stocker des futures ou pour la récursion. Si vous voulez l’histoire complète, le chapitre du Rust Book dans la liste « Pour aller plus loin » l’explique pas à pas.

async fn dans les traits et closures async

Depuis Rust 1.75, vous pouvez écrire async fn directement dans un trait. Depuis Rust 1.85, vous avez aussi les closures async, écrites async |x| ..., avec les traits correspondants AsyncFn, AsyncFnMut et AsyncFnOnce (les versions async des traits de closures). Voici les deux dans un même exemple :

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; // imaginons qu'on parle à un serveur de mail
        format!("e-mail envoyé à {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;
}

Sortie :

e-mail envoyé à ana
e-mail envoyé à ben

▶ Essayez par vous-même

Il y a un piège avec async fn dans les traits. Le trait ne promet pas que le future renvoyé est Send, et tokio::spawn a besoin de futures Send (ils peuvent passer d’un thread à l’autre). Lancer un appel via un générique N: Notifier échoue donc :

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;

Le compilateur explique que le future ne peut pas être envoyé d’un thread à l’autre en toute sécurité, car il n’est pas Send. Sa suggestion est la solution habituelle : écrire la méthode du trait sous la forme fn ... -> impl Future<Output = String> + Send. Les implémentations peuvent continuer à utiliser une simple async fn. Notez aussi que les traits avec async fn ne peuvent pas encore servir de trait objects dyn Notifier. Si vous en avez besoin, la crate populaire async-trait met les futures dans des Box pour vous.

Erreurs courantes

Bloquer le runtime. std::thread::sleep, des calculs lourds ou de grosses lectures std::fs dans une tâche async figent tout le thread du worker, et toutes les autres tâches qui tournent dessus attendent. La page des bases montrait la version avec sleep. Pour un travail gourmand en CPU ou bloquant, confiez-le à tokio::task::spawn_blocking, qui l’exécute sur un pool de threads séparé, prévu pour ça :

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!("on répond toujours aux requêtes pendant le comptage...");

    let primes = handle.await.unwrap();
    println!("il y a {primes} nombres premiers en dessous de 50000");
}

Sortie :

on répond toujours aux requêtes pendant le comptage...
il y a 5133 nombres premiers en dessous de 50000

▶ Essayez par vous-même

Garder un guard de std::sync::Mutex à travers un .await. Ce code a l’air innocent :

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

Le compilateur vous protège. Il signale que le MutexGuard n’est pas Send et qu’il est encore utilisé à travers un .await. Pendant que la tâche dort sur .await, elle garderait le verrou, et toute autre tâche qui veut ce verrou bloquerait son thread entier. La solution consiste à libérer le guard avant le .await, en plaçant le verrouillage dans son propre bloc :

let count = {
    let mut count = visits.lock().unwrap();
    *count += 1;
    *count
}; // le guard est libéré ici
save_to_disk(count).await;

Si vous avez vraiment besoin de garder un verrou pendant une attente, utilisez tokio::sync::Mutex, dont le lock().await attend sans bloquer le thread. (Pour en savoir plus sur les verrous, voyez L’état partagé.)

Oublier .await. Les futures sont paresseux : save_file(); sans .await ne fait absolument rien. Le compilateur vous prévient avec « futures do nothing unless you .await or poll them » (les futures ne font rien tant que vous ne les attendez pas avec .await ou ne les pollez pas). Voyez Async/Await pour l’explication complète, et n’ignorez pas cet avertissement.

Lancer une tâche et ne jamais l’attendre. Quand main se termine, le runtime s’arrête et toutes les tâches inachevées sont libérées :

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

#[tokio::main]
async fn main() {
    tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        println!("rapport enregistré"); // jamais affiché !
    });

    println!("main a terminé");
}

Sortie :

main a terminé

▶ Essayez par vous-même

Le rapport n’est jamais enregistré. Si le travail compte, gardez le JoinHandle et attendez-le avec .await (ou mettez les tâches dans un JoinSet et videz-le) avant la fin de main.

Exercices

1. Comparer des prix avec une limite de temps

Quatre boutiques répondent à une demande de prix au bout de 20, 40, 300 et 60 millisecondes. Interrogez les quatre en même temps avec un JoinSet, mais n’accordez que 150ms à chaque boutique. Affichez les prix reçus (triés par nom de boutique), puis le nombre de boutiques qui n’ont pas répondu à temps. Indice : lancez timeout(limit, ask_shop(name, delay)) dans le set. Chaque tâche terminée vous donne alors un Ok ou un Err(Elapsed) : renvoyez le nom de la boutique avec ce résultat.

Voir la solution

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 prix inventé
}

#[tokio::main]
async fn main() {
    let shops = [("Le Petit Marché", 20), ("MégaMart", 40), ("Escargot & Cie", 300), ("Bric & Broc", 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} boutique(s) hors délai");
}

Sortie :

Bric & Broc : 160
Le Petit Marché : 120
MégaMart : 140
1 boutique(s) hors délai

▶ Essayez par vous-même

Toutes les boutiques attendent en même temps, donc la vérification complète prend environ 150ms, quelle que soit la lenteur d’Escargot & Cie. Quand son timeout se déclenche, son future est libéré : il ne peut pas continuer à tourner en arrière-plan.

2. Taille des dossiers avec la récursion async

Un Folder a un nom, une liste de tailles de fichiers en Ko et une liste de sous-dossiers. Écrivez async fn total_size(folder: &Folder) -> u64, qui additionne ses propres fichiers et la taille de chaque sous-dossier, en s’appelant elle-même pour chacun d’eux. Imaginez que chaque dossier ait d’abord besoin d’une courte « lecture disque » async, avec tokio::task::yield_now().await. Indice : l’appel récursif doit être entouré de Box::pin(...), sinon vous aurez l’erreur E0733.

Voir la solution

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

async fn total_size(folder: &Folder) -> u64 {
    tokio::task::yield_now().await; // imaginons qu'on lit le disque

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

#[tokio::main]
async fn main() {
    let home = Folder {
        name: "home",
        files_kb: vec![4],
        subfolders: vec![
            Folder { name: "photos", files_kb: vec![2048, 1024], subfolders: vec![] },
            Folder {
                name: "code",
                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} Ko");
}

Sortie :

photos : 3072 Ko
target : 500 Ko
code : 520 Ko
home : 3596 Ko
total : 3596 Ko

▶ Essayez par vous-même

Chaque dossier s’affiche après tous ses sous-dossiers, car il doit d’abord attendre leurs tailles. Box::pin place le future de chaque appel imbriqué sur le tas, si bien que le future extérieur garde une taille fixe. (Les noms des dossiers, comme home et target, sont ceux qu’on trouve vraiment sur un disque : on les garde tels quels.)

3. Vider la file, puis s’arrêter

Modifiez l’exemple du signal d’arrêt pour qu’il n’ait plus besoin du sleep(100ms) avant l’envoi du signal. Envoyez les trois travaux et le signal d’arrêt tout de suite, mais assurez-vous que le worker termine quand même tous les travaux en attente avant de s’arrêter. Indice : sans aide, select! choisit une branche au hasard quand les deux sont prêtes. Ajoutez biased; et placez la branche des travaux en premier.

Voir la solution

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 ["redimensionner la photo", "envoyer l'e-mail", "mettre à jour l'index de recherche"] {
        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!("en cours : {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("la file est vide, arrêt");
                    break;
                }
            }
        }
        done
    });

    let done = worker.await.unwrap();
    println!("le worker a terminé {done} travaux");
}

Sortie :

en cours : redimensionner la photo
en cours : envoyer l'e-mail
en cours : mettre à jour l'index de recherche
la file est vide, arrêt
le worker a terminé 3 travaux

▶ Essayez par vous-même

Avec biased;, select! essaie toujours la branche des travaux en premier : il ne regarde le signal d’arrêt que lorsque recv() n’a plus rien de prêt. Sans lui, le worker pourrait s’arrêter après un ou deux travaux, un bug qui n’apparaîtrait que de temps en temps.

Récapitulatif

• join! attend tous les futures, try_join! s’arrête à la première Err, et JoinSet exécute de nombreuses tâches lancées avec spawn et rend les résultats au fur et à mesure qu’elles terminent (triez-les si l’ordre compte).
• select! attend que le premier future termine. Les autres sont libérés (dropped) : c’est ainsi que fonctionne l’annulation en Rust async. Dans une boucle, utilisez des futures cancel-safe, et biased; quand l’ordre compte.
• timeout abandonne un future trop lent. interval donne un rythme régulier, sleep attend une seule fois.
• Un stream est un itérateur async. Le récepteur d’un canal fonctionne comme tel, et tokio-stream ajoute map, take et compagnie.
• Pin promet qu’un future ne bougera pas en mémoire. Utilisez pin! pour poller un même future plusieurs fois, et Box::pin pour les futures stockés et la récursion async.
• Ne bloquez pas le runtime (spawn_blocking), ne gardez pas un verrou std à travers un .await, et attendez les tâches que vous lancez.

Petit quiz

Petit quiz

1. Dans tokio::select!, qu'arrive-t-il aux branches qui ne terminent pas en premier ?

Elles continuent de tourner en arrière-plan Elles sont libérées : elles s'arrêtent et font leur nettoyage Elles s'exécutent à nouveau au tour de boucle suivant

2. Que renvoie tokio::time::timeout quand le temps est écoulé ?

Il panique Ok avec une valeur par défaut Err(Elapsed), et le future intérieur est libéré

3. Pourquoi faut-il Box::pin quand une async fn s'appelle elle-même ?

Un future qui se contient lui-même aurait une taille infinie La récursion est totalement interdite dans le code async Pour que la fonction s'exécute sur un autre thread

4. Vous devez faire 2 secondes de calcul intensif pour le CPU dans un programme tokio. Quel est le meilleur outil ?

std::thread::sleep autour du calcul tokio::select! Un JoinSet avec tokio::spawn tokio::task::spawn_blocking

Score : 0 / 4

Pour aller plus loin

• Select — the Tokio tutorial (en anglais) : select! en détail : comment fonctionnent les branches, ce que signifie l’annulation, et son utilisation dans une boucle.
• Streams — the Tokio tutorial (en anglais) : comment utiliser, adapter et construire des streams avec tokio-stream.
• A Closer Look at the Traits for Async — The Rust Book (en anglais) : un regard doux et progressif sur Future, Pin, Unpin et Stream.
• std::pin — la bibliothèque standard de Rust (en anglais) : l’explication officielle de l’épinglage, de Pin, Unpin et de la macro pin!.
• tokio::time sur docs.rs (en anglais) : sleep, timeout, interval et la gestion des ticks manqués.