Les canaux (channels) en Rust : passage de messages entre threads

Sur la page La concurrence en Rust, les threads partageaient des données en les verrouillant avec Arc<Mutex<T>>. Ça marche, mais il existe une autre approche, souvent plus simple : au lieu de partager des données, les threads s’envoient des messages.

Imaginez un canal (channel) comme un tuyau à deux bouts. Un thread dépose des valeurs dans le bout émetteur (sender). Un autre thread les récupère par le bout récepteur (receiver), dans le même ordre. Personne ne partage rien : chaque valeur passe d’un thread à l’autre. La communauté Go a un dicton que Rust reprend à son compte : « Ne communiquez pas en partageant la mémoire ; partagez plutôt la mémoire en communiquant. »

Votre premier canal

Le canal de la bibliothèque standard se trouve dans std::sync::mpsc. Ce nom signifie multiple producer, single consumer (plusieurs producteurs, un seul consommateur) : de nombreux émetteurs, un seul récepteur.

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

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

    thread::spawn(move || {
        let message = String::from("bonjour de la part du worker");
        tx.send(message).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("main a reçu : {received}");
}

Sortie :

main a reçu : bonjour de la part du worker

▶ Essayez par vous-même

Étape par étape :

• mpsc::channel() renvoie une paire : un Sender et un Receiver. Par tradition, on les appelle tx (transmit, émettre) et rx (receive, recevoir).
• move confie tx au nouveau thread, de la même façon que vous avez déplacé des données dans des threads sur la page La concurrence en Rust.
• tx.send(value) dépose une valeur dans le canal. Il renvoie un Result, car l’envoi échoue si le récepteur a déjà disparu.
• rx.recv() attend qu’une valeur arrive. Il ne renvoie Err que lorsque tous les émetteurs ont disparu et que le canal est vide.

Remarquez qu’on n’a pas eu besoin de join(). recv() attend déjà le message, donc main ne peut pas se terminer trop tôt. Ici, « worker » désigne simplement le thread qui fait le travail.

Recevoir plusieurs valeurs

Un Receiver fonctionne comme un itérateur. Une boucle for dessus appelle recv() pour vous, encore et encore, et la boucle s’arrête d’elle-même quand tous les émetteurs ont été détruits (dropped) :

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

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

    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i * 10).unwrap();
        }
        // tx est détruit ici, à la fin du thread
    });

    for value in rx {
        println!("reçu {value}");
    }
    println!("canal fermé, tout est terminé");
}

Sortie :

reçu 10
reçu 20
reçu 30
canal fermé, tout est terminé

▶ Essayez par vous-même

Il n’y a qu’un seul émetteur ici, donc les valeurs arrivent exactement dans l’ordre où elles ont été envoyées. Cette règle, « la boucle s’arrête quand les émetteurs ont disparu », est importante. Gardez-la en tête pour la section suivante.

Plusieurs producteurs : cloner l’émetteur

Pour que plusieurs threads puissent envoyer, donnez à chacun sa propre copie de l’émetteur avec tx.clone(). Toutes les copies alimentent le même canal :

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

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

    for id in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(format!("le worker {id} au rapport")).unwrap();
        });
    }
    drop(tx); // détruire l'original, sinon la boucle ci-dessous ne finit jamais

    let mut messages: Vec<String> = rx.iter().collect();
    messages.sort();
    for m in messages {
        println!("{m}");
    }
}

Sortie :

le worker 1 au rapport
le worker 2 au rapport
le worker 3 au rapport

▶ Essayez par vous-même

Deux choses à remarquer :

• drop(tx) compte. Chaque thread détruit son clone en se terminant, mais main garde toujours l’original. Tant qu’un émetteur existe, le récepteur continue d’attendre, et le programme resterait bloqué pour toujours. Le détruire revient à dire « main n’enverra rien ».
• Les threads terminent dans n’importe quel ordre. Les messages peuvent arriver dans l’ordre worker 2, worker 1, worker 3. On les rassemble et on les trie pour que la sortie soit toujours la même. Supprimez le sort() et lancez le programme plusieurs fois pour voir l’ordre changer.

Ne pas attendre indéfiniment : try_recv et recv_timeout

recv() bloque jusqu’à ce que quelque chose arrive. Parfois, vous ne voulez pas attendre. Deux cousines viennent à la rescousse :

• try_recv() répond immédiatement : une valeur, Empty (rien pour l’instant) ou Disconnected (tous les émetteurs ont disparu).
• recv_timeout(duration) attend, mais seulement jusqu’à une limite de temps. Si rien n’arrive à temps, vous obtenez Timeout.

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

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

    // Rien n'a encore été envoyé : try_recv répond tout de suite.
    println!("{:?}", rx.try_recv());

    thread::spawn(move || tx.send(7).unwrap()).join().unwrap();

    // Une valeur attend déjà, donc on la récupère sans attendre.
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // L'émetteur a disparu et le canal est vide.
    println!("{:?}", rx.try_recv());
}

Sortie :

Err(Empty)
Ok(7)
Err(Disconnected)

▶ Essayez par vous-même

On appelle join() sur le thread avant de recevoir : la valeur attend donc déjà dans le canal, et la sortie est toujours la même.

try_recv() est pratique dans une boucle qui a autre chose à faire, comme la boucle d’un jeu qui vérifie s’il y a de nouveaux messages une fois par image. recv_timeout() convient bien pour « abandonner si le worker met trop de temps ».

Les canaux bornés avec sync_channel

Un channel() normal n’a pas de limite de taille. Si l’émetteur est beaucoup plus rapide que le récepteur, les messages s’entassent en mémoire. mpsc::sync_channel(n) crée un canal qui contient au maximum n messages. Quand il est plein, send() attend que le récepteur en retire un. C’est ce qu’on appelle la backpressure (contre-pression) : un consommateur lent ralentit automatiquement un producteur rapide.

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

fn main() {
    // De la place pour seulement 2 messages à la fois.
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // bloque tant que le tampon est plein
        }
    });

    let total: i32 = rx.iter().sum();
    producer.join().unwrap();
    println!("total = {total}");
}

Sortie :

total = 15

▶ Essayez par vous-même

Le producteur ne peut jamais avoir plus de deux nombres d’avance sur main. Avec sync_channel(0), il n’y a pas de tampon du tout : chaque send() attend que le récepteur soit prêt à prendre la valeur, comme quand on remet un colis en main propre.

Un petit pool de workers

Maintenant, assemblons le tout. Un pool de workers est un petit groupe de threads qui prennent des tâches dans une file partagée et renvoient des résultats. Il utilise deux canaux : un pour les tâches, un pour les résultats.

Il y a un piège : un Receiver ne peut pas être cloné (il n’y a qu’un seul consommateur). Les workers partagent donc le récepteur des tâches derrière un Arc<Mutex<_>>, et se relaient pour y piocher :

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

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u64>();
    let (result_tx, result_rx) = mpsc::channel::<(u64, u64)>();

    // Un Receiver ne se clone pas : les workers en partagent un derrière Arc<Mutex<_>>.
    let job_rx = Arc::new(Mutex::new(job_rx));

    let mut workers = Vec::new();
    for _ in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        workers.push(thread::spawn(move || loop {
            let job = job_rx.lock().unwrap().recv();
            match job {
                Ok(n) => result_tx.send((n, n * n)).unwrap(),
                Err(_) => break, // plus aucune tâche
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // indique aux workers qu'il n'y a plus de travail

    let mut results: Vec<(u64, u64)> = result_rx.iter().collect();
    for w in workers {
        w.join().unwrap();
    }
    results.sort();
    for (n, square) in results {
        println!("{n} au carré vaut {square}");
    }
}

Sortie :

1 au carré vaut 1
2 au carré vaut 4
3 au carré vaut 9
4 au carré vaut 16
5 au carré vaut 25
6 au carré vaut 36

▶ Essayez par vous-même

Comment le pool s’arrête proprement :

1. drop(job_tx) ferme le canal des tâches. Le recv() de chaque worker renvoie alors Err, et le worker sort de sa boucle.
2. Quand un worker se termine, son clone de result_tx est détruit. Comme on a déjà détruit l’original, le canal des résultats se ferme dès que les trois workers ont fini.
3. result_rx.iter() s’arrête alors, et on trie les résultats, car les workers terminent dans n’importe quel ordre.

Un petit détail : job_rx.lock().unwrap().recv() est écrit exprès dans une seule instruction let. Le verrou est libéré à la fin de cette instruction, ce qui permet aux autres workers de prendre la tâche suivante pendant que celui-ci calcule encore son carré.

Canaux ou Arc<Mutex<T>> ?

Les deux sont sûrs. Ils correspondent simplement à des formes de problèmes différentes :

Utilisez un canal quand…	Utilisez Arc<Mutex<T>> quand…
Le travail circule dans un seul sens : des producteurs fabriquent, un consommateur traite	Plusieurs threads doivent lire et modifier la même valeur (un compteur, un cache)
Vous voulez une file de tâches, d’événements ou de lignes de log	Les données sont volumineuses, et les copier ou les déplacer serait du gaspillage
Chaque valeur n’a qu’un seul propriétaire à la fois	Les threads ont besoin de voir l’état actuel, pas un flux de mises à jour

Une bonne règle empirique : si vous pouvez décrire votre programme par « ces threads disent quelque chose à ce thread-là », utilisez un canal. Si c’est plutôt « ces threads modifient tous cette chose », utilisez un Mutex. Beaucoup de vrais programmes utilisent les deux, comme notre pool de workers.

Erreurs courantes

Utiliser une valeur après l’avoir envoyée. send() prend possession de la valeur, exactement comme quand on passe une valeur à une fonction. Une fois envoyée, elle appartient au récepteur :

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

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

    thread::spawn(move || {
        let message = String::from("salut");
        tx.send(message).unwrap();
        println!("envoyé : {message}");
    });

    println!("{}", rx.recv().unwrap());
}

error[E0382]: borrow of moved value: `message`
  --> src/main.rs:10:29
   |
 8 |         let message = String::from("salut");
   |             ------- move occurs because `message` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(message).unwrap();
   |                 ------- value moved here
10 |         println!("envoyé : {message}");
   |                             ^^^^^^^ value borrowed here after move

Le compilateur dit (en anglais) que message est emprunté après avoir été déplacé. C’est Rust qui vous protège : si l’émetteur pouvait encore modifier message, les deux threads toucheraient les mêmes données. Affichez la valeur avant de l’envoyer, ou envoyez un .clone().

Déplacer l’émetteur dans une boucle de threads. Le premier thread prend tx, et il ne reste plus rien pour le deuxième :

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

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

    for id in 1..=3 {
        thread::spawn(move || {
            tx.send(id).unwrap();
        });
    }

    for value in rx {
        println!("{value}");
    }
}

error[E0382]: use of moved value: `tx`
  --> src/main.rs:8:23
   |
 5 |     let (tx, rx) = mpsc::channel();
   |          -- move occurs because `tx` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait
...
 8 |         thread::spawn(move || {
   |                       ^^^^^^^ value moved into closure here, in previous iteration of loop

Ici, le compilateur signale que tx a déjà été déplacé dans la closure lors d’un tour de boucle précédent. La solution : let tx = tx.clone(); en haut de la boucle, comme dans l’exemple avec plusieurs producteurs.

Un programme qui ne se termine jamais. Celle-ci ne provoque aucune erreur de compilation, ce qui la rend sournoise. Si vous bouclez sur un récepteur (for x in rx) alors qu’un émetteur est encore vivant, souvent le tx d’origine dans main, la boucle attend pour toujours. Quand votre programme « reste bloqué », vérifiez que chaque émetteur finit bien par être détruit.

Et ensuite ?

std::sync::mpsc couvre beaucoup de besoins, mais vous croiserez deux alternatives populaires :

• crossbeam-channel (en anglais) ajoute des canaux à plusieurs consommateurs (des récepteurs que l’on peut cloner) et une macro select! pour attendre sur plusieurs canaux à la fois.
• tokio::sync::mpsc est le canal du code asynchrone. Si vous utilisez async/.await (voir Async/Await en Rust), préférez-le à celui de la bibliothèque standard, car son recv().await ne bloque pas tout le thread.

Récapitulatif

• mpsc::channel() vous donne une paire (Sender, Receiver). send() y dépose une valeur, recv() attend la suivante.
• Clonez l’émetteur pour avoir plusieurs producteurs. Il n’y a jamais qu’un seul récepteur.
• Une boucle sur un récepteur se termine quand tous les émetteurs ont été détruits : pensez à appeler drop() sur l’original.
• try_recv() et recv_timeout() permettent de vérifier sans attendre indéfiniment. sync_channel(n) limite la file à n messages.
• Les messages venant de plusieurs threads arrivent dans n’importe quel ordre : triez-les ou additionnez-les si vous voulez une sortie stable.
• Utilisez des canaux quand les données circulent d’un thread à l’autre. Utilisez Arc<Mutex<T>> quand des threads partagent et modifient une même valeur.

Petit quiz

Petit quiz

1. Que signifie « mpsc » ?

Many processes, shared cache (plusieurs processus, cache partagé) Message passing, safe concurrency (passage de messages, concurrence sûre) Multiple producer, single consumer (plusieurs producteurs, un seul consommateur) Mutex protected shared channel (canal partagé protégé par un mutex)

2. Comment permettre à trois threads d'envoyer dans le même canal ?

Donner à chaque thread un tx.clone() Envelopper le récepteur dans un Rc Appeler mpsc::channel() trois fois et fusionner les canaux C'est impossible : un canal n'a qu'un seul émetteur

3. Une boucle for value in rx ne se termine jamais. Quelle est la cause la plus probable ?

Le canal est plein Un émetteur (souvent le tx d'origine dans main) n'a jamais été détruit recv() a été appelé trop tôt Les valeurs n'étaient pas Copy

4. Que fait send() quand un sync_channel(2) contient déjà 2 messages ?

Il supprime le message le plus ancien Il renvoie immédiatement une erreur Il provoque une panic Il attend que le récepteur retire un message

Score : 0 / 4

Pour aller plus loin

• Using Message Passing to Transfer Data Between Threads — The Rust Book (en anglais) : l’introduction officielle aux canaux, avec d’autres exemples.
• std::sync::mpsc — documentation de la bibliothèque standard (en anglais) : toutes les méthodes de Sender, SyncSender et Receiver.
• Channels — Rust by Example (en anglais) : un court exemple exécutable avec plusieurs threads producteurs.
• From Single-Threaded to Multithreaded Server — The Rust Book (en anglais) : construit pas à pas un pool de threads complet avec des canaux et Arc<Mutex<Receiver>>.
• tokio::sync::mpsc — docs.rs (en anglais) : le canal asynchrone que vous utiliserez avec Tokio, et ce qui le distingue de celui de la bibliothèque standard.