Concurrence & état partagé

// Condition de course logique : compile, mais faux
use std::sync::Mutex;
let stock = Mutex::new(1);

// Deux threads font ceci en même temps :
let disponible = *stock.lock().unwrap() > 0;   // verrou pris puis relâché
if disponible {
    *stock.lock().unwrap() -= 1;               // l'autre thread a pu passer entre les deux : stock = -1
}

// Correct : lire et écrire sous le même verrou
let mut s = stock.lock().unwrap();
if *s > 0 {
    *s -= 1;
}

use std::thread;

let h = thread::spawn(|| {
    (1..=10).sum::<u32>()
});

let somme = h.join().unwrap();       // attendre la fin ; Err si le thread a paniqué

let v = vec![1, 2, 3];
let h = thread::spawn(move || v.iter().sum::<i32>());   // v appartient au thread
let s = h.join().unwrap();

let handles: Vec<_> = (0..4)
    .map(|i| thread::spawn(move || i * i))
    .collect();

let resultats: Vec<i32> = handles.into_iter().map(|h| h.join().unwrap()).collect();

let donnees = vec![1, 2, 3, 4, 5, 6];
let (gauche, droite) = donnees.split_at(3);

let (a, b) = thread::scope(|s| {
    let ha = s.spawn(|| gauche.iter().sum::<i32>());   // emprunte, sans move ni Arc
    let hb = s.spawn(|| droite.iter().sum::<i32>());
    (ha.join().unwrap(), hb.join().unwrap())
});

Un thread lancé dans un scope ne peut pas survivre au bloc : il peut donc emprunter les variables locales. C'est souvent la façon la plus simple de paralléliser un calcul.

let h = thread::Builder::new()
    .name("nettoyage".into())
    .spawn(|| { /* ... */ })
    .unwrap();

Le nom apparaît dans les messages de panique et les outils de profilage.

use std::sync::Arc;
use std::thread;

let config = Arc::new(vec!["a", "b"]);
let mut handles = vec![];
for i in 0..3 {
    let c = Arc::clone(&config);         // copie du pointeur, compteur + 1
    handles.push(thread::spawn(move || println!("{i} : {:?}", c)));
}
for h in handles {
    h.join().unwrap();
}
println!("{}", Arc::strong_count(&config));   // 1 : les threads ont libéré leur copie

use std::sync::Mutex;

let m = Mutex::new(0);
{
    let mut garde = m.lock().unwrap();   // un seul thread à la fois ici
    *garde += 1;
}                                        // garde libérée : verrou relâché

lock() renvoie un MutexGuard : il se déréférence vers la valeur, et son Drop relâche le verrou. On ne peut pas oublier de déverrouiller.

use std::sync::{Arc, Mutex};
use std::thread;

let compteur = Arc::new(Mutex::new(0));
let handles: Vec<_> = (0..10)
    .map(|_| {
        let c = Arc::clone(&compteur);
        thread::spawn(move || *c.lock().unwrap() += 1)
    })
    .collect();
for h in handles {
    h.join().unwrap();
}
assert_eq!(*compteur.lock().unwrap(), 10);

// Mauvais : le verrou est tenu pendant tout le traitement
let liste = Mutex::new(vec![1, 2, 3]);
{
    let garde = liste.lock().unwrap();
    for x in garde.iter() {
        traitement_long(*x);             // les autres threads attendent
    }
}

// Mieux : copier, relâcher, traiter
let copie = liste.lock().unwrap().clone();   // verrou relâché à la fin de l'instruction
for x in copie {
    traitement_long(x);
}

fn traitement_long(_: i32) {}

match m.try_lock() {
    Ok(mut g) => g.push(2),
    Err(_) => println!("occupé : on réessaiera plus tard"),
}

let mut m = Mutex::new(5);
*m.get_mut().unwrap() += 1;        // &mut m prouve qu'aucun autre accès n'existe : pas de verrou
let valeur = m.into_inner().unwrap();   // récupérer la donnée et détruire le Mutex

use std::sync::{Arc, RwLock};

let config = Arc::new(RwLock::new(String::from("v1")));

let lecture = config.read().unwrap();      // plusieurs lecteurs en même temps
println!("{}", *lecture);
drop(lecture);

let mut ecriture = config.write().unwrap(); // un seul écrivain, aucun lecteur
*ecriture = String::from("v2");

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::thread;

let compteur = Arc::new(AtomicU64::new(0));
let handles: Vec<_> = (0..10)
    .map(|_| {
        let c = Arc::clone(&compteur);
        thread::spawn(move || {
            c.fetch_add(1, Ordering::Relaxed);
        })
    })
    .collect();
for h in handles {
    h.join().unwrap();
}
assert_eq!(compteur.load(Ordering::Relaxed), 10);

static ARRET: AtomicBool = AtomicBool::new(false);
ARRET.store(true, Ordering::Release);
if ARRET.load(Ordering::Acquire) { /* arrêter */ }

use std::sync::mpsc;
use std::thread;

let (tx, rx) = mpsc::channel();
for i in 0..3 {
    let tx = tx.clone();
    thread::spawn(move || tx.send(i * 10).unwrap());
}
drop(tx);                                   // fermer : sinon rx attend pour toujours

let total: i32 = rx.iter().sum();           // reçoit jusqu'à la fermeture

let (tx, rx) = mpsc::sync_channel::<String>(10);   // au-delà de 10, send bloque

// 1. Mémoire partagée : plusieurs threads modifient la même Vec sous un verrou
// 2. Messages : chaque thread envoie ses résultats ; un seul propriétaire assemble

Les messages évitent les verrous et les interblocages : une seule tâche possède la donnée. Pour l'asynchrone, tokio fournit ses propres canaux (mpsc, oneshot, broadcast, watch, voir async-tokio). La crate crossbeam-channel offre des canaux std plus riches (plusieurs récepteurs, select!).

use std::collections::HashMap;
use std::sync::LazyLock;

static CODES: LazyLock<HashMap<u16, &'static str>> = LazyLock::new(|| {
    HashMap::from([(404, "introuvable"), (500, "erreur interne")])
});

fn message(code: u16) -> &'static str {
    CODES.get(&code).copied().unwrap_or("inconnu")
}

• La closure s'exécute une seule fois, au premier accès, même si plusieurs threads y accèdent en même temps.
• Disponible depuis Rust 1.80 : remplace lazy_static! et once_cell::sync::Lazy.

// lazy_static (crate)
// lazy_static! { static ref CODES: HashMap<u16, &'static str> = { .. }; }

// once_cell (crate)
// static CODES: once_cell::sync::Lazy<HashMap<..>> = Lazy::new(|| ..);

// Aujourd'hui : std uniquement
// static CODES: std::sync::LazyLock<HashMap<..>> = LazyLock::new(|| ..);

use std::sync::OnceLock;

static URL_BASE: OnceLock<String> = OnceLock::new();

fn initialiser(url: String) {
    if URL_BASE.set(url).is_err() {
        panic!("URL_BASE déjà initialisée");
    }
}

fn url_base() -> &'static str {
    URL_BASE.get().map(String::as_str).expect("initialiser() doit être appelée au démarrage")
}

// Ou : calculée à la première demande, depuis une fonction
fn cache() -> &'static Vec<u32> {
    static C: OnceLock<Vec<u32>> = OnceLock::new();
    C.get_or_init(|| (1..=10).collect())
}

use std::sync::{Arc, Mutex};
use std::thread;

let a = Arc::new(Mutex::new(0));
let b = Arc::new(Mutex::new(0));

let (a1, b1) = (Arc::clone(&a), Arc::clone(&b));
let t1 = thread::spawn(move || {
    let _ga = a1.lock().unwrap();
    let _gb = b1.lock().unwrap();      // attend b
});

let (a2, b2) = (Arc::clone(&a), Arc::clone(&b));
let t2 = thread::spawn(move || {
    let _gb = b2.lock().unwrap();
    let _ga = a2.lock().unwrap();      // attend a : si t1 tient a et attend b, les deux attendent pour toujours
});

use rayon::prelude::*;

let v: Vec<u64> = (1..=1_000_000).collect();

let somme: u64 = v.iter().sum();                      // un cœur
let somme_par: u64 = v.par_iter().sum();              // tous les cœurs

let carres: Vec<u64> = v.par_iter().map(|x| x * x).collect();
let mut a_trier = v.clone();
a_trier.par_sort_unstable();

// runique/src/utils/config/env.rs (extrait)
static DEBUG: LazyLock<bool> = LazyLock::new(|| debug_from(read_after_dotenv("DEBUG").as_deref()));

pub fn is_debug() -> bool {
    *DEBUG
}

// Jeton de version des fichiers statiques (?v=...), calculé une seule fois :
// une empreinte du contenu des .css/.js du projet et de Runique.
static CSS_TOKEN: LazyLock<String> = LazyLock::new(|| { /* ... */ });

read_after_dotenv charge .env avant de lire la variable : c'est la réponse au piège de la section 8.5. Et puisque la valeur n'est calculée qu'une fois, un fichier CSS modifié en production n'est pris en compte qu'au redémarrage du serveur.

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

static TENTATIVES: LazyLock<Mutex<HashMap<String, u32>>> = LazyLock::new(|| Mutex::new(HashMap::new()));

fn echec_de_connexion(ip: &str) {
    *TENTATIVES.lock().unwrap().entry(ip.to_string()).or_insert(0) += 1;
}
// Chaque adresse IP qui se trompe ajoute une entrée, jamais retirée.

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

fn compter(t: &str) -> HashMap<String, usize> {
    let mut m = HashMap::new();
    for mot in t.split_whitespace() {
        *m.entry(mot.to_lowercase()).or_insert(0) += 1;
    }
    m
}

fn fusionner(total: &mut HashMap<String, usize>, partiel: HashMap<String, usize>) {
    for (k, v) in partiel {
        *total.entry(k).or_insert(0) += v;
    }
}

// 1. Mémoire partagée : chaque thread écrit dans une table commune sous verrou
fn v1(textes: Vec<String>) -> HashMap<String, usize> {
    let total = Arc::new(Mutex::new(HashMap::new()));
    let hs: Vec<_> = textes.into_iter().map(|t| {
        let total = Arc::clone(&total);
        thread::spawn(move || {
            let partiel = compter(&t);                       // calcul hors verrou
            fusionner(&mut total.lock().unwrap(), partiel);  // verrou seulement pour fusionner
        })
    }).collect();
    hs.into_iter().for_each(|h| h.join().unwrap());
    Arc::try_unwrap(total).unwrap().into_inner().unwrap()
}

// 2. Canal : chaque thread envoie son résultat, le thread principal fusionne
fn v2(textes: Vec<String>) -> HashMap<String, usize> {
    let (tx, rx) = mpsc::channel();
    for t in textes {
        let tx = tx.clone();
        thread::spawn(move || tx.send(compter(&t)).unwrap());
    }
    drop(tx);
    let mut total = HashMap::new();
    for partiel in rx {
        fusionner(&mut total, partiel);
    }
    total
}

// 3. scope + join : les threads empruntent les textes, renvoient leur résultat
fn v3(textes: &[String]) -> HashMap<String, usize> {
    thread::scope(|s| {
        let hs: Vec<_> = textes.iter().map(|t| s.spawn(move || compter(t))).collect();
        let mut total = HashMap::new();
        for h in hs {
            fusionner(&mut total, h.join().unwrap());
        }
        total
    })
}

// 4. rayon : map en parallèle, puis réduction
fn v4(textes: &[String]) -> HashMap<String, usize> {
    use rayon::prelude::*;
    textes
        .par_iter()
        .map(|t| compter(t))
        .reduce(HashMap::new, |mut a, b| { fusionner(&mut a, b); a })
}

thread::spawn(move || ..).join()         thread::scope(|s| { s.spawn(|| ..); })
Arc::new(v)   Arc::clone(&a)             Arc::try_unwrap(a)
Mutex::new(v)   m.lock().unwrap()        m.lock().unwrap_or_else(|e| e.into_inner())
m.try_lock()   m.get_mut()   m.into_inner()
RwLock::new(v)   r.read()   r.write()
AtomicU64::new(0)   a.fetch_add(1, Ordering::Relaxed)   a.load(..)   a.fetch_update(..)
mpsc::channel()   mpsc::sync_channel(n)   drop(tx)   rx.iter()
static X: LazyLock<T> = LazyLock::new(|| ..);
static Y: OnceLock<T> = OnceLock::new();   Y.set(v)   Y.get()   Y.get_or_init(|| ..)
v.par_iter().map(f).sum()                // rayon