Rust

La documentation officielle parle de « concurrence sans peur ». La formule est juste. Les règles d'ownership et d'emprunt du chapitre 3 empêchent, à la compilation, deux threads de modifier la même donnée sans synchronisation. La course de données, le bug le plus sournois de la programmation concurrente, n'est pas difficile à éviter en Rust : elle est impossible à compiler. Ce chapitre couvre les threads du système, les canaux, l'état partagé, puis l'asynchrone avec Tokio.

Lancer un thread

use std::thread;
use std::time::Duration;
 
fn main() {
    let poignee = thread::spawn(|| {
        for i in 1..=3 {
            println!("thread secondaire : {i}");
            thread::sleep(Duration::from_millis(10));
        }
        42
    });
 
    println!("thread principal");
    let resultat = poignee.join().unwrap(); // attend la fin et récupère la valeur
    println!("résultat : {resultat}");
}

join attend la fin du thread et renvoie un Result : Err si le thread a paniqué. Sans join, le programme peut se terminer avant le thread secondaire.

move : donner les données au thread

use std::thread;
 
fn main() {
    let donnees = vec![1, 2, 3];
    let poignee = thread::spawn(move || {
        let somme: i32 = donnees.iter().sum();
        println!("somme : {somme}");
    });
    // println!("{donnees:?}"); // erreur : donnees a été déplacée dans le thread
    poignee.join().unwrap();
}

Sans move, le compilateur refuse : le thread pourrait vivre plus longtemps que main, et la référence vers donnees deviendrait invalide. move transfère la propriété au thread, et le problème disparaît.

Les threads à portée limitée

Quand les threads doivent seulement emprunter, thread::scope garantit qu'ils se terminent tous avant la fin du bloc. Plus besoin de move ni de copie :

use std::thread;
 
fn main() {
    let nombres: Vec<u64> = (1..=1_000_000).collect();
    let (gauche, droite) = nombres.split_at(nombres.len() / 2);
 
    let total = thread::scope(|s| {
        let a = s.spawn(|| gauche.iter().sum::<u64>());
        let b = s.spawn(|| droite.iter().sum::<u64>());
        a.join().unwrap() + b.join().unwrap()
    });
 
    println!("{total}"); // 500000500000
}

Les canaux : communiquer en s'envoyant des messages

Un canal relie des producteurs à un consommateur. Ce qui est envoyé change de propriétaire : l'expéditeur ne peut plus y toucher.

use std::sync::mpsc;
use std::thread;
 
fn main() {
    let (tx, rx) = mpsc::channel();
 
    for id in 0..3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let resultat = id * 10;
            tx.send(format!("ouvrier {id} : {resultat}")).unwrap();
        });
    }
    drop(tx); // ferme l'émetteur d'origine, sinon la boucle ne finit jamais
 
    for message in rx {
        println!("{message}");
    }
}

mpsc veut dire « plusieurs producteurs, un seul consommateur ». La boucle for message in rx s'arrête quand tous les émetteurs ont disparu. D'où le drop(tx) : oubliez-le, et le programme reste bloqué.

L'état partagé : Arc et Mutex

Parfois, plusieurs threads doivent modifier la même donnée. Il faut deux choses : plusieurs propriétaires (Arc, la version thread-safe de Rc vue au chapitre 7) et un verrou (Mutex).

use std::sync::{Arc, Mutex};
use std::thread;
 
fn main() {
    let compteur = Arc::new(Mutex::new(0));
    let mut poignees = Vec::new();
 
    for _ in 0..8 {
        let compteur = Arc::clone(&compteur);
        poignees.push(thread::spawn(move || {
            for _ in 0..1000 {
                *compteur.lock().unwrap() += 1;
            }
        }));
    }
 
    for p in poignees {
        p.join().unwrap();
    }
    println!("{}", *compteur.lock().unwrap()); // 8000, toujours
}

lock() renvoie un garde qui donne accès à la donnée et libère le verrou automatiquement quand il sort de portée. Impossible d'oublier de déverrouiller, et impossible d'accéder à la donnée sans verrouiller : elle est à l'intérieur du Mutex.

Essayez de remplacer Arc par Rc : le compilateur refuse, car Rc n'implémente pas le trait Send. Deux traits marqueurs font tout le travail :

Le compilateur les déduit automatiquement pour vos types. Vous n'avez presque jamais à y penser, sauf quand une erreur les mentionne.

Quand les lectures sont bien plus nombreuses que les écritures, RwLock autorise plusieurs lecteurs simultanés. Pour un simple compteur, les types atomiques (AtomicUsize) évitent le verrou.

Paralléliser sans effort : Rayon

Pour un calcul sur une grosse collection, la crate rayon transforme un itérateur séquentiel en itérateur parallèle en changeant un mot :

cargo add rayon

use rayon::prelude::*;
 
fn est_premier(n: u64) -> bool {
    n >= 2 && (2..).take_while(|d| d * d <= n).all(|d| n % d != 0)
}
 
fn main() {
    let total = (1..2_000_000u64).into_par_iter().filter(|&n| est_premier(n)).count();
    println!("{total} nombres premiers");
}

into_iter devient into_par_iter, et le travail est réparti sur tous les cœurs. Si votre programme fait du calcul pur sur des données en mémoire, c'est souvent tout ce dont vous avez besoin.

L'asynchrone

Les threads conviennent au calcul. Pour attendre beaucoup de choses à la fois (des milliers de connexions réseau, des requêtes HTTP, des fichiers), un thread par attente coûte trop cher. L'asynchrone permet à quelques threads de gérer des milliers de tâches : une tâche qui attend le réseau cède sa place aux autres.

En Rust, une fonction async renvoie un future, une valeur qui représente un calcul pas encore fait. .await attend son résultat sans bloquer le thread. Contrairement à JavaScript, le langage ne fournit pas de moteur pour exécuter ces futures : on choisit un runtime, et le standard de fait est Tokio.

cargo add tokio --features full

use std::time::Duration;
use tokio::time::sleep;
 
async fn telecharger(id: u32, ms: u64) -> String {
    sleep(Duration::from_millis(ms)).await; // simule une attente réseau
    format!("fichier {id} ({ms} ms)")
}
 
#[tokio::main]
async fn main() {
    // séquentiel : 300 + 200 + 100 = 600 ms
    let a = telecharger(1, 300).await;
    println!("{a}");
 
    // concurrent : environ 300 ms au total
    let (x, y, z) = tokio::join!(telecharger(2, 300), telecharger(3, 200), telecharger(4, 100));
    println!("{x}, {y}, {z}");
}

#[tokio::main] démarre le runtime et exécute main dedans. tokio::join! lance plusieurs futures en même temps et attend qu'ils finissent tous.

Les tâches

tokio::spawn lance une tâche indépendante, l'équivalent asynchrone d'un thread, mais qui ne coûte que quelques centaines d'octets :

use std::time::Duration;
 
#[tokio::main]
async fn main() {
    let mut taches = Vec::new();
    for id in 0..10_000 {
        taches.push(tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(100)).await;
            id * 2
        }));
    }
 
    let mut somme = 0;
    for t in taches {
        somme += t.await.unwrap();
    }
    println!("{somme}"); // 99990000, en à peine plus de 100 ms
}

Dix mille threads système seraient lourds ; dix mille tâches Tokio ne sont rien.

Les pièges de l'async

Trois erreurs reviennent chez tous les débutants :

1. Bloquer le runtime. std::thread::sleep ou un gros calcul dans une fonction async immobilise un thread du runtime, et toutes les tâches qui y tournent. Utilisez tokio::time::sleep, et envoyez les calculs lourds dans tokio::task::spawn_blocking.
2. Oublier .await. Un future non attendu ne fait rien. Le compilateur avertit, lisez l'avertissement.
3. Garder un std::sync::Mutex verrouillé pendant un .await. Relâchez le verrou avant d'attendre, ou utilisez tokio::sync::Mutex.

Threads ou async ?

L'async a un coût en complexité. Ne l'adoptez pas par réflexe : un outil en ligne de commande qui traite un fichier n'en a pas besoin. Un serveur qui gère des connexions persistantes, comme ceux décrits dans l'article sur WebSocket, SSE et polling, en a besoin.

Exercice

Calculez en parallèle, avec quatre threads et un canal, la somme des carrés de 1 à 10 millions. Chaque thread traite un quart de l'intervalle et envoie son résultat partiel.

use std::sync::mpsc;
use std::thread;
 
fn main() {
    const N: u64 = 10_000_000;
    const THREADS: u64 = 4;
    let (tx, rx) = mpsc::channel();
 
    for t in 0..THREADS {
        let tx = tx.clone();
        thread::spawn(move || {
            let debut = t * N / THREADS + 1;
            let fin = (t + 1) * N / THREADS;
            let partiel: u128 = (debut..=fin).map(|n| (n as u128) * (n as u128)).sum();
            tx.send(partiel).unwrap();
        });
    }
    drop(tx);
 
    let total: u128 = rx.iter().sum();
    println!("{total}"); // 333333383333335000000
}

u128 parce que le résultat dépasse la capacité d'un u64. Vérifiez avec la formule n(n+1)(2n+1)/6.

Threads, canaux, Arc<Mutex<T>>, Rayon et Tokio : Rust donne tous les outils de la concurrence, et son compilateur refuse les combinaisons dangereuses. Il est temps d'assembler tout le cours dans un vrai programme. Le dernier chapitre construit un outil en ligne de commande complet.