Canais (channels) em Rust: troca de mensagens entre threads

Na página de Concorrência, as threads compartilhavam dados travando-os com Arc<Mutex<T>>. Isso funciona, mas existe outro caminho que muitas vezes é mais simples: em vez de compartilhar dados, as threads mandam mensagens umas para as outras.

Pense em um channel (canal) como um cano com duas pontas. Uma thread coloca valores na ponta do sender (remetente). Outra thread tira esses valores da ponta do receiver (destinatário), na mesma ordem. Ninguém compartilha nada: cada valor se move de uma thread para a outra. A comunidade de Go tem um ditado que o Rust pega emprestado aqui: “Não se comunique compartilhando memória; em vez disso, compartilhe memória se comunicando.”

Seu primeiro canal

O canal da biblioteca padrão fica em std::sync::mpsc. O nome significa multiple producer, single consumer (vários produtores, um único consumidor): muitos remetentes, um destinatário.

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

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

    thread::spawn(move || {
        let message = String::from("olá, aqui é o worker");
        tx.send(message).unwrap();
    });

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

Saída:

main recebeu: olá, aqui é o worker

▶ Experimente você mesmo

Passo a passo:

• mpsc::channel() retorna um par: um Sender e um Receiver. Por tradição, eles se chamam tx (de transmit, transmitir) e rx (de receive, receber).
• O move entrega o tx para a nova thread, do mesmo jeito que você moveu dados para dentro das threads na página de Concorrência.
• tx.send(valor) coloca um valor no canal. Ele retorna um Result, porque falha se o receiver já não existir mais.
• rx.recv() espera até que um valor chegue. Ele só retorna Err quando todos os senders já se foram e o canal está vazio.

Repare que não precisamos de join(). O recv() já espera pela mensagem, então o main não tem como terminar cedo demais.

Recebendo vários valores

Um Receiver funciona como um iterador. Um laço for sobre ele fica chamando recv() para você, e o laço termina sozinho quando todos os senders forem descartados:

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();
        }
        // o tx é descartado aqui, quando a thread termina
    });

    for value in rx {
        println!("recebi {value}");
    }
    println!("canal fechado, tudo pronto");
}

Saída:

recebi 10
recebi 20
recebi 30
canal fechado, tudo pronto

▶ Experimente você mesmo

Aqui só existe um sender, então os valores chegam exatamente na ordem em que foram enviados. Essa regra de “o laço para quando os senders somem” é importante. Guarde-a para a próxima seção.

Vários produtores: clone o sender

Para deixar várias threads enviarem, dê a cada uma a sua própria cópia do sender com tx.clone(). Todas as cópias alimentam o mesmo 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!("worker {id} se apresentando")).unwrap();
        });
    }
    drop(tx); // descarte o original, senão o laço abaixo nunca termina

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

Saída:

worker 1 se apresentando
worker 2 se apresentando
worker 3 se apresentando

▶ Experimente você mesmo

Duas coisas para reparar:

• O drop(tx) faz diferença. Cada thread descarta o seu clone quando termina, mas o main ainda segura o original. Enquanto existir um sender, o receiver continua esperando, e o programa ficaria travado para sempre. Descartar o original é dizer “o main não vai mandar nada”.
• As threads terminam em qualquer ordem. As mensagens podem chegar como worker 2, worker 1, worker 3. Nós juntamos todas e ordenamos para que a saída impressa seja sempre a mesma. Tire o sort() e rode algumas vezes para ver a ordem mudar.

Sem esperar para sempre: try_recv e recv_timeout

O recv() bloqueia até que algo chegue. Às vezes você não quer esperar. Dois primos ajudam:

• try_recv() retorna na hora: um valor, Empty (nada ainda) ou Disconnected (todos os senders se foram).
• recv_timeout(duration) espera, mas só até um limite de tempo. Se nada chegar a tempo, você recebe Timeout.

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

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

    // Nada foi enviado ainda: try_recv retorna na hora.
    println!("{:?}", rx.try_recv());

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

    // Já tem um valor esperando, então ele é retornado sem espera.
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // O sender se foi e o canal está vazio.
    println!("{:?}", rx.try_recv());
}

Saída:

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

▶ Experimente você mesmo

Fazemos join() na thread antes de receber, então o valor já está esperando e a saída é sempre a mesma.

O try_recv() é útil em um laço que tem outras coisas para fazer, como o laço de um jogo que verifica se há mensagens novas uma vez por quadro. O recv_timeout() é bom para “desista se o worker demorar demais”.

Canais limitados com sync_channel

Um channel() normal não tem limite de tamanho. Se o sender for muito mais rápido que o receiver, as mensagens vão se acumulando na memória. mpsc::sync_channel(n) cria um canal que guarda no máximo n mensagens. Quando ele está cheio, o send() espera até que o receiver tire uma. Isso se chama backpressure (contrapressão): um consumidor lento desacelera automaticamente um produtor rápido.

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

fn main() {
    // Espaço para só 2 mensagens por vez.
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // bloqueia enquanto o buffer está cheio
        }
    });

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

Saída:

total = 15

▶ Experimente você mesmo

O produtor nunca consegue ficar mais de dois números à frente do main. Com sync_channel(0), não existe buffer nenhum: cada send() espera até o receiver estar pronto para pegar o valor, como entregar um pacote em mãos.

Um pequeno pool de workers

Agora vamos juntar tudo. Um pool de workers é um punhado de threads que pegam tarefas de uma fila compartilhada e mandam os resultados de volta. Ele usa dois canais: um para as tarefas e outro para os resultados.

Tem um porém: um Receiver não pode ser clonado (ele é de consumidor único). Então os workers compartilham o receiver de tarefas por trás de um Arc<Mutex<_>> e se revezam para tirar tarefas dele:

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

    // Receivers não podem ser clonados, então os workers compartilham um por trás de 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, // não há mais tarefas
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // avisa os workers que não há mais trabalho

    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} ao quadrado é {square}");
    }
}

Saída:

1 ao quadrado é 1
2 ao quadrado é 4
3 ao quadrado é 9
4 ao quadrado é 16
5 ao quadrado é 25
6 ao quadrado é 36

▶ Experimente você mesmo

Como ele encerra de forma limpa:

1. drop(job_tx) fecha o canal de tarefas. A partir daí, o recv() de cada worker retorna Err, e o worker sai do seu laço.
2. Quando um worker termina, o clone dele de result_tx é descartado. Nós já descartamos o original, então assim que os três workers terminam, o canal de resultados se fecha.
3. Aí o result_rx.iter() para, e nós ordenamos os resultados porque os workers terminam em qualquer ordem.

Um detalhe: job_rx.lock().unwrap().recv() está escrito em uma única instrução let de propósito. A trava é liberada no fim dessa instrução, então os outros workers podem pegar a próxima tarefa enquanto este ainda está calculando o quadrado do seu número.

Canais vs Arc<Mutex<T>>

Os dois são seguros. Eles só se encaixam em formatos diferentes de problema:

Use um canal quando…	Use Arc<Mutex<T>> quando…
O trabalho flui em uma direção: produtores fazem coisas, um consumidor cuida delas	Muitas threads precisam ler e atualizar o mesmo valor (um contador, um cache)
Você quer uma fila de tarefas, eventos ou linhas de log	Os dados são grandes e copiar ou mover tudo de um lado para o outro seria desperdício
Cada valor tem um único dono por vez	As threads precisam olhar o estado atual, e não uma sequência de atualizações

Uma boa regra prática: se você consegue descrever o seu programa como “estas threads contam algo para aquela thread”, use um canal. Se for “estas threads todas editam esta coisa”, use um Mutex. Muitos programas de verdade usam os dois, como o nosso pool de workers.

Erros comuns

Usar um valor depois de enviá-lo. O send() toma o ownership (posse), igual a passar um valor para uma função. Depois de enviado, o valor pertence ao receiver:

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

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

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

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

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

O compilador diz que message foi emprestada depois de ter sido movida. É o Rust protegendo você: se o remetente ainda pudesse alterar message, as duas threads mexeriam nos mesmos dados. Imprima antes de enviar, ou envie um .clone().

Mover o sender para dentro de um laço de threads. A primeira thread leva o tx, então não sobra nada para a segunda:

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

A mensagem diz que tx já foi movido para dentro da closure na iteração anterior do laço. A correção é let tx = tx.clone(); no começo do laço, como no exemplo de vários produtores.

Um programa que nunca termina. Este não gera erro de compilação, o que o torna traiçoeiro. Se você percorre um receiver (for x in rx) enquanto algum sender ainda está vivo, muitas vezes o tx original no main, o laço espera para sempre. Quando o seu programa “simplesmente trava”, verifique se todos os senders são descartados.

Próximos passos

O std::sync::mpsc resolve muita coisa, mas você vai encontrar duas alternativas populares:

• crossbeam-channel (em inglês) traz canais com vários consumidores (receivers que você pode clonar) e uma macro select! para esperar em vários canais ao mesmo tempo.
• tokio::sync::mpsc é o canal para código assíncrono. Se você está usando async/.await (veja Async/Await), use ele em vez do da biblioteca padrão, porque o seu recv().await não bloqueia a thread inteira.

Resumo

• mpsc::channel() te dá um par (Sender, Receiver). O send() coloca um valor lá dentro, e o recv() espera pelo próximo.
• Clone o sender para ter vários produtores. Sempre existe um único receiver.
• Percorrer um receiver termina quando todos os senders são descartados, então lembre de fazer drop() no original.
• try_recv() e recv_timeout() permitem verificar sem esperar para sempre. sync_channel(n) limita a fila a n mensagens.
• Mensagens de várias threads chegam em qualquer ordem: ordene ou some os valores se precisar de uma saída estável.
• Use canais quando os dados fluem de thread para thread. Use Arc<Mutex<T>> quando as threads compartilham e editam um único valor.

Teste o que aprendeu

Teste o que aprendeu

1. O que significa "mpsc"?

Many processes, shared cache (muitos processos, cache compartilhado) Message passing, safe concurrency (troca de mensagens, concorrência segura) Multiple producer, single consumer (vários produtores, um único consumidor) Mutex protected shared channel (canal compartilhado protegido por mutex)

2. Como você deixa três threads enviarem para o mesmo canal?

Dando a cada thread um tx.clone() Embrulhando o receiver em um Rc Chamando mpsc::channel() três vezes e juntando os canais Não dá; um canal tem um único sender

3. Um laço for value in rx nunca termina. Qual é a causa mais provável?

O canal está cheio Um sender (muitas vezes o tx original no main) nunca foi descartado O recv() foi chamado cedo demais Os valores não eram Copy

4. O que o send() faz quando um sync_channel(2) já guarda 2 mensagens?

Descarta a mensagem mais antiga Retorna um erro na hora Dá panic Espera até o receiver tirar uma mensagem

Pontuação: 0 / 4

Para saber mais

• Using Message Passing to Transfer Data Between Threads — The Rust Book (em inglês): a introdução oficial aos canais, com mais exemplos.
• std::sync::mpsc — documentação da biblioteca padrão (em inglês): todos os métodos de Sender, SyncSender e Receiver.
• Channels — Rust by Example (em inglês): um exemplo curto e executável com várias threads produtoras.
• From Single-Threaded to Multithreaded Server — The Rust Book (em inglês): monta um pool de threads completo com canais e Arc<Mutex<Receiver>>, passo a passo.
• tokio::sync::mpsc — docs.rs (em inglês): o canal assíncrono que você vai usar com Tokio, e como ele difere do da biblioteca padrão.