Rust niveau 7 — Concurrence, async et API web

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc;
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

fn est_premier(n: u64) -> bool {
    if n < 2 { return false; }
    let mut d = 2;
    while d * d <= n {
        if n % d == 0 { return false; }
        d += 1;
    }
    true
}

fn pseudo_aleatoire(graine: u64) -> u64 {
    let t = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().subsec_nanos() as u64;
    (t ^ graine.wrapping_mul(2654435761)) % 40 + 10
}

fn main() {
    // 1. Premiers : un fil, puis quatre
    const MAX: u64 = 200_000;
    let debut = Instant::now();
    let seul = (0..MAX).filter(|&n| est_premier(n)).count();
    println!("1 fil  : {seul} premiers en {:?}", debut.elapsed());

    let debut = Instant::now();
    let quart = MAX / 4;
    let total: usize = thread::scope(|s| {
        let poignees: Vec<_> = (0..4)
            .map(|i| s.spawn(move || (i * quart..(i + 1) * quart).filter(|&n| est_premier(n)).count()))
            .collect();
        poignees.into_iter().map(|p| p.join().unwrap()).sum()
    });
    println!("4 fils : {total} premiers en {:?}", debut.elapsed());

    // 2. Téléchargements simulés
    let (tx, rx) = mpsc::channel();
    for fil in 1..=4u64 {
        let tx = tx.clone();
        thread::spawn(move || {
            for fichier in 1..=5u64 {
                let ms = pseudo_aleatoire(fil * 10 + fichier);
                thread::sleep(Duration::from_millis(ms));
                tx.send((fil, fichier, ms)).unwrap();
            }
        });
    }
    drop(tx);
    let mut cumul = 0;
    for (fil, fichier, ms) in rx {
        println!("fil {fil} : fichier {fichier} en {ms} ms");
        cumul += ms;
    }
    println!("temps cumulé : {cumul} ms");

    // 3. Compteur atomique
    let debut = Instant::now();
    let compteur = Arc::new(AtomicU64::new(0));
    let poignees: Vec<_> = (0..8)
        .map(|_| {
            let c = Arc::clone(&compteur);
            thread::spawn(move || {
                for _ in 0..1000 {
                    c.fetch_add(1, Ordering::Relaxed);
                }
            })
        })
        .collect();
    for p in poignees { p.join().unwrap(); }
    println!("atomique : {} en {:?}", compteur.load(Ordering::Relaxed), debut.elapsed());
}

Explication

Le découpage en quatre intervalles donne des charges inégales (les grands nombres sont plus longs à tester), mais le gain reste net sur une machine à plusieurs cœurs. Le compteur atomique est plus rapide que le Mutex car il n’y a ni attente ni verrou : le processeur garantit que l’addition est indivisible.

// Cargo.toml : tokio = { version = "1", features = ["full"] }
use std::time::{Duration, Instant};
use tokio::net::TcpStream;
use tokio::time::timeout;

async fn verifier(hote: &str, port: u16) -> (String, bool, u128) {
    let debut = Instant::now();
    let cible = format!("{hote}:{port}");
    let ok = matches!(
        timeout(Duration::from_secs(2), TcpStream::connect(&cible)).await,
        Ok(Ok(_))
    );
    (cible, ok, debut.elapsed().as_millis())
}

#[tokio::main]
async fn main() {
    let cibles = [
        ("192.168.56.10", 22), ("192.168.56.10", 80), ("192.168.56.11", 3389),
        ("192.168.56.12", 445), ("192.168.56.20", 53), ("127.0.0.1", 3000),
    ];
    let debut = Instant::now();
    let taches: Vec<_> = cibles
        .iter()
        .map(|&(h, p)| tokio::spawn(async move { verifier(h, p).await }))
        .collect();

    let mut resultats = Vec::new();
    for t in taches {
        resultats.push(t.await.expect("tâche interrompue"));
    }
    resultats.sort_by_key(|r| r.2);
    for (cible, ok, ms) in &resultats {
        println!("{cible:<22} {:<9} {ms:>5} ms", if *ok { "ouvert" } else { "fermé" });
    }
    println!("total : {:?}", debut.elapsed());
}

Explication

timeout renvoie une erreur si la future n’a pas abouti dans le délai ; connect renvoie elle-même un Result. D’où le motif Ok(Ok(_)) : « pas de dépassement de délai, et connexion réussie ». Toutes les tâches démarrent immédiatement : le temps total est celui de la plus lente, borné à 2 secondes.

Pour aller plus loin

Ce programme est un petit scanner de ports. Utilisé sur un réseau qui n’est pas le tien, il constitue une tentative d’intrusion au regard de la loi.

Le corrigé complet dépasse la place d’un livret ; voici le squelette et les éléments décisifs. Les dépendances : axum = "0.8", tokio (full), serde (derive), serde_json, subtle, et pour les tests tower avec la fonctionnalité util.

#![forbid(unsafe_code)]
use axum::{
    extract::{Path, Query, Request, State},
    http::{header, StatusCode},
    middleware::{self, Next},
    response::{IntoResponse, Response},
    routing::get,
    Json, Router,
};
use serde::{Deserialize, Serialize};
use std::net::IpAddr;
use std::sync::Arc;
use tokio::sync::RwLock;

// ---------- types validés ----------
#[derive(Clone, Serialize, Deserialize, PartialEq)]
#[serde(try_from = "String", into = "String")]
struct NomPoste(String);

impl TryFrom<String> for NomPoste {
    type Error = String;
    fn try_from(s: String) -> Result<Self, String> {
        let ok = (3..=32).contains(&s.len())
            && s.chars().all(|c| c.is_ascii_uppercase() || c.is_ascii_digit() || c == '-');
        if ok { Ok(NomPoste(s)) } else { Err("nom : 3 à 32 caractères A-Z, 0-9 ou -".into()) }
    }
}
impl From<NomPoste> for String {
    fn from(n: NomPoste) -> String { n.0 }
}

#[derive(Clone, Serialize, Deserialize)]
struct Poste { id: u32, nom: NomPoste, salle: u16, ram_go: u32, ip: IpAddr }

#[derive(Deserialize)]
struct Saisie { nom: NomPoste, salle: u16, ram_go: u32, ip: IpAddr }

#[derive(Deserialize)]
struct Filtre { salle: Option<u16>, ram_max: Option<u32> }

// ---------- erreurs ----------
enum ErreurApi { Introuvable, Conflit, NonAutorise, Interne }

impl IntoResponse for ErreurApi {
    fn into_response(self) -> Response {
        let (code, msg) = match self {
            ErreurApi::Introuvable => (StatusCode::NOT_FOUND, "poste introuvable"),
            ErreurApi::Conflit => (StatusCode::CONFLICT, "ce nom est déjà utilisé"),
            ErreurApi::NonAutorise => (StatusCode::UNAUTHORIZED, "authentification requise"),
            ErreurApi::Interne => (StatusCode::INTERNAL_SERVER_ERROR, "erreur interne"),
        };
        (code, Json(serde_json::json!({ "erreur": msg }))).into_response()
    }
}

// ---------- état ----------
struct Etat { postes: RwLock<Vec<Poste>>, jeton: String, fichier: String }
type Partage = Arc<Etat>;

async fn sauvegarder(e: &Etat, postes: &[Poste]) -> Result<(), ErreurApi> {
    let json = serde_json::to_string_pretty(postes).map_err(|_| ErreurApi::Interne)?;
    tokio::fs::write(&e.fichier, json).await.map_err(|err| {
        eprintln!("écriture de {} impossible : {err}", e.fichier);   // détail au journal
        ErreurApi::Interne                                            // réponse sobre
    })
}

// ---------- authentification ----------
async fn exiger_jeton(State(e): State<Partage>, req: Request, next: Next) -> Result<Response, ErreurApi> {
    use subtle::ConstantTimeEq;
    let fourni = req.headers().get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .unwrap_or("");
    if fourni.as_bytes().ct_eq(e.jeton.as_bytes()).into() {
        Ok(next.run(req).await)
    } else {
        Err(ErreurApi::NonAutorise)
    }
}

// ---------- gestionnaires ----------
async fn lister(State(e): State<Partage>, Query(f): Query<Filtre>) -> Json<Vec<Poste>> {
    let postes = e.postes.read().await;
    Json(postes.iter()
        .filter(|p| f.salle.is_none_or(|s| p.salle == s))
        .filter(|p| f.ram_max.is_none_or(|r| p.ram_go <= r))
        .cloned().collect())
}

async fn lire(State(e): State<Partage>, Path(id): Path<u32>) -> Result<Json<Poste>, ErreurApi> {
    e.postes.read().await.iter().find(|p| p.id == id).cloned().map(Json).ok_or(ErreurApi::Introuvable)
}

async fn creer(State(e): State<Partage>, Json(s): Json<Saisie>) -> Result<(StatusCode, Json<Poste>), ErreurApi> {
    let mut postes = e.postes.write().await;               // contrôle ET ajout sous le même verrou
    if postes.iter().any(|p| p.nom == s.nom) {
        return Err(ErreurApi::Conflit);
    }
    let id = postes.iter().map(|p| p.id).max().unwrap_or(0) + 1;
    let p = Poste { id, nom: s.nom, salle: s.salle, ram_go: s.ram_go, ip: s.ip };
    postes.push(p.clone());
    sauvegarder(&e, &postes).await?;
    Ok((StatusCode::CREATED, Json(p)))
}

async fn supprimer(State(e): State<Partage>, Path(id): Path<u32>) -> Result<StatusCode, ErreurApi> {
    let mut postes = e.postes.write().await;
    let avant = postes.len();
    postes.retain(|p| p.id != id);
    if postes.len() == avant { return Err(ErreurApi::Introuvable); }
    sauvegarder(&e, &postes).await?;
    Ok(StatusCode::NO_CONTENT)
}

fn application(etat: Partage) -> Router {
    let lecture = Router::new()
        .route("/postes", get(lister))
        .route("/postes/{id}", get(lire));
    let ecriture = Router::new()
        .route("/postes", axum::routing::post(creer))
        .route("/postes/{id}", axum::routing::delete(supprimer))
        .route_layer(middleware::from_fn_with_state(etat.clone(), exiger_jeton));
    lecture.merge(ecriture).with_state(etat)
}

#[tokio::main]
async fn main() {
    let jeton = std::env::var("API_JETON").expect("la variable API_JETON doit être définie");
    assert!(jeton.len() >= 32, "API_JETON doit faire au moins 32 caractères");
    let fichier = String::from("inventaire.json");
    let postes = match tokio::fs::read_to_string(&fichier).await {
        Ok(t) => serde_json::from_str(&t).expect("inventaire.json invalide"),
        Err(_) => Vec::new(),
    };
    let etat = Arc::new(Etat { postes: RwLock::new(postes), jeton, fichier });
    let ecoute = tokio::net::TcpListener::bind("127.0.0.1:3000").await.unwrap();
    axum::serve(ecoute, application(etat)).await.unwrap();
}

Explication

#[serde(try_from = "String")] fait passer chaque nom reçu par TryFrom : un nom invalide est refusé pendant la désérialisation, avec un code 422, sans une ligne de validation dans les gestionnaires. Les routes d’écriture sont regroupées sous une couche (route_layer) qui exige le jeton : impossible d’oublier la vérification sur une nouvelle route de ce groupe. La comparaison du jeton se fait en temps constant (subtle) pour ne pas révéler, par le temps de réponse, combien de caractères sont justes. Le contrôle d’unicité et l’ajout se font sous le même verrou d’écriture : deux requêtes simultanées ne peuvent pas créer deux fois le même nom. Le verrou est ici celui de Tokio (tokio::sync::RwLock), qui peut être conservé pendant un .await, contrairement à celui de la bibliothèque standard.

is_none_or demande Rust 1.82 ou plus récent ; sur une version plus ancienne, écris f.salle.map_or(true, |s| p.salle == s).

Pour le test des routes, tower::ServiceExt::oneshot envoie une requête à l’application sans ouvrir de port :

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::Body;
    use axum::http::Request;
    use tower::ServiceExt;

    fn etat_test() -> Partage {
        Arc::new(Etat { postes: RwLock::new(Vec::new()), jeton: "x".repeat(32), fichier: "test.json".into() })
    }

    #[tokio::test]
    async fn creation_sans_jeton_refusee() {
        let req = Request::post("/postes")
            .header("content-type", "application/json")
            .body(Body::from(r#"{"nom":"PC-01","salle":207,"ram_go":8,"ip":"10.0.0.1"}"#))
            .unwrap();
        let rep = application(etat_test()).oneshot(req).await.unwrap();
        assert_eq!(rep.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn poste_inconnu() {
        let req = Request::get("/postes/42").body(Body::empty()).unwrap();
        let rep = application(etat_test()).oneshot(req).await.unwrap();
        assert_eq!(rep.status(), StatusCode::NOT_FOUND);
    }

    #[test]
    fn nom_invalide_refuse() {
        assert!(NomPoste::try_from(String::from("pc 01")).is_err());
        assert!(NomPoste::try_from(String::from("PC-207-01")).is_ok());
    }
}

Pour aller plus loin

Les refus de désérialisation (422) renvoient pour l’instant le message texte par défaut d’Axum. Pour des erreurs toutes au format JSON, on intercepte ce rejet : la documentation d’Axum le décrit dans sa partie consacrée aux extracteurs.

Les routes PUT et POST /import suivent les mêmes modèles. Pour l’import, ajoute la bibliothèque du niveau 6 en dépendance locale : analyse-journal = { path = "../analyse-journal" }, et appelle analyse::analyser(corps.lines(), 5). C’est l’intérêt d’avoir séparé bibliothèque et exécutable.