Em sistemas assíncronos, a comunicação entre tarefas concorrentes é essencial. Canais são uma abstração que permite transferir dados de um produtor para um consumidor de forma segura e eficiente. Nesta aula, vamos nos aprofundar nos canais fornecidos pelo crate tokio::sync, que são projetados para funcionar perfeitamente com o runtime assíncrono do Tokio. Entender as diferenças entre os tipos de canal e quando aplicá-los é crucial para construir aplicações concorrentes robustas.

Vamos começar com uma visão geral dos três principais tipos de canais: mpsc (multi-produtor, único consumidor), oneshot (um único envio) e broadcast (múltiplos consumidores). Cada um tem características específicas que os tornam adequados para diferentes cenários. Também discutiremos o conceito de backpressure, que é a capacidade de um sistema de sinalizar ao produtor para desacelerar quando o consumidor não consegue acompanhar, evitando sobrecarga e perda de dados.

tokio::sync (mpsc, oneshot, broadcast)

O módulo tokio::sync oferece várias primitivas de sincronização, incluindo canais. Os três tipos mais comuns são:

  • mpsc (Multi-Producer, Single-Consumer): permite que múltiplos produtores enviem mensagens para um único consumidor. É útil para distribuir trabalho entre várias tarefas que alimentam um único processador.
  • oneshot: é um canal de uso único, onde exatamente uma mensagem pode ser enviada. Ideal para comunicação de resposta única, como aguardar o resultado de uma computação.
  • broadcast: envia cada mensagem para todos os consumidores registrados. Perfeito para eventos que precisam ser disseminados para várias partes interessadas.

Abaixo, um exemplo de cada tipo:

use tokio::sync::{mpsc, oneshot, broadcast};

#[tokio::main]
async fn main() {
    // mpsc: múltiplos produtores, um consumidor
    let (tx, mut rx) = mpsc::channel(32);
    tokio::spawn(async move {
        tx.send("mensagem").await.unwrap();
    });
    if let Some(msg) = rx.recv().await {
        println!("mpsc recebeu: {}", msg);
    }

    // oneshot: um envio, uma recepção
    let (tx, rx) = oneshot::channel();
    tokio::spawn(async move {
        tx.send(42).unwrap();
    });
    if let Ok(val) = rx.await {
        println!("oneshot recebeu: {}", val);
    }

    // broadcast: muitos consumidores
    let (tx, mut rx1) = broadcast::channel(16);
    let mut rx2 = tx.subscribe();
    tx.send("evento").unwrap();
    println!("broadcast rx1: {:?}", rx1.recv().await);
    println!("broadcast rx2: {:?}", rx2.recv().await);
}

No exemplo acima, note que mpsc::channel retorna um par (transmissor, receptor). O transmissor pode ser clonado para múltiplos produtores. oneshot::channel retorna um par onde o receptor é um futuro que resolve para o valor enviado. broadcast::channel retorna um transmissor que pode ser clonado e também permite criar novos receptores via subscribe().

Quando usar cada um

A escolha do canal depende do padrão de comunicação necessário. Aqui estão diretrizes práticas:

  • mpsc: use quando você tem várias tarefas produzindo dados que precisam ser processados por uma única tarefa. Exemplo: um servidor que recebe requisições de múltiplas conexões e as enfileira para um worker processá-las.
  • oneshot: use para comunicação de resposta única, como aguardar o resultado de uma operação assíncrona. Exemplo: um cliente que envia uma requisição e aguarda a resposta do servidor.
  • broadcast: use quando você precisa notificar todos os interessados sobre um evento. Exemplo: um sistema de chat onde cada mensagem deve ser enviada a todos os participantes.

Vamos ver um cenário mais concreto: um pipeline de processamento de imagens. Várias tarefas produtoras geram miniaturas, e um único consumidor as salva em disco. Isso é um caso típico para mpsc. Para aguardar a conclusão de uma tarefa específica, como a geração de um relatório, oneshot é ideal. Para transmitir atualizações de status para vários observadores, broadcast é o mais adequado.

É importante lembrar que mpsc e oneshot são canais com backpressure: se o consumidor não estiver lendo, o produtor pode ser pausado (via await) até que haja espaço. Já broadcast tem um comportamento diferente: se um consumidor estiver lento, mensagens podem ser descartadas para ele, a menos que você use o modo de espera (recv() com await) que pode bloquear até a próxima mensagem, mas sem backpressure no produtor.

Backpressure

Backpressure é um mecanismo crucial em sistemas assíncronos para evitar que produtores sobrecarreguem consumidores. Em canais com buffer limitado, como mpsc e oneshot, quando o buffer está cheio, o send().await aguarda até que haja espaço. Isso naturalmente desacelera o produtor, aplicando backpressure.

Vamos ver um exemplo prático de backpressure com mpsc:

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(2); // buffer pequeno

    tokio::spawn(async move {
        for i in 0..10 {
            tx.send(i).await.unwrap();
            println!("Enviado {}", i);
            sleep(Duration::from_millis(100)).await;
        }
    });

    // Consumidor lento
    while let Some(val) = rx.recv().await {
        println!("Recebido {}", val);
        sleep(Duration::from_millis(300)).await;
    }
}

Neste exemplo, o buffer tem capacidade para 2 mensagens. O produtor tenta enviar 10 mensagens rapidamente, mas como o consumidor é lento, o buffer enche e o send().await bloqueia até que o consumidor libere espaço. Isso impede que o produtor fique muito à frente, garantindo que o sistema não acumule trabalho não processado.

Em broadcast, o backpressure é diferente: o produtor não é bloqueado, mas cada receptor tem seu próprio buffer. Se um receptor não conseguir acompanhar, as mensagens mais antigas são descartadas para ele (a menos que você use recv() que retorna um erro Lagged). Isso pode ser aceitável para eventos onde a perda de mensagens antigas é tolerável, mas não para dados críticos.

Em resumo, escolha canais com backpressure (mpsc, oneshot) quando a ordem e a integridade dos dados são importantes, e use broadcast quando a disseminação rápida é mais importante que a entrega garantida a todos.

Boas práticas e observações finais

Ao trabalhar com canais assíncronos, considere as seguintes boas práticas:

  • Sempre trate os erros de send() e recv() (por exemplo, quando o receptor foi dropado).
  • Use buffers de tamanho adequado: muito pequeno causa bloqueios frequentes, muito grande pode consumir memória desnecessária.
  • Para broadcast, monitore o erro Lagged e decida se deseja reconectar ou ignorar.
  • Evite usar mpsc para comunicação de resposta; prefira oneshot para evitar complexidade.

Dominar os canais assíncronos é fundamental para construir aplicações concorrentes eficientes em Rust. Pratique com os exercícios abaixo para consolidar o conhecimento.

Exercícios

  1. Crie um programa que use mpsc com dois produtores enviando 5 mensagens cada, e um consumidor que imprime todas as mensagens recebidas. Use um buffer de capacidade 3.

    ✓ Resposta:
    use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(3);

    for id in 0..2 {
        let tx = tx.clone();
        tokio::spawn(async move {
            for i in 0..5 {
                tx.send(format!("produtor {} msg {}", id, i)).await.unwrap();
            }
        });
    }
    drop(tx); // fecha o canal original

    while let Some(msg) = rx.recv().await {
        println!("{}", msg);
    }
}

  2. Implemente uma função assíncrona que receba um oneshot::Receiver e aguarde o valor, retornando-o. Depois, crie um canal oneshot, envie um valor e chame a função.

    ✓ Resposta:
    use tokio::sync::oneshot;

async fn aguardar_valor(rx: oneshot::Receiver<i32>) -> i32 {
    rx.await.expect("canal cancelado")
}

#[tokio::main]
async fn main() {
    let (tx, rx) = oneshot::channel();
    tx.send(42).unwrap();
    let valor = aguardar_valor(rx).await;
    println!("Valor recebido: {}", valor);
}

  3. Crie um sistema com broadcast onde três receptores estão inscritos. Envie 3 mensagens e faça cada receptor imprimir as mensagens que receber. Observe que todos recebem todas as mensagens.

    ✓ Resposta:
    use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, _) = broadcast::channel(16);
    let mut rx1 = tx.subscribe();
    let mut rx2 = tx.subscribe();
    let mut rx3 = tx.subscribe();

    for i in 0..3 {
        tx.send(i).unwrap();
    }

    for (nome, rx) in [("rx1", &mut rx1), ("rx2", &mut rx2), ("rx3", &mut rx3)] {
        println!("{}:", nome);
        for _ in 0..3 {
            match rx.recv().await {
                Ok(val) => println!("  {}", val),
                Err(e) => println!("  erro: {}", e),
            }
        }
    }
}

  4. Explique como o backpressure funciona em mpsc e escreva um exemplo onde o produtor é bloqueado porque o consumidor é lento.

    ✓ Resposta: O backpressure em mpsc é implementado pelo send().await: se o buffer está cheio, a operação aguarda até que o consumidor remova um item, liberando espaço. Exemplo:
    use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1);
    tokio::spawn(async move {
        for i in 0..5 {
            tx.send(i).await.unwrap();
            println!("enviado {}", i);
        }
    });

    sleep(Duration::from_millis(100)).await; // atraso
    while let Some(val) = rx.recv().await {
        println!("recebido {}", val);
        sleep(Duration::from_millis(200)).await;
    }
}

  5. Discuta as diferenças entre mpsc e broadcast em relação a backpressure e entrega de mensagens. Em que cenário você escolheria um em vez do outro?

    ✓ Resposta: mpsc tem backpressure: o produtor é bloqueado quando o buffer enche, garantindo que nenhuma mensagem seja perdida. broadcast não tem backpressure global: o produtor nunca é bloqueado, mas receptores lentos podem perder mensagens (erro Lagged). Escolha mpsc quando a entrega garantida e a ordem são críticas (ex.: processamento de jobs). Escolha broadcast quando você precisa notificar muitos consumidores sobre eventos e é aceitável que alguns percam mensagens se não conseguirem acompanhar (ex.: feeds de notícias).

Referências