Nesta aula, vamos colocar em prática os conceitos de concorrência em Rust construindo um programa completo. Você vai aprender a escolher entre threads e async, a usar canais para comunicação entre tarefas, a compartilhar dados com segurança usando Arc e Mutex, e a coordenar a execução de múltiplas tarefas. O objetivo é que, ao final, você tenha um entendimento sólido de como criar programas concorrentes robustos e eficientes em Rust.

Vamos construir um exemplo realista: um pequeno sistema de processamento de tarefas que simula o download de arquivos, processamento e agregação de resultados. Usaremos threads para paralelismo e canais para comunicação, além de um estado compartilhado para estatísticas. Ao longo do caminho, discutiremos as decisões de design e as armadilhas comuns.

Threads ou async

A primeira decisão em qualquer programa concorrente é escolher entre threads e async/await. Threads são mais simples e adequadas para tarefas que envolvem I/O bloqueante ou CPU intensiva. Async é melhor para I/O não-bloqueante, como servidores web de alta concorrência. Em Rust, threads são fornecidas pela biblioteca padrão (std::thread), enquanto async requer um runtime como Tokio.

Para o nosso projeto, usaremos threads porque as tarefas são CPU-bound e a comunicação entre elas é mais natural com canais. Threads também são mais fáceis de entender para fins didáticos. Se precisássemos de alta escalabilidade com muitas conexões de rede, async seria a escolha.

use std::thread;

fn main() {
    let handle = thread::spawn(|| {
        println!("Olá da thread!");
    });
    handle.join().unwrap();
}

O exemplo acima mostra como criar uma thread simples. O método spawn recebe um closure que será executado na nova thread. join espera a thread terminar. Em nosso projeto, criaremos várias threads para processar tarefas em paralelo.

Canais

Canais são um mecanismo de comunicação entre threads. Em Rust, usamos o módulo std::sync::mpsc (multiple producer, single consumer). Um canal tem um remetente (Sender) e um receptor (Receiver). Podemos enviar valores do remetente e recebê-los no receptor.

Vamos usar canais para distribuir tarefas para as threads trabalhadoras e para coletar resultados. O padrão típico é criar um canal de tarefas e um canal de resultados. Cada thread trabalhadora recebe tarefas do canal de tarefas, processa e envia o resultado pelo canal de resultados.

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

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

    thread::spawn(move || {
        tx.send(42).unwrap();
    });

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

O método send envia um valor e recv bloqueia até receber. Também podemos usar try_recv para não bloquear. Em nosso projeto, usaremos send e recv para sincronização.

Compartilhamento seguro

Às vezes, precisamos compartilhar estado entre threads. Rust garante segurança através do sistema de tipos. Para compartilhar dados imutáveis, usamos Arc (contagem de referência atômica). Para dados mutáveis, combinamos Arc com Mutex ou RwLock.

No nosso projeto, vamos manter um contador de tarefas processadas e um total de tempo gasto. Para isso, usaremos Arc<Mutex<Stats>>. Cada thread trabalhadora terá uma referência clonada e atualizará as estatísticas com segurança.

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

#[derive(Default)]
struct Stats {
    processed: u32,
    total_time: u64,
}

fn main() {
    let stats = Arc::new(Mutex::new(Stats::default()));
    let mut handles = vec![];

    for _ in 0..4 {
        let stats = Arc::clone(&stats);
        handles.push(thread::spawn(move || {
            let mut stats = stats.lock().unwrap();
            stats.processed += 1;
            stats.total_time += 10;
        }));
    }

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

    let stats = stats.lock().unwrap();
    println!("Processadas: {}, tempo total: {}", stats.processed, stats.total_time);
}

O Mutex garante que apenas uma thread acesse os dados por vez, evitando corridas de dados. O Arc permite que múltiplas threads possuam a mesma referência sem cópias.

Coordenação

Coordenar tarefas é essencial para garantir que o programa funcione corretamente. Usamos join para esperar threads terminarem, canais para sincronizar, e barreiras (Barrier) para sincronizar várias threads em um ponto.

No nosso projeto, usaremos join para esperar todas as threads trabalhadoras terminarem antes de coletar os resultados finais. Também podemos usar um canal para sinalizar quando todas as tarefas foram processadas. Para algo mais complexo, como um pipeline, podemos usar barreiras.

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

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

    for i in 0..3 {
        let barrier = Arc::clone(&barrier);
        handles.push(thread::spawn(move || {
            println!("Thread {} antes da barreira", i);
            barrier.wait();
            println!("Thread {} depois da barreira", i);
        }));
    }

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

O Barrier faz com que cada thread espere até que todas as threads cheguem ao mesmo ponto. Isso é útil para fases de processamento.

Projeto completo: processador de tarefas

Vamos agora montar o projeto completo. O programa criará um conjunto de tarefas (números que representam trabalhos), distribuirá essas tarefas para um pool de threads trabalhadoras, cada thread processará a tarefa (simulando um atraso) e enviará o resultado de volta. Usaremos um canal para distribuir tarefas e outro para coletar resultados. As estatísticas serão compartilhadas com Arc<Mutex<Stats>>.

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

#[derive(Default)]
struct Stats {
    processed: u32,
    total_time: u64,
}

fn main() {
    let (task_tx, task_rx) = mpsc::channel();
    let (result_tx, result_rx) = mpsc::channel();
    let stats = Arc::new(Mutex::new(Stats::default()));

    let num_workers = 4;
    let mut handles = vec![];

    // Criar threads trabalhadoras
    for _ in 0..num_workers {
        let task_rx = task_rx.clone();
        let result_tx = result_tx.clone();
        let stats = Arc::clone(&stats);
        handles.push(thread::spawn(move || loop {
            match task_rx.recv() {
                Ok(task) => {
                    // Simula processamento
                    thread::sleep(Duration::from_millis(10 * task));
                    let result = task * 2;
                    result_tx.send(result).unwrap();

                    // Atualiza estatísticas
                    let mut stats = stats.lock().unwrap();
                    stats.processed += 1;
                    stats.total_time += 10 * task;
                }
                Err(_) => break, // canal fechado
            }
        }));
    }

    drop(task_rx); // Fecha o receptor na main
    drop(result_tx); // Fecha o remetente de resultados na main

    // Enviar tarefas
    let tasks = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
    for task in tasks {
        task_tx.send(task).unwrap();
    }
    drop(task_tx); // Fecha o canal de tarefas

    // Coletar resultados
    let mut results = vec![];
    for _ in 0..10 {
        let result = result_rx.recv().unwrap();
        results.push(result);
    }

    // Esperar threads terminarem
    for handle in handles {
        handle.join().unwrap();
    }

    // Imprimir resultados
    println!("Resultados: {:?}", results);
    let stats = stats.lock().unwrap();
    println!("Processadas: {}, tempo total: {}", stats.processed, stats.total_time);
}

Este programa cria 4 threads trabalhadoras que ficam em loop aguardando tarefas. A main envia 10 tarefas e depois coleta 10 resultados. As estatísticas são atualizadas concorrentemente. Note que usamos drop para fechar canais e sinalizar que não haverá mais envios, o que faz as threads saírem do loop.

Boas práticas

Ao escrever programas concorrentes, siga estas boas práticas:

  • Prefira canais a compartilhar estado com Mutex sempre que possível, pois canais são mais seguros e fáceis de raciocinar.
  • Quando precisar compartilhar estado, use Arc<Mutex<T>> e mantenha o escopo do lock o menor possível.
  • Evite deadlocks: nunca segure um lock enquanto tenta adquirir outro.
  • Use join para esperar todas as threads terminarem antes de acessar dados que elas podem ter modificado.
  • Teste com múltiplas execuções para garantir que não há corridas de dados.

Referências

Exercícios

  1. Explique a diferença entre threads e async em Rust e dê um exemplo de caso de uso para cada.

    ✓ Resposta: Threads são adequadas para tarefas CPU-bound ou I/O bloqueante, pois podem usar múltiplos núcleos. Async é melhor para I/O não-bloqueante, como servidores web, pois permite alta concorrência com poucas threads. Exemplo: threads para processamento de imagem, async para um servidor HTTP.

  2. Como você usaria um canal para enviar um resultado de uma thread para a main? Escreva um pequeno exemplo.

    ✓ Resposta: 
    use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        tx.send(10).unwrap();
    });
    let result = rx.recv().unwrap();
    println!("Resultado: {}", result);
}

  3. O que é Arc e por que é necessário para compartilhar dados entre threads? Explique com um exemplo.

    ✓ Resposta: Arc (Atomic Reference Counting) permite que múltiplas threads possuam a mesma referência a um dado, contando referências de forma atômica. Sem Arc, não seria possível compartilhar uma referência entre threads com segurança, pois o compilador não saberia quando liberar a memória. Exemplo: let data = Arc::new(5); e depois let data2 = Arc::clone(&data);.

  4. Como evitar deadlocks ao usar Mutex? Dê uma dica prática.

    ✓ Resposta: Evite segurar múltiplos locks ao mesmo tempo. Se precisar de mais de um, adquira-os sempre na mesma ordem em todas as threads. Além disso, mantenha o escopo do lock o menor possível, liberando-o antes de fazer outras operações.

  5. Modifique o projeto completo para usar 8 threads em vez de 4 e observe a saída. O que acontece com o tempo total? Explique.

    ✓ Resposta: O tempo total permanece o mesmo porque a simulação de processamento é determinística (cada tarefa leva um tempo fixo). O tempo de parede (wall-clock) diminui, pois as tarefas são processadas em paralelo, mas o tempo total somado é o mesmo. A saída mostrará que o número de processadas é 10 e o tempo total é a soma dos tempos individuais.