Padrões de concorrência em Rust

Nesta aula, vamos mergulhar nos padrões de concorrência mais utilizados em Rust. Concorrência é a arte de executar múltiplas tarefas simultaneamente, e Rust oferece ferramentas poderosas e seguras para isso, como threads, canais e tipos como Arc e Mutex. Dominar esses padrões é essencial para escrever sistemas eficientes, desde servidores web até processamento de dados em larga escala.

Vamos explorar quatro padrões fundamentais: worker pool, pipeline, fan-out/fan-in e coordenação. Cada um resolve um problema específico de organização do trabalho concorrente. Ao longo da aula, apresentarei exemplos completos e práticos, com explicações detalhadas, para que você possa aplicar esses conceitos em seus próprios projetos.

Worker pool

O padrão worker pool consiste em um conjunto fixo de threads (os workers) que ficam aguardando tarefas para executar. Em vez de criar uma thread para cada tarefa (o que é caro e pode levar à exaustão de recursos), criamos um número limitado de threads e distribuímos as tarefas por meio de uma fila compartilhada. Esse padrão é ideal para processamento de tarefas independentes e intensivas, como requisições HTTP ou processamento de imagens.

Em Rust, implementamos um worker pool usando canais: um canal de transmissão (sender) para enviar tarefas e um canal de resposta (receiver) para coletar resultados. Os workers ficam em loop, aguardando mensagens no receptor. Para parar os workers, usamos uma mensagem especial ou fechamos o canal. Vamos ver um exemplo prático:

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

fn main() {
    const NUM_WORKERS: usize = 4;
    let (tx, rx) = mpsc::channel::<String>();
    let mut workers = Vec::new();

    // Criar workers
    for id in 0..NUM_WORKERS {
        let rx = rx.clone();
        workers.push(thread::spawn(move || loop {
            match rx.recv() {
                Ok(task) => {
                    println!("Worker {} processando: {}", id, task);
                    thread::sleep(Duration::from_millis(100)); // simula trabalho
                }
                Err(_) => break, // canal fechado, encerra
            }
        }));
    }

    // Enviar tarefas
    for i in 0..10 {
        tx.send(format!("tarefa {}", i)).unwrap();
    }
    drop(tx); // fecha o canal, fazendo os workers saírem do loop

    // Aguardar workers
    for worker in workers {
        worker.join().unwrap();
    }
    println!("Todos os workers concluíram.");
}

No código acima, criamos 4 workers que compartilham o mesmo receptor. Note que rx é clonado para cada worker, pois o receptor pode ser clonado (ao contrário do sender). Enviamos 10 tarefas e, em seguida, descartamos o tx original para fechar o canal. Quando não há mais mensagens, recv retorna erro e os workers encerram.

Esse padrão é eficiente porque o número de threads é limitado, evitando sobrecarga de criação e troca de contexto. Em aplicações reais, você pode usar crates como rayon para gerenciar pools automaticamente, mas entender a implementação manual é crucial para casos específicos.

Pipeline

O padrão pipeline divide um processo em estágios, onde cada estágio executa uma transformação nos dados e os passa para o próximo. Cada estágio pode rodar em uma thread separada, e os dados fluem por canais. Isso é útil para processamento de streams de dados, como leitura de arquivos, transformação e escrita.

Em Rust, implementamos pipelines criando uma cadeia de canais. Cada estágio tem um receptor do canal anterior e um transmissor para o próximo. O primeiro estágio produz os dados iniciais, e o último consome o resultado final. Vamos ver um exemplo simples que transforma números: primeiro duplica, depois soma 10 e por fim imprime.

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

fn main() {
    let (tx1, rx1) = mpsc::channel::<i32>();
    let (tx2, rx2) = mpsc::channel::<i32>();

    // Estágio 1: produz números de 0 a 9
    let produtor = thread::spawn(move || {
        for i in 0..10 {
            tx1.send(i).unwrap();
        }
        drop(tx1); // encerra o estágio
    });

    // Estágio 2: duplica
    let duplicador = thread::spawn(move || {
        while let Ok(valor) = rx1.recv() {
            tx2.send(valor * 2).unwrap();
        }
        drop(tx2);
    });

    // Estágio 3: soma 10 e imprime
    let consumidor = thread::spawn(move || {
        while let Ok(valor) = rx2.recv() {
            println!("Resultado: {}", valor + 10);
        }
    });

    produtor.join().unwrap();
    duplicador.join().unwrap();
    consumidor.join().unwrap();
}

No exemplo, cada thread executa um estágio. O produtor envia os números, o duplicador os transforma e o consumidor os imprime. Observe que usamos drop para fechar o canal quando o estágio termina, sinalizando ao próximo que não há mais dados.

Pipelines são excelentes para paralelizar processamento em fluxo, pois cada estágio pode trabalhar em um item enquanto outros processam itens diferentes. No entanto, é preciso cuidado com gargalos: se um estágio for mais lento, os canais podem acumular dados. Em Rust, os canais são ilimitados (ou podem ser limitados com crates como crossbeam), então é possível controlar o backpressure.

Fan-out/fan-in

O padrão fan-out/fan-in envolve distribuir tarefas para múltiplos workers (fan-out) e depois coletar os resultados em um ponto central (fan-in). Isso é útil quando você tem um conjunto de tarefas independentes que podem ser executadas em paralelo e os resultados precisam ser agregados.

Em Rust, implementamos fan-out usando um canal para enviar tarefas a vários workers (cada um em uma thread) e outro canal para os workers enviarem os resultados de volta ao coletor. O coletor aguarda todos os resultados e os processa. Vamos ver um exemplo que calcula o quadrado de números em paralelo:

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

fn main() {
    const NUM_WORKERS: usize = 4;
    let (task_tx, task_rx) = mpsc::channel::<i32>();
    let (result_tx, result_rx) = mpsc::channel::<i32>();

    // Worker threads
    for _ in 0..NUM_WORKERS {
        let task_rx = task_rx.clone();
        let result_tx = result_tx.clone();
        thread::spawn(move || {
            while let Ok(n) = task_rx.recv() {
                result_tx.send(n * n).unwrap();
            }
        });
    }
    drop(task_rx); // não precisamos mais do receptor no main
    drop(result_tx); // não precisamos mais do transmissor no main

    // Enviar tarefas
    for i in 0..10 {
        task_tx.send(i).unwrap();
    }
    drop(task_tx); // fecha o canal de tarefas

    // Coletar resultados
    let mut resultados = Vec::new();
    while let Ok(res) = result_rx.recv() {
        resultados.push(res);
    }

    println!("Resultados: {:?}", resultados);
}

Aqui, clonamos task_rx para cada worker, e cada worker envia o resultado pelo mesmo canal result_tx (clonado). No main, após enviar todas as tarefas, fechamos o canal de tarefas. Os workers, ao receberem erro de recv, encerram e o canal de resultados é fechado quando todos os workers terminarem (pois os transmissores clonados são destruídos). Assim, o loop de coleta termina.

Esse padrão é muito usado em computação paralela, como em sistemas de busca ou análise de dados. Uma alternativa mais ergonômica é usar o crate rayon, que oferece par_iter e map/reduce para fan-out/fan-in automático.

Coordenação

Coordenação refere-se a sincronizar threads para que elas não interfiram umas nas outras e para que eventos ocorram na ordem correta. Em Rust, os mecanismos de coordenação incluem Mutex, RwLock, Condvar, Barrier e Atomic. Eles são essenciais para evitar corridas de dados e deadlocks.

Vamos explorar um exemplo de coordenação com Barrier, que faz com que um grupo de threads espere até que todas cheguem a um ponto comum. Isso é útil em algoritmos de sincronização por fases, como em simulações.

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

fn main() {
    let barrier = Arc::new(Barrier::new(3)); // 3 threads
    let mut handles = vec![];

    for id in 0..3 {
        let barrier = Arc::clone(&barrier);
        handles.push(thread::spawn(move || {
            println!("Thread {} iniciou", id);
            thread::sleep(std::time::Duration::from_millis(100 * id));
            barrier.wait(); // espera as outras
            println!("Thread {} passou da barreira", id);
        }));
    }

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

No exemplo, cada thread dorme um tempo diferente (100ms, 200ms, 300ms). A barreira garante que nenhuma thread imprima "passou da barreira" até que todas tenham chamado wait. Isso coordena o início da próxima fase.

Outro mecanismo importante é o Condvar, que permite que threads esperem por uma condição. Combinado com Mutex, é usado para implementar filas de trabalho com espera condicional. Por exemplo, um worker pode dormir até que uma tarefa seja adicionada à fila, em vez de ficar em polling.

Vejamos um exemplo de fila com Mutex e Condvar:

use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::collections::VecDeque;

fn main() {
    let pair = Arc::new((Mutex::new(VecDeque::new()), Condvar::new()));
    let pair_producer = Arc::clone(&pair);
    let pair_consumer = Arc::clone(&pair);

    // Produtor
    let producer = thread::spawn(move || {
        let (lock, cvar) = &*pair_producer;
        for i in 0..5 {
            let mut queue = lock.lock().unwrap();
            queue.push_back(i);
            println!("Produzido: {}", i);
            cvar.notify_one(); // acorda um consumidor
            drop(queue);
            thread::sleep(std::time::Duration::from_millis(50));
        }
    });

    // Consumidor
    let consumer = thread::spawn(move || {
        let (lock, cvar) = &*pair_consumer;
        for _ in 0..5 {
            let mut queue = lock.lock().unwrap();
            while queue.is_empty() {
                queue = cvar.wait(queue).unwrap();
            }
            let item = queue.pop_front().unwrap();
            println!("Consumido: {}", item);
        }
    });

    producer.join().unwrap();
    consumer.join().unwrap();
}

Nesse código, o consumidor espera no Condvar quando a fila está vazia, liberando o Mutex enquanto espera. O produtor adiciona itens e notifica o consumidor. Isso evita que o consumidor fique em loop ativo (busy waiting).

Coordenação é um tópico sutil, e erros podem levar a deadlocks ou corridas. Rust ajuda em tempo de compilação para evitar corridas de dados (com o sistema de ownership e borrow), mas deadlocks são lógicos e precisam ser evitados com cuidado na ordem de aquisição de locks.

Boas práticas e observações finais

Ao trabalhar com concorrência em Rust, siga estas diretrizes:

• Prefira Arc para compartilhar dados imutáveis entre threads, e Mutex ou RwLock para dados mutáveis.
• Use canais para comunicação entre threads; eles são seguros e evitam compartilhamento de memória.
• Evite unwrap() em produção; trate erros adequadamente.
• Considere usar crates como rayon para paralelismo de dados, e crossbeam para canais com múltiplos produtores/consumidores.
• Teste exaustivamente com cargo test e, se possível, use ferramentas como loom para verificar concorrência.

Esses padrões são a base de sistemas concorrentes robustos. Pratique implementando variações e combinando-os para resolver problemas complexos.

Referências

• The Rust Book: Concurrency
• Documentação oficial de threads
• Documentação de canais mpsc
• Documentação de Mutex
• Documentação de Condvar
• Documentação de Barrier
• Rayon crate
• Crossbeam crate

Exercícios

1. Implemente um worker pool com 4 threads que processe uma lista de números, calculando o fatorial de cada um. Os resultados devem ser coletados e impressos.

Ver resposta

✓ Resposta:

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

fn fatorial(n: u64) -> u64 {
    (1..=n).product()
}

fn main() {
    let (tx, rx) = mpsc::channel::<u64>();
    let (res_tx, res_rx) = mpsc::channel::<(u64, u64)>();
    let mut workers = vec![];

    for _ in 0..4 {
        let rx = rx.clone();
        let res_tx = res_tx.clone();
        workers.push(thread::spawn(move || {
            while let Ok(n) = rx.recv() {
                let result = fatorial(n);
                res_tx.send((n, result)).unwrap();
            }
        }));
    }
    drop(rx);
    drop(res_tx);

    for i in 1..=10 {
        tx.send(i).unwrap();
    }
    drop(tx);

    let mut results = vec![];
    while let Ok((n, r)) = res_rx.recv() {
        results.push((n, r));
    }
    results.sort();
    for (n, r) in results {
        println!("{}! = {}", n, r);
    }

    for w in workers {
        w.join().unwrap();
    }
}

2. Crie um pipeline de 3 estágios: o primeiro gera números de 0 a 100, o segundo filtra os pares, e o terceiro imprime a raiz quadrada (como f64). Use canais.

Ver resposta

✓ Resposta:

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

fn main() {
    let (tx1, rx1) = mpsc::channel::<i32>();
    let (tx2, rx2) = mpsc::channel::<i32>();

    let gerador = thread::spawn(move || {
        for i in 0..=100 {
            tx1.send(i).unwrap();
        }
        drop(tx1);
    });

    let filtro = thread::spawn(move || {
        while let Ok(n) = rx1.recv() {
            if n % 2 == 0 {
                tx2.send(n).unwrap();
            }
        }
        drop(tx2);
    });

    let consumidor = thread::spawn(move || {
        while let Ok(n) = rx2.recv() {
            println!("sqrt({}) = {:.2}", n, (n as f64).sqrt());
        }
    });

    gerador.join().unwrap();
    filtro.join().unwrap();
    consumidor.join().unwrap();
}

3. Implemente um algoritmo fan-out/fan-in para calcular a soma dos quadrados de 1 a 100, distribuindo em 8 workers.

Ver resposta

✓ Resposta:

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

fn main() {
    const NUM_WORKERS: usize = 8;
    let (task_tx, task_rx) = mpsc::channel::<i32>();
    let (result_tx, result_rx) = mpsc::channel::<i64>();

    for _ in 0..NUM_WORKERS {
        let task_rx = task_rx.clone();
        let result_tx = result_tx.clone();
        thread::spawn(move || {
            while let Ok(n) = task_rx.recv() {
                let n = n as i64;
                result_tx.send(n * n).unwrap();
            }
        });
    }
    drop(task_rx);
    drop(result_tx);

    for i in 1..=100 {
        task_tx.send(i).unwrap();
    }
    drop(task_tx);

    let mut sum = 0;
    while let Ok(sq) = result_rx.recv() {
        sum += sq;
    }
    println!("Soma dos quadrados: {}", sum);
}

4. Use uma barreira para sincronizar 5 threads que devem imprimir seus IDs em duas fases: primeiro todos imprimem "Fase 1", depois todos imprimem "Fase 2".

Ver resposta

✓ Resposta:

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

fn main() {
    let barrier = Arc::new(Barrier::new(5));
    let mut handles = vec![];

    for id in 0..5 {
        let barrier = Arc::clone(&barrier);
        handles.push(thread::spawn(move || {
            println!("Thread {}: Fase 1", id);
            barrier.wait();
            println!("Thread {}: Fase 2", id);
        }));
    }

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

5. Crie uma fila de trabalho com Mutex e Condvar: um produtor adiciona 10 itens, e 2 consumidores processam (apenas imprimem) até a fila esvaziar. Garanta que os consumidores não façam polling.

Ver resposta

✓ Resposta:

use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::collections::VecDeque;

fn main() {
    let pair = Arc::new((Mutex::new(VecDeque::new()), Condvar::new()));

    // Produtor
    let prod_pair = Arc::clone(&pair);
    let producer = thread::spawn(move || {
        let (lock, cvar) = &*prod_pair;
        for i in 0..10 {
            let mut queue = lock.lock().unwrap();
            queue.push_back(i);
            println!("Produzido: {}", i);
            cvar.notify_all();
        }
    });

    // Consumidores
    let mut consumers = vec![];
    for id in 0..2 {
        let cons_pair = Arc::clone(&pair);
        consumers.push(thread::spawn(move || {
            let (lock, cvar) = &*cons_pair;
            loop {
                let mut queue = lock.lock().unwrap();
                while queue.is_empty() {
                    queue = cvar.wait(queue).unwrap();
                }
                let item = queue.pop_front().unwrap();
                drop(queue);
                println!("Consumidor {} processou: {}", id, item);
                if item == 9 {
                    break; // último item
                }
            }
        }));
    }

    producer.join().unwrap();
    for c in consumers {
        c.join().unwrap();
    }
}