Concurrence sûre dans Rust avec Mutex et channel

1. Tổng quan

Rust appelle sa capacité « fearless concurrency ». Le compilateur empêche les data race dès la compilation, avant que le programme ne s’exécute. Cet article présente deux outils principaux de la bibliothèque standard : Mutex (avec Arc) pour partager l’état, et les channels mpsc pour transmettre des messages entre threads. Après la lecture, vous saurez utiliser chacun d’eux, écrire des tests pour ces outils et éviter les pièges courants comme les deadlocks ou garder un verrou trop longtemps.

Tous les exemples utilisent uniquement la bibliothèque standard, aucune dépendance externe requise. Vous pouvez créer un projet avec cargo new orders-demo puis coller le code dans src/main.rs.

2. Vấn đề : chia sẻ dữ liệu giữa les threads

Supposons que vous ayez plusieurs threads traitant les commandes simultanément et que vous vouliez compter le nombre total de commandes. Voici une première approche naïve :

use std::thread;

fn main() {
    let mut total_orders = 0;

    let handles: Vec<_> = (0..4)
        .map(|_| thread::spawn(|| total_orders += 1))
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

Ce code ne compile pas :

error[E0373]: closure may outlive the current function, but it borrows `total_orders`, which is owned by the current function

Le compilateur donne deux raisons :

• thread::spawn exige que la closure soit 'static. Le thread enfant peut vivre plus longtemps que main, elle ne peut donc pas emprunter une variable locale.
• Même en ajoutant move, chaque thread recevra sa propre copie de total_orders. Personne n’accumule vraiment dans une variable commune.

Pour résoudre cela, vous avez besoin de deux choses : partager la propriété avec Arc et synchroniser l’accès avec Mutex.

3. Partager l’état avec Arc<Mutex<T>>

3.1. Compter les commandes depuis plusieurs threads

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

fn main() {
    let total_orders = Arc::new(Mutex::new(0u32));
    let mut handles = Vec::new();

    for _ in 0..4 {
        let counter = Arc::clone(&total_orders);
        handles.push(thread::spawn(move || {
            for _ in 0..1_000 {
                *counter.lock().unwrap() += 1;
            }
        }));
    }

    for h in handles {
        h.join().unwrap();
    }

    println!("Nombre total de commandes : {}", *total_orders.lock().unwrap());
}

Nombre total de commandes : 4000

Explications partie par partie :

• Arc::new(Mutex::new(0u32)) : Arc (Atomic Reference Counted) permet à plusieurs threads de posséder simultanément une valeur. Mutex garantit qu’à tout moment seul un thread peut accéder à la valeur à l’intérieur.
• Arc::clone(&total_orders) augmente juste le compteur de références, ne copie pas les données. Chaque thread reçoit son propre Arc via move.
• counter.lock() bloque le thread actuel jusqu’à obtenir le verrou. La fonction retourne Result<MutexGuard<u32>, PoisonError<…>>.
• MutexGuard se dérécence en &mut u32, donc *... += 1 modifie directement la valeur. Le guard est droppé à la fin de l’instruction, et le verrou est libéré automatiquement à ce moment.

Vous ne pouvez pas « oublier de déverrouiller » dans Rust. Vous ne pouvez pas non plus lire une valeur sans verrouillage, car les données se trouvent à l’intérieur du Mutex.

Note : Vous ne pouvez pas remplacer Arc par Rc. Rc n’implémente pas Send, et le compilateur signalera une erreur dès que vous le passez à thread::spawn. De même, vous ne pouvez pas remplacer Mutex par RefCell, car RefCell n’est pas Sync.

3.2. Portée du MutexGuard

Le verrou est maintenu jusqu’à ce que le guard soit dropé. Si vous assignez le guard à une variable, le verrou vit jusqu’à la fin de la portée de cette variable :

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

fn reserve(inventory: &Mutex<HashMap<String, u32>>, sku: &str) -> bool {
    let reserved = {
        let mut stock = inventory.lock().unwrap();
        match stock.get_mut(sku) {
            Some(qty) if *qty > 0 => {
                *qty -= 1;
                true
            }
            _ => false,
        }
    }; // le guard est droppé ici, le verrou est libéré

    if reserved {
        // Simulation d'appel PaymentService lent, sans garder le verrou
        thread::sleep(Duration::from_millis(50));
    }
    reserved
}

fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("AO-THUN".to_string(), 2)])));

    let handles: Vec<_> = (0..3)
        .map(|i| {
            let inv = Arc::clone(&inventory);
            thread::spawn(move || println!("Client {i}: {}", reserve(&inv, "AO-THUN")))
        })
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

Client 0: true
Client 1: true
Client 2: false

L’ordre des lignes peut varier entre les exécutions. Le résultat est toujours exactement deux true et un false.

Le bloc { ... } limite le temps de verrouillage au minimum. La partie lente (appel de paiement) s’exécute après que le verrou a été libéré, donc les autres threads n’ont pas à attendre. Si vous devez libérer plus tôt au milieu du traitement, vous pouvez appeler drop(stock) explicitement.

3.3. Mutex empoisonné

Si un thread panique en tenant un verrou, Rust marque le Mutex comme empoisonné. La raison est que les données peuvent être dans un état inconsistant :

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

fn main() {
    let balance = Arc::new(Mutex::new(100i64));

    let b = Arc::clone(&balance);
    let _ = thread::spawn(move || {
        let mut guard = b.lock().unwrap();
        *guard -= 30;
        panic!("PaymentService erreur intermédiaire");
    })
    .join();

    match balance.lock() {
        Ok(v) => println!("Solde : {}", *v),
        Err(poisoned) => {
            let v = poisoned.into_inner();
            println!("Mutex empoisonné, solde actuel : {}", *v);
        }
    }
}

thread '<unnamed>' panicked at src/main.rs:10:9:
PaymentService erreur intermédiaire
Mutex empoisonné, solde actuel : 70

Ainsi, .unwrap() après lock() signifie essentiellement « si un autre thread a paniqué en tenant le verrou, paniquez aussi ». Pour la plupart des applications, c’est un choix raisonnable. Si vous savez comment récupérer les données, into_inner() retourne le guard pour que vous puissiez vérifier et corriger.

4. Transmettre des messages avec des channels

Mutex répond à la question « plusieurs threads modifient ensemble une donnée ». Les channels vont dans la direction opposée : chaque donnée n’a qu’un seul propriétaire à la fois, et la propriété est transférée via le channel. Pour de nombreux problèmes, cette approche est plus simple et moins sujette aux erreurs.

4.1. Plusieurs producteurs, un consommateur

std::sync::mpsc est l’abréviation de multiple producer, single consumer :

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

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

fn main() {
    let (tx, rx) = mpsc::channel::<Order>();

    for shop in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let order = Order {
                id: shop * 100,
                customer: format!("client-{shop}"),
                amount: 150_000 * shop as u64,
            };
            tx.send(order).unwrap();
        });
    }
    drop(tx); // important : libérer le sender d'origine

    let mut revenue = 0;
    for order in rx {
        println!("Commande reçue #{} de {}", order.id, order.customer);
        revenue += order.amount;
    }
    println!("Chiffre d'affaires : {revenue}");
}

Commande reçue #100 de client-1
Commande reçue #300 de client-3
Commande reçue #200 de client-2
Chiffre d'affaires : 900000

Explications :

• mpsc::channel() retourne la paire (Sender, Receiver). Chaque producteur reçoit un clone tx.clone().
• tx.send(order) déplace order dans le channel. Après cette ligne, le thread expéditeur ne peut plus utiliser order, et le compilateur signalera une erreur si vous essayez. De cette façon, deux threads ne modifient jamais une Order.
• for order in rx boucle jusqu’à ce que tous les Sender soient droppés. Si vous oubliez la ligne drop(tx), le sender d’origine reste vivant dans main, et la boucle attendra indéfiniment.

L’ordre de réception dépend de la planification des threads. Le chiffre d’affaires total est toujours 900000.

4.2. Channel avec limite et backpressure

mpsc::channel() a un buffer illimité. Si le producteur est plus rapide que le consommateur, la mémoire gonflera progressivement. sync_channel(n) limite le buffer à n éléments, et send bloquera si le buffer est plein :

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u32>(2);

    let producer = thread::spawn(move || {
        for id in 1..=5 {
            tx.send(id).unwrap();
            println!("Commande {id} envoyée");
        }
    });

    thread::sleep(Duration::from_millis(100));
    for id in rx {
        println!("Traitement de la commande {id}");
        thread::sleep(Duration::from_millis(20));
    }
    producer.join().unwrap();
}

Commande 1 envoyée
Commande 2 envoyée
Traitement de la commande 1
Commande 3 envoyée
Traitement de la commande 2
...

Le producteur envoie deux commandes puis doit attendre que le consommateur en retire certaines. C’est un mécanisme de backpressure naturel, très utile pour les pipelines de traitement de données. Avec sync_channel(0), chaque send doit attendre qu’il y ait un recv (rendez-vous).

5. Quand ne pas besoin de Arc : threads limités

Si les threads se terminent certainement avant le retour de la fonction, thread::scope vous permet d’emprunter directement des données sans Arc :

use std::sync::Mutex;
use std::thread;

fn main() {
    let processed = Mutex::new(Vec::new());
    let orders = vec![101, 102, 103, 104];

    thread::scope(|s| {
        for chunk in orders.chunks(2) {
            s.spawn(|| processed.lock().unwrap().extend_from_slice(chunk));
        }
    }); // tous les threads dans le scope ont rejoint ici

    let mut result = processed.into_inner().unwrap();
    result.sort();
    println!("{result:?}");
}

[101, 102, 103, 104]

Grâce à scope, le compilateur sait que les threads ne vivent pas plus longtemps que processed et orders. Par conséquent, emprunter avec & est valide. Après le scope, into_inner() extrait les données du Mutex sans avoir besoin d’un verrou.

6. Écrire des tests pour du code concurrent

Vous devez séparer la logique en fonctions pures pour faciliter les tests. Mettez le code suivant dans src/lib.rs :

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

pub fn count_orders(workers: u32, per_worker: u32) -> u32 {
    let total = Arc::new(Mutex::new(0));
    let handles: Vec<_> = (0..workers)
        .map(|_| {
            let total = Arc::clone(&total);
            thread::spawn(move || {
                for _ in 0..per_worker {
                    *total.lock().unwrap() += 1;
                }
            })
        })
        .collect();
    handles.into_iter().for_each(|h| h.join().unwrap());
    let result = *total.lock().unwrap();
    result
}

pub fn sum_payments(payments: Vec<u64>) -> u64 {
    let (tx, rx) = mpsc::channel();
    for amount in payments {
        let tx = tx.clone();
        thread::spawn(move || tx.send(amount).unwrap());
    }
    drop(tx);
    rx.iter().sum()
}

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

    #[test]
    fn compte_chaque_commande_une_seule_fois() {
        assert_eq!(count_orders(8, 10_000), 80_000);
    }

    #[test]
    fn somme_tous_les_paiements_du_channel() {
        assert_eq!(sum_payments(vec![50_000, 120_000, 30_000]), 200_000);
    }

    #[test]
    fn paiements_vides_retourne_zero() {
        assert_eq!(sum_payments(vec![]), 0);
    }
}

cargo test

running 3 tests
test tests::compte_chaque_commande_une_seule_fois ... ok
test tests::somme_tous_les_paiements_du_channel ... ok
test tests::paiements_vides_retourne_zero ... ok

test result: ok. 3 passed; 0 failed

Le test paiements_vides_retourne_zero vérifie un cas limite : si vous oubliez drop(tx), ce test se bloquera indéfiniment au lieu de retourner 0. C’est un indicateur clair pour détecter l’erreur.

Il y a un détail dans count_orders : la valeur est assignée à la variable result avant le retour. C’est parce que le guard temporaire dans l’expression finale de la fonction pourrait vivre plus longtemps que total, et le compilateur signalerait une erreur d’emprunt.

7. Erreurs courantes

Deadlock dû à l’ordre inverse du verrouillage. Le thread A verrouille orders puis customers, tandis que le thread B verrouille customers puis orders. Les deux vont se bloquer mutuellement indéfiniment. Rust n’empêche que les data race, pas les deadlocks. Vous devez convenir d’un ordre de verrouillage fixe, ou fusionner les deux données dans un seul Mutex<(A, B)>.

Le guard temporaire vit plus longtemps que vous le pensez. Voir le code suivant :

while let Some(order) = queue.lock().unwrap().pop() {
    process(order); // le verrou est toujours maintenu ici !
}

Le temporaire dans l’expression de while let (et match) vit jusqu’à la fin du corps de la boucle. Tout autre thread voulant ajouter une commande à queue sera bloqué pendant toute la durée de process. La solution est de séparer : let next = queue.lock().unwrap().pop(); puis traiter.

Appeler à nouveau lock() sur le même thread. std::sync::Mutex n’est pas réentrant. Appeler lock() deux fois sur le même thread causera un deadlock ou une panique.

Garder le verrou en attendant I/O ou recv(). Toute opération bloquante en tenant un guard transforme le Mutex en goulot d’étranglement. Vous devez copier ou cloner les données nécessaires, libérer le verrou, puis faire le travail lent.

Oublier de drop Sender. Comme vu ci-dessus, le consommateur attendra indéfiniment.

Note : Dans le code async (Tokio), ne conservez pas std::sync::MutexGuard au-delà de .await. Utilisez tokio::sync::Mutex ou les channels de Tokio. De plus, pour un simple entier comme un compteur, AtomicU32 est beaucoup plus rapide que Mutex<u32>.

Quand choisir quel outil ?

• Utilisez Mutex quand plusieurs threads doivent lire/modifier le même état de longue durée, par exemple un cache ou un inventaire. Si les lectures dépassent nettement les écritures, envisagez RwLock.
• Utilisez les channels quand les données circulent d’un endroit à un autre, par exemple une queue de commandes, un pipeline ou un worker pool. Les channels assurent que chaque valeur n’a qu’un propriétaire, rendant la logique plus facile à raisonner.

8. Conclusion

Rust transforme les erreurs courantes de concurrence en erreurs de compilation. Vous ne pouvez pas partager de données sans Arc, vous ne pouvez pas modifier de données sans Mutex, et vous ne pouvez pas réutiliser une valeur envoyée via un channel. Arc<Mutex<T>> convient à l’état partagé, tandis que les channels mpsc conviennent au transfert de propriété entre threads. sync_channel ajoute la backpressure, et thread::scope aide à éliminer Arc quand la durée de vie des threads est claire. Ce que le compilateur ne peut pas protéger, c’est les deadlocks, les guards qui vivent trop longtemps et les senders non droppés. Ce sont des endroits où vous devez examiner attentivement et écrire des tests soigneusement.