Multithreading & Fearless Concurrency
Módulo 02
Fila Concorrente Lock-Free (Algoritmo Michael-Scott)
Uma implementação de alto desempenho para sistemas paralelos onde múltiplas threads produzem e consomem dados simultaneamente sem sofrer bloqueios de Mutex, contenção no escalonador do SO (*context switches*) ou perigo de inversão de prioridades.
01. O Custo Oculto de Primitivas Bloqueantes (Mutex)
Em sistemas de alta vazão (*high-throughput*), primitivas como std::sync::Mutex delegam a resolução de contenção ao kernel do Sistema Operacional. Quando uma thread encontra um lock ocupado, o SO realiza uma troca de contexto (*context switch*), pausando a execução da thread e invalidando linhas de cache da CPU.
Vantagens das Estruturas Lock-Free:
- Progresso Global Garantido: Pelo menos uma thread sempre avança a cada intervalo de tempo, eliminando deadlocks.
- Imunidade a Interrupções: Se uma thread morrer ou sofrer preempção no meio da operação, as demais threads continuam operando normalmente.
- Execução Direta no Hardware: As operações ocorrem em nível de registradores de CPU através de instruções de máquina como
CMPXCHG (x86) ou LDREX/STREX (ARM).
02. Anatomia do Algoritmo Michael-Scott
Publicado em 1996 por Maged M. Michael e Michael L. Scott, este algoritmo resolve o principal problema de filas concorrentes: como enfileirar e desenfileirar simultaneamente sem que as duas operações entrem em conflito quando há poucos elementos?
1. O Nó Sentinela (Dummy Node)
A fila nunca fica estritamente com zero nós físicos. Ela sempre contém um nó artificial sentinela. O head sempre aponta para o sentinela, e o primeiro dado real está em head.next. Isso desacopla completamente os ponteiros de cabeça e cauda, permitindo que um produtor e um consumidor operem ao mesmo tempo sem tocar na mesma variável atômica.
2. Cooperação Entre Threads
A inserção ocorre em dois passos atômicos: primeiro liga o novo nó em tail.next, e depois avança tail. Se uma thread for interrompida entre esses dois passos, qualquer outra thread que tentar inserir detecta que a cauda está atrasada e a move para a frente cooperativamente antes de prosseguir.
03. Modelo de Consistência de Memória (Memory Ordering)
Em Rust, operações atômicas não usam apenas a ordem sequencial estrita (Ordering::SeqCst), que impõe barreiras de memória pesadas na CPU. Utilizamos a semântica de Acquire-Release:
| Atomic Ordering |
Comportamento de Hardware |
Uso na Fila |
| Ordering::Acquire |
Impede que leituras e escritas subsequentes sejam reordenadas para antes deste ponto na CPU. |
Ao carregar ponteiros de head, tail e next. |
| Ordering::Release |
Garante que todas as escritas anteriores estejam visíveis para outras threads que fizerem Acquire. |
No compare_exchange que publica o novo nó ou atualiza a cauda. |
| Ordering::Relaxed |
Garante apenas atomicidade do valor, sem impor barreiras de reordenação entre instruções adjacentes. |
Em casos de falha do CAS onde o loop recomeça imediatamente. |
04. O Problema ABA & Reclamação Segura de Memória (SMR)
⚠️ Consideração de Engenharia de Produção:
Em estruturas baseadas em CAS que desalocam memória no Heap, existe o risco do Problema ABA: uma thread lê o endereço 0x1000, outra thread desenfileira esse nó, o desaloca, aloca outro objeto no mesmo endereço 0x1000 e a primeira thread conclui o CAS achando que nada mudou.
Em bibliotecas de produção em Rust (como crossbeam), essa questão é resolvida com técnicas de Epoch-Based Memory Reclamation (EBR) ou Hazard Pointers, que adiam a liberação da memória até que nenhuma thread concorrente mantenha referências ativas àquele nó.
use std::ptr;
use std::sync::atomic::{AtomicPtr, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::Duration;
struct Node<T> {
value: Option<T>,
next: AtomicPtr<Node<T>>,
}
impl<T> Node<T> {
fn new(value: Option<T>) -> Self {
Self {
value,
next: AtomicPtr::new(ptr::null_mut()),
}
}
}
pub struct LockFreeQueue<T> {
head: AtomicPtr<Node<T>>,
tail: AtomicPtr<Node<T>>,
}
unsafe impl<T: Send> Send for LockFreeQueue<T> {}
unsafe impl<T: Send> Sync for LockFreeQueue<T> {}
impl<T> LockFreeQueue<T> {
pub fn new() -> Self {
let sentinel = Box::into_raw(Box::new(Node::new(None)));
Self {
head: AtomicPtr::new(sentinel),
tail: AtomicPtr::new(sentinel),
}
}
pub fn enqueue(&self, val: T) {
let new_node = Box::into_raw(Box::new(Node::new(Some(val))));
loop {
let tail = self.tail.load(Ordering::Acquire);
let next = unsafe { (*tail).next.load(Ordering::Acquire) };
if tail == self.tail.load(Ordering::Acquire) {
if next.is_null() {
if unsafe {
(*tail)
.next
.compare_exchange(
ptr::null_mut(),
new_node,
Ordering::Release,
Ordering::Relaxed,
)
.is_ok()
} {
let _ = self.tail.compare_exchange(
tail,
new_node,
Ordering::Release,
Ordering::Relaxed,
);
return;
}
} else {
let _ = self.tail.compare_exchange(
tail,
next,
Ordering::Release,
Ordering::Relaxed,
);
}
}
}
}
pub fn dequeue(&self) -> Option<T> {
loop {
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let next = unsafe { (*head).next.load(Ordering::Acquire) };
if head == self.head.load(Ordering::Acquire) {
if head == tail {
if next.is_null() {
return None;
}
let _ = self.tail.compare_exchange(
tail,
next,
Ordering::Release,
Ordering::Relaxed,
);
} else if !next.is_null() {
let val = unsafe { (*next).value.take() };
if self
.head
.compare_exchange(head, next, Ordering::Release, Ordering::Relaxed)
.is_ok()
{
unsafe {
let _ = Box::from_raw(head);
}
return val;
}
}
}
}
}
}
impl<T> Drop for LockFreeQueue<T> {
fn drop(&mut self) {
while self.dequeue().is_some() {}
let sentinel = self.head.load(Ordering::Relaxed);
if !sentinel.is_null() {
unsafe {
let _ = Box::from_raw(sentinel);
}
}
}
}
fn main() {
println!("============================================================");
println!("🦀 RUST CONCURRENCY: MICHAEL-SCOTT LOCK-FREE QUEUE");
println!("============================================================");
let queue = Arc::new(LockFreeQueue::<i32>::new());
let mut handles = vec![];
let producers_count = 4;
let items_per_producer = 25;
let consumers_count = 2;
println!("\n1. Inicializando carga de alta concorrência:");
println!(" -> {} Threads Produtoras gerando {} itens cada.", producers_count, items_per_producer);
println!(" -> {} Threads Consumidoras drenando concorrentemente.", consumers_count);
println!(" -> Total de operações atômicas programadas: 100 Enqueues.\n");
for prod_id in 0..producers_count {
let q_clone = Arc::clone(&queue);
let handle = thread::spawn(move || {
for i in 0..items_per_producer {
let val = (prod_id * 100) + i;
q_clone.enqueue(val);
if i % 10 == 0 {
thread::sleep(Duration::from_micros(10));
}
}
println!(" [Produtor #{}] finalizou o envio de {} itens.", prod_id, items_per_producer);
});
handles.push(handle);
}
let mut consumer_handles = vec![];
let total_consumed = Arc::new(std::sync::atomic::AtomicUsize::new(0));
for cons_id in 0..consumers_count {
let q_clone = Arc::clone(&queue);
let count_clone = Arc::clone(&total_consumed);
let handle = thread::spawn(move || {
let mut local_count = 0;
for _ in 0..60 {
while let Some(_val) = q_clone.dequeue() {
local_count += 1;
count_clone.fetch_add(1, Ordering::Relaxed);
}
thread::sleep(Duration::from_millis(1));
}
println!(" [Consumidor #{}] processou {} mensagens sem travas de Mutex!", cons_id, local_count);
});
consumer_handles.push(handle);
}
for h in handles {
h.join().unwrap();
}
for h in consumer_handles {
h.join().unwrap();
}
let mut remaining = 0;
while let Some(_) = queue.dequeue() {
remaining += 1;
total_consumed.fetch_add(1, Ordering::Relaxed);
}
println!("\n2. Auditoria e Integridade dos Dados:");
println!(" Total de itens processados com sucesso: {}", total_consumed.load(Ordering::SeqCst));
println!(" Itens residuais drenados: {}", remaining);
println!(" Status da Fila: {}", if queue.dequeue().is_none() { "Totalmente Esvaziada ✅" } else { "Inconsistente ❌" });
println!("\n3. Destruição segura de memória (Drop):");
println!(" Desalocando nó sentinela sem memory leaks ou locks de SO.");
}