Canali in Rust: scambio di messaggi tra thread

Nella pagina sulla concorrenza i thread condividevano dati bloccandoli con Arc<Mutex<T>>. Funziona, ma c’è un altro modo, spesso più semplice: invece di condividere i dati, i thread si mandano messaggi.

Immagina un canale come un tubo con due estremità. Un thread infila valori nell’estremità del mittente (sender). Un altro thread li tira fuori dall’estremità del destinatario (receiver), nello stesso ordine. Nessuno condivide niente: ogni valore si sposta da un thread all’altro. La comunità di Go ha un motto che Rust riprende volentieri: «Non comunicare condividendo la memoria; condividi la memoria comunicando.»

Il tuo primo canale

Il canale della libreria standard si trova in std::sync::mpsc. Il nome significa multiple producer, single consumer (più produttori, un solo consumatore): tanti mittenti, un destinatario.

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

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

    thread::spawn(move || {
        let message = String::from("ciao dal worker");
        tx.send(message).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("main ha ricevuto: {received}");
}

Output:

main ha ricevuto: ciao dal worker

▶ Provalo tu

Passo dopo passo:

• mpsc::channel() restituisce una coppia: un Sender e un Receiver. Per tradizione si chiamano tx (transmit, trasmettere) e rx (receive, ricevere).
• move consegna tx al nuovo thread, come quando spostavi dati nei thread nella pagina sulla concorrenza.
• tx.send(valore) mette un valore nel canale. Restituisce un Result, perché fallisce se il destinatario non esiste più.
• rx.recv() aspetta finché non arriva un valore. Restituisce Err solo quando tutti i mittenti sono spariti e il canale è vuoto.

Nota che non è servito join(). recv() aspetta già il messaggio, quindi main non può terminare troppo presto.

Ricevere tanti valori

Un Receiver funziona come un iteratore. Un ciclo for su di lui continua a chiamare recv() al posto tuo, e il ciclo finisce da solo quando tutti i mittenti sono stati eliminati (drop):

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 viene eliminato qui, quando il thread finisce
    });

    for value in rx {
        println!("ricevuto {value}");
    }
    println!("canale chiuso, tutto fatto");
}

Output:

ricevuto 10
ricevuto 20
ricevuto 30
canale chiuso, tutto fatto

▶ Provalo tu

Qui c’è un solo mittente, quindi i valori arrivano esattamente nell’ordine in cui sono stati inviati. La regola «il ciclo si ferma quando i mittenti non ci sono più» è importante: tienila a mente per la prossima sezione.

Più produttori: clona il mittente

Per far inviare più thread, dai a ciascuno la sua copia del mittente con tx.clone(). Tutte le copie alimentano lo stesso canale:

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!("worker {id} presente")).unwrap();
        });
    }
    drop(tx); // elimina l'originale, o il ciclo qui sotto non finisce mai

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

Output:

worker 1 presente
worker 2 presente
worker 3 presente

▶ Provalo tu

Due cose da notare:

• drop(tx) è importante. Ogni thread elimina il suo clone quando finisce, ma main tiene ancora l’originale. Finché esiste anche un solo mittente, il destinatario continua ad aspettare e il programma resterebbe bloccato per sempre. Eliminarlo vuol dire «main non invierà niente».
• I thread finiscono in qualsiasi ordine. I messaggi possono arrivare come worker 2, worker 1, worker 3. Li raccogliamo e li ordiniamo, così l’output stampato è sempre lo stesso. Togli il sort() ed esegui il programma qualche volta per vedere l’ordine cambiare.

Non aspettare per sempre: try_recv e recv_timeout

recv() si blocca finché non arriva qualcosa. A volte non vuoi aspettare. Ti aiutano due cugini:

• try_recv() ritorna subito: un valore, Empty (ancora niente) oppure Disconnected (tutti i mittenti spariti).
• recv_timeout(durata) aspetta, ma solo fino a un limite di tempo. Se non arriva niente in tempo, ottieni Timeout.

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

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

    // Ancora niente inviato: try_recv ritorna subito.
    println!("{:?}", rx.try_recv());

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

    // Un valore è in attesa, quindi lo restituisce senza aspettare.
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // Il mittente non c'è più e il canale è vuoto.
    println!("{:?}", rx.try_recv());
}

Output:

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

▶ Provalo tu

Facciamo join() sul thread prima di ricevere, così il valore è già in attesa e l’output è sempre lo stesso.

try_recv() è comodo in un ciclo che ha anche altro da fare, come il ciclo di un gioco che controlla i nuovi messaggi una volta per frame. recv_timeout() va bene per «lascia perdere se il worker ci mette troppo».

Canali limitati con sync_channel

Un normale channel() non ha limiti di dimensione. Se il mittente è molto più veloce del destinatario, i messaggi si accumulano in memoria. mpsc::sync_channel(n) crea un canale che contiene al massimo n messaggi. Quando è pieno, send() aspetta che il destinatario ne tolga uno. Si chiama backpressure (contropressione): un consumatore lento rallenta automaticamente un produttore veloce.

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

fn main() {
    // Spazio per soli 2 messaggi alla volta.
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // si blocca mentre il buffer è pieno
        }
    });

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

Output:

totale = 15

▶ Provalo tu

Il produttore non può mai trovarsi più di due numeri avanti rispetto a main. Con sync_channel(0) non c’è proprio nessun buffer: ogni send() aspetta che il destinatario sia pronto a prendere il valore, come consegnare un pacco a mano.

Un piccolo pool di worker

Ora mettiamo tutto insieme. Un pool di worker è un gruppo di thread che prendono lavori da una coda condivisa e rimandano indietro i risultati. Usa due canali: uno per i lavori, uno per i risultati.

C’è un intoppo: un Receiver non si può clonare (il consumatore è uno solo). Quindi i worker condividono il destinatario dei lavori dietro un Arc<Mutex<_>> e, a turno, ne prelevano un lavoro:

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)>();

    // I Receiver non si possono clonare, quindi i worker ne condividono uno dietro 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, // non ci sono più lavori
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // dice ai worker che il lavoro è finito

    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} al quadrato fa {square}");
    }
}

Output:

1 al quadrato fa 1
2 al quadrato fa 4
3 al quadrato fa 9
4 al quadrato fa 16
5 al quadrato fa 25
6 al quadrato fa 36

▶ Provalo tu

Ecco come si chiude in modo pulito:

1. drop(job_tx) chiude il canale dei lavori. A quel punto il recv() di ogni worker restituisce Err e il worker esce dal suo ciclo.
2. Quando un worker termina, il suo clone di result_tx viene eliminato. L’originale l’abbiamo già eliminato, quindi quando tutti e tre i worker hanno finito, il canale dei risultati si chiude.
3. A quel punto result_rx.iter() si ferma, e ordiniamo i risultati perché i worker finiscono in qualsiasi ordine.

Un piccolo dettaglio: job_rx.lock().unwrap().recv() è scritto apposta in un’unica istruzione let. Il lock viene rilasciato alla fine di quell’istruzione, così gli altri worker possono prendere il lavoro successivo mentre questo sta ancora calcolando il suo quadrato.

Canali o Arc<Mutex<T>>?

Sono sicuri entrambi. Semplicemente si adattano a problemi di forma diversa:

Usa un canale quando…	Usa Arc<Mutex<T>> quando…
Il lavoro scorre in una direzione: i produttori creano cose, un consumatore le gestisce	Tanti thread devono leggere e aggiornare lo stesso valore (un contatore, una cache)
Vuoi una coda di lavori, eventi o righe di log	I dati sono grandi e copiarli o spostarli sarebbe uno spreco
Ogni valore ha un solo proprietario alla volta	I thread devono guardare lo stato attuale, non un flusso di aggiornamenti

Una buona regola pratica: se puoi descrivere il tuo programma come «questi thread dicono qualcosa a quel thread», usa un canale. Se invece è «questi thread modificano tutti questa cosa», usa un Mutex. Molti programmi reali usano entrambi, come il nostro pool di worker.

Errori comuni

Usare un valore dopo averlo inviato. send() prende l’ownership, come quando passi un valore a una funzione. Dopo l’invio, il valore appartiene al destinatario:

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

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

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

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

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

Il compilatore dice che message è stato spostato in send() e poi usato di nuovo (il messaggio resta in inglese, come lo stampa rustc). È Rust che ti protegge: se il mittente potesse ancora modificare message, entrambi i thread toccherebbero gli stessi dati. Stampa prima di inviare, oppure invia un .clone().

Spostare il mittente in un ciclo di thread. Il primo thread si prende tx, e per il secondo non resta niente:

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

Qui rustc spiega che tx è già stato spostato nella closure durante un giro precedente del ciclo. La soluzione è let tx = tx.clone(); all’inizio del ciclo, come nell’esempio con più produttori.

Un programma che non finisce mai. Questo non dà nessun errore del compilatore, ed è per questo che è subdolo. Se fai un ciclo su un destinatario (for x in rx) mentre qualche mittente è ancora vivo, spesso l’originale tx in main, il ciclo aspetta per sempre. Quando il tuo programma «resta lì bloccato», controlla che ogni mittente venga eliminato.

Prossimi passi

std::sync::mpsc copre molto, ma incontrerai due alternative molto usate:

• crossbeam-channel (in inglese) aggiunge canali con più consumatori (destinatari che puoi clonare) e una macro select! per aspettare su più canali contemporaneamente.
• tokio::sync::mpsc è il canale per il codice asincrono. Se usi async/.await (vedi Async/await), usa questo al posto di quello della std, perché il suo recv().await non blocca l’intero thread.

Riepilogo

• mpsc::channel() ti dà una coppia (Sender, Receiver). send() inserisce un valore spostandolo, recv() aspetta il successivo.
• Clona il mittente per avere più produttori. Il destinatario è sempre uno solo.
• Un ciclo su un destinatario finisce quando tutti i mittenti sono stati eliminati, quindi ricordati di fare drop() dell’originale.
• try_recv() e recv_timeout() ti permettono di controllare senza aspettare per sempre. sync_channel(n) limita la coda a n messaggi.
• I messaggi di più thread arrivano in qualsiasi ordine: ordinali o sommali se ti serve un output stabile.
• Usa i canali quando i dati passano da un thread all’altro. Usa Arc<Mutex<T>> quando i thread condividono e modificano un unico valore.

Verifica veloce

Verifica veloce

1. Che cosa significa "mpsc"?

Many processes, shared cache Message passing, safe concurrency Multiple producer, single consumer (più produttori, un solo consumatore) Mutex protected shared channel

2. Come fai a far inviare tre thread nello stesso canale?

Dai a ogni thread un tx.clone() Avvolgi il destinatario in Rc Chiami mpsc::channel() tre volte e li unisci Non si può: un canale ha un solo mittente

3. Un ciclo for value in rx non finisce mai. Qual è la causa più probabile?

Il canale è pieno Un mittente (spesso l'originale tx in main) non è mai stato eliminato recv() è stato chiamato troppo presto I valori non erano Copy

4. Che cosa fa send() quando un sync_channel(2) contiene già 2 messaggi?

Elimina il messaggio più vecchio Restituisce subito un errore Va in panic Aspetta che il destinatario tolga un messaggio

Punteggio: 0 / 4

Per approfondire

• Using Message Passing to Transfer Data Between Threads — The Rust Book (in inglese): l’introduzione ufficiale ai canali, con altri esempi.
• std::sync::mpsc — documentazione della libreria standard (in inglese): tutti i metodi di Sender, SyncSender e Receiver.
• Channels — Rust by Example (in inglese): un esempio breve ed eseguibile con diversi thread produttori.
• From Single-Threaded to Multithreaded Server — The Rust Book (in inglese): costruisce passo dopo passo un thread pool completo con i canali e Arc<Mutex<Receiver>>.
• tokio::sync::mpsc — docs.rs (in inglese): il canale asincrono che userai con Tokio, e in cosa differisce da quello della std.