Async Rust avec Tokio : async/await, task et channel

1. Vue d’ensemble

Rust dispose de la syntaxe async/await mais n’inclut pas de runtime pour exécuter les Future. Tokio est le runtime le plus populaire pour cela. Il fournit un scheduler, des timers, de l’I/O asynchrone et des outils de synchronisation.

Cet article part de votre première fonction async jusqu’aux opérations courantes : spawn task, exécuter en parallèle avec join!, définir des timeouts avec select!, partager l’état entre les tasks, envoyer des données via des channels et gérer le code blocking. La fin parle de comment écrire des tests et des erreurs courantes.

2. Installation

Vous ajoutez Tokio à Cargo.toml :

[dependencies]
tokio = { version = "1", features = ["full"] }

[dev-dependencies]
tokio = { version = "1", features = ["full", "test-util"] }

La feature full active tous les modules : runtime multi-threadé, time, sync, net, macros… C’est pratique pour apprendre. Dans un vrai projet, vous devriez n’activer que les features que vous utilisez vraiment pour réduire le temps de compilation, par exemple ["rt-multi-thread", "macros", "time", "sync"]. La feature test-util permet de contrôler le temps dans les tests. Nous l’utiliserons à la section 8.

3. Votre première fonction async

3.1. #[tokio::main] et .await

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

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100)).await; // simulation d'un appel à la base de données
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let order = fetch_order(42).await;
    println!("{order}");
}

Explications :

• async fn fetch_order ne retourne pas une String immédiatement. Elle retourne un impl Future<Output = String>.
• tokio::time::sleep est la version async de sleep. Pendant qu’elle attend, le thread est redonné au runtime pour exécuter autre chose.
• #[tokio::main] est une macro. Elle crée un runtime multi-threadé puis exécute main à l’intérieur de ce runtime avec block_on.

Order #42

3.2. Les Future sont lazy

C’est la grande différence avec les Promise en JavaScript : une future en Rust ne fait rien du tout jusqu’à ce qu’elle soit .await-ée ou poll-ée.

async fn load_customer(id: u32) -> String {
    println!("Chargement du client {id}");
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let fut = load_customer(7); // aucune ligne de la fonction ne s'est exécutée
    println!("Avant le await");
    let customer = fut.await;
    println!("{customer}");
}

Avant le await
Chargement du client 7
Customer 7

La ligne “Chargement du client 7” apparaît seulement après avoir appelé .await. Si vous oubliez .await, la fonction ne s’exécutera jamais. Le compilateur vous avertira unused implementer of Future that must be used, ne ignorez pas cet avertissement.

4. Spawn task

.await exécute les futures séquentiellement. Pour qu’elles s’exécutent simultanément, vous utilisez tokio::spawn pour créer une task. Une task est un “green thread” léger géré par Tokio.

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

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100 * id as u64)).await;
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let mut handles = Vec::new();

    for id in 1..=3 {
        handles.push(tokio::spawn(async move { fetch_order(id).await }));
    }

    for handle in handles {
        let order = handle.await.expect("la task a paniqué");
        println!("{order}");
    }
    println!("Tổng thời gian: ~{} ms", start.elapsed().as_millis());
}

Order #1
Order #2
Order #3
Tổng thời gian: ~300 ms

Si c’était exécuté séquentiellement, le temps total serait 100 + 200 + 300 = 600 ms. Avec spawn, les trois tasks s’exécutent simultanément donc le total est seulement environ 300 ms.

Quelques points à retenir :

• tokio::spawn retourne un JoinHandle<T>. Quand vous .await le handle, vous obtenez un Result<T, JoinError>. C’est Err si la task a paniqué ou a été annulée.
• La future passée doit satisfaire Send + 'static, car la task peut s’exécuter sur un autre thread et vivre plus longtemps que la portée actuelle. C’est pourquoi on a besoin d’async move.
• La task commence à s’exécuter dès qu’elle est spawned, sans avoir besoin d’attendre le .await du handle.

Note: Dropper un JoinHandle ne tue pas la task. La task continue de s’exécuter en mode détaché. Pour l’annuler, vous appelez handle.abort().

Quand le nombre de tasks varie, vous pouvez utiliser tokio::task::JoinSet. Elle permet d’obtenir les résultats dans l’ordre d’achèvement des tasks avec join_set.join_next().await.

5. Exécuter simultanément dans la même task : join!, select!, timeout

5.1. join! et try_join!

Vous n’avez pas toujours besoin de spawn. tokio::join! poll plusieurs futures simultanément dans la même task et attend que toutes se terminent :

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

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

async fn fetch_customer(id: u32) -> String {
    sleep(Duration::from_millis(150)).await;
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let (order, customer) = tokio::join!(fetch_order(1), fetch_customer(7));
    println!("{order} - {customer} ({} ms)", start.elapsed().as_millis());
}

Order #1 - Customer 7 (200 ms)

Le temps total égale la future la plus lente, environ 200 ms. Les futures n’ont pas besoin d’être 'static, donc elles peuvent emprunter des variables locales.

Quand les fonctions retournent Result, vous utilisez tokio::try_join!. Cette macro s’arrête à la première erreur et drope les futures restantes :

let (order, customer) = tokio::try_join!(fetch_order_checked(1), fetch_customer_checked(7))?;

Ici fetch_order_checked et fetch_customer_checked sont des fonctions async qui retournent Result<String, AppError>.

5.2. timeout

use tokio::time::{timeout, Duration};

match timeout(Duration::from_millis(50), fetch_order(1)).await {
    Ok(order) => println!("{order}"),
    Err(_) => println!("Quá thời gian chờ"),
}

Quá thời gian chờ

fetch_order prend 200 ms, ce qui dépasse la limite de 50 ms. Alors timeout retourne Err(Elapsed), et la future à l’intérieur est droppée, c’est-à-dire annulée.

5.3. select!

select! attend que la première branche se termine, puis annule les autres branches :

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    tokio::spawn(async move {
        sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(());
    });

    tokio::select! {
        order = fetch_order(1) => println!("Xử lý xong {order}"),
        _ = shutdown_rx => println!("Nhận tín hiệu shutdown, dừng xử lý"),
    }
}

Nhận tín hiệu shutdown, dừng xử lý

Le signal d’arrêt arrive après 100 ms, plus tôt que fetch_order (200 ms), donc la deuxième branche gagne.

Note: L’annulation en Rust signifie que la future est droppée au point .await le plus proche. Si la branche annulée était en train de faire une opération non atomique, par exemple qu’elle avait lu la moitié d’un message du socket, les données pourraient être perdues. Vous devriez vérifier la section “cancel safety” dans la documentation de chaque fonction avant de l’utiliser dans select! dans une boucle.

6. Partager des données entre les tasks

6.1. Arc<Mutex<T>>

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

#[tokio::main]
async fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("laptop", 10u32)])));
    let mut handles = Vec::new();

    for _ in 0..5 {
        let inventory = Arc::clone(&inventory);
        handles.push(tokio::spawn(async move {
            let mut stock = inventory.lock().unwrap();
            *stock.get_mut("laptop").unwrap() -= 1;
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    println!("Còn lại: {}", inventory.lock().unwrap()["laptop"]);
}

Còn lại: 5

Ici nous utilisons std::sync::Mutex, pas tokio::sync::Mutex. C’est le choix recommandé quand la section critique est courte et n’a pas de .await à l’intérieur, car le mutex standard est plus rapide.

Vous n’avez besoin de tokio::sync::Mutex que si vous devez garder le lock à travers un point .await, par exemple quand vous gardez une connexion partagée pendant qu’elle attend I/O.

6.2. Channel mpsc

Au lieu de partager la mémoire, vous pouvez passer des messages entre les tasks via un channel :

use tokio::sync::mpsc;

#[derive(Debug)]
struct Order {
    id: u32,
    amount: u64,
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<Order>(32);

    for id in 1..=3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(Order { id, amount: id as u64 * 100 }).await.unwrap();
        });
    }
    drop(tx); // fermer le sender original pour que la boucle ci-dessous puisse se terminer

    let mut total = 0;
    while let Some(order) = rx.recv().await {
        println!("Nhận {order:?}");
        total += order.amount;
    }
    println!("Tổng: {total}");
}

Nhận Order { id: 1, amount: 100 }
Nhận Order { id: 3, amount: 300 }
Nhận Order { id: 2, amount: 200 }
Tổng: 600

L’ordre de réception peut être différent à chaque exécution. Le paramètre 32 est la capacité du buffer. Quand le buffer est plein, send attend, créant un mécanisme de back-pressure naturel. recv() retourne None quand tous les senders ont été droppés. Si vous oubliez drop(tx), le programme restera bloqué dans la boucle while.

En plus de mpsc, Tokio a aussi :

• oneshot : envoyer une valeur unique.
• broadcast : plusieurs consumers, chaque consumer reçoit tous les messages.
• watch : le consumer se soucie seulement de la valeur la plus récente.

7. Gérer le code blocking

Tokio utilise le cooperative scheduling. Les tasks ne cèdent le thread que aux points .await. Une opération CPU intensive ou un I/O synchrone occupera complètement un worker thread et affamera les autres tasks.

fn hash_password(password: &str) -> String {
    // simulation d'une opération CPU intensive, par exemple bcrypt/argon2
    std::thread::sleep(std::time::Duration::from_millis(500));
    format!("hashed:{password}")
}

#[tokio::main]
async fn main() {
    let password = String::from("s3cret");
    let hashed = tokio::task::spawn_blocking(move || hash_password(&password))
        .await
        .expect("task blocking bị panic");
    println!("{hashed}");
}

hashed:s3cret

spawn_blocking pousse la closure dans un pool de threads séparé dédié au blocking. Cela permet aux worker threads du runtime de rester libres pour gérer d’autres requêtes.

Note: N’appelez pas std::thread::sleep, std::fs::read, ou un client HTTP synchrone directement dans une fonction async. Utilisez plutôt les versions async (tokio::time::sleep, tokio::fs) ou enveloppez-les dans spawn_blocking. Pour du calcul parallèle à grande échelle, Rayon est généralement plus approprié.

8. Tester le code async

La macro #[tokio::test] crée un runtime pour chaque test. Par défaut, c’est un runtime mono-thread.

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

pub async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn fetch_order_returns_formatted_id() {
        assert_eq!(fetch_order(42).await, "Order #42");
    }

    #[tokio::test(start_paused = true)]
    async fn timeout_fires_when_order_service_is_slow() {
        let result = timeout(Duration::from_millis(50), fetch_order(1)).await;
        assert!(result.is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn order_arrives_within_generous_timeout() {
        let result = timeout(Duration::from_secs(1), fetch_order(1)).await;
        assert_eq!(result.unwrap(), "Order #1");
    }
}

cargo test

running 3 tests
test tests::fetch_order_returns_formatted_id ... ok
test tests::timeout_fires_when_order_service_is_slow ... ok
test tests::order_arrives_within_generous_timeout ... ok

start_paused = true (nécessite la feature test-util) gèle l’horloge de Tokio. Quand toutes les tasks attendent un timer, le runtime saute automatiquement l’heure jusqu’à la prochaine étape. Cela permet aux tests de timeout de s’exécuter presque instantanément et de donner des résultats stables, indépendamment de la vitesse de la machine CI.

Si vous devez vérifier le comportement sur plusieurs threads, vous utilisez #[tokio::test(flavor = "multi_thread", worker_threads = 2)].

9. Erreurs courantes

Garder un std::sync::MutexGuard à travers un .await. Ce guard n’est pas Send, donc la future ne l’est pas non plus. Alors tokio::spawn génère une erreur de compilation future cannot be sent between threads safely. La correction consiste à réduire la portée du guard avec un bloc { ... } avant le .await, ou à passer à tokio::sync::Mutex.

Créer un runtime imbriqué. Appeler Runtime::new().block_on(...) à l’intérieur du code s’exécutant déjà sur Tokio causera une panique avec le message Cannot start a runtime from within a runtime. Dans un contexte async, il suffit de .await.

Utiliser async quand ce n’est pas nécessaire. Un outil CLI qui lit quelques fichiers, ou un job purement computationnel, ne bénéficie pas de Tokio. Vous ne payez que du temps de compilation supplémentaire et de la complexité. Async brille quand il y a beaucoup d’opérations I/O simultanées : serveur web, crawler, proxy, client appelant plusieurs services.

Spawn illimité. Les tasks sont bon marché mais pas gratuites. Spawner une task pour chaque élément d’une liste d’un million d’éléments peut épuiser les ressources en aval, comme le pool de connexions de la base de données. Limitez la concurrence avec tokio::sync::Semaphore ou utilisez buffer_unordered du crate futures.

Ignorer JoinError. Si vous .await le handle avec let _ =, la panique dans la task sera avalée. Vous devriez enregistrer l’erreur ou la propager.

10. Conclusion

Cet article a couvert les éléments essentiels du Rust async avec Tokio :

• Les futures sont lazy et ne s’exécutent que quand elles sont .await-ées.
• tokio::spawn crée des tasks qui s’exécutent simultanément. La future passée doit être Send + 'static.
• join! et try_join! exécutent plusieurs futures dans la même task. select! et timeout traitent les conditions de course et l’annulation.
• Les données partagées peuvent être partagées via Arc<Mutex<T>>, ou transmises via des channels comme mpsc et oneshot.
• Le code blocking doit être mis dans spawn_blocking.
• #[tokio::test(start_paused = true)] rend les tests de timing rapides et stables.

Une fois que vous maîtrisez ces éléments, vous pouvez continuer avec l’I/O réseau via tokio::net, ou construire des services web sur Tokio avec des frameworks comme Axum.