Fila Limitada em Zig: Mutex, Condition e Backpressure

Uma fila limitada e thread-safe é uma das formas mais simples de tornar concorrência previsível em Zig. Produtores colocam tarefas na fila, workers retiram essas tarefas, e um limite fixo impede que um pico de entrada consuma memória até derrubar o processo. Quando a fila enche, o produtor espera: isso é backpressure aplicado no ponto em que o excesso nasce.

Em Zig, você pode implementar esse padrão com std.Thread.Mutex e std.Thread.Condition, sem alocação por item e sem uma estrutura lock-free difícil de auditar. A regra central é: todo estado compartilhado fica protegido pelo mutex; consumidores esperam enquanto a fila está vazia; produtores esperam enquanto ela está cheia; e o shutdown usa broadcast para acordar todas as threads bloqueadas.

Este guia constrói a estrutura, explica por que cada while existe e mostra como encaixá-la em pools de workers, pipelines de arquivos e serviços. Para revisar os fundamentos antes, veja concorrência avançada em Zig e a referência de std.Thread.

Resposta rápida: o desenho correto

O mutex não existe para proteger apenas o array. Ele protege o predicado inteiro: “a fila está vazia?”, “há espaço?”, “o sistema foi fechado?”. A Condition permite dormir sem ficar consultando esse predicado em um loop que desperdiça CPU.

Implementação de uma bounded queue genérica

A estrutura abaixo usa um ring buffer de capacidade conhecida em tempo de compilação. Ela não aloca memória no heap e funciona com qualquer tipo que possa ser copiado para dentro do array.

const std = @import("std");

pub fn BoundedQueue(comptime T: type, comptime capacity: usize) type {
    if (capacity == 0) @compileError("capacity deve ser maior que zero");

    return struct {
        const Self = @This();

        items: [capacity]T = undefined,
        head: usize = 0,
        tail: usize = 0,
        len: usize = 0,
        closed: bool = false,

        mutex: std.Thread.Mutex = .{},
        not_empty: std.Thread.Condition = .{},
        not_full: std.Thread.Condition = .{},

        pub fn push(self: *Self, item: T) !void {
            self.mutex.lock();
            defer self.mutex.unlock();

            while (self.len == capacity and !self.closed) {
                self.not_full.wait(&self.mutex);
            }

            if (self.closed) return error.QueueClosed;

            self.items[self.tail] = item;
            self.tail = (self.tail + 1) % capacity;
            self.len += 1;

            self.not_empty.signal();
        }

        pub fn pop(self: *Self) ?T {
            self.mutex.lock();
            defer self.mutex.unlock();

            while (self.len == 0 and !self.closed) {
                self.not_empty.wait(&self.mutex);
            }

            if (self.len == 0 and self.closed) return null;

            const item = self.items[self.head];
            self.head = (self.head + 1) % capacity;
            self.len -= 1;

            self.not_full.signal();
            return item;
        }

        pub fn close(self: *Self) void {
            self.mutex.lock();
            defer self.mutex.unlock();

            if (self.closed) return;
            self.closed = true;

            self.not_empty.broadcast();
            self.not_full.broadcast();
        }

        pub fn count(self: *Self) usize {
            self.mutex.lock();
            defer self.mutex.unlock();
            return self.len;
        }
    };
}

O ring buffer evita mover todos os elementos depois de cada pop. head indica o próximo item a sair, tail indica a próxima posição de entrada, e o operador % capacity faz os índices voltarem ao início do array.

len é necessário porque head == tail sozinho é ambíguo: pode significar fila vazia ou fila completamente cheia. Como todos os quatro campos mutáveis são acessados sob o mesmo mutex, não precisamos transformá-los em atomics.

Por que wait sempre fica dentro de while

Este padrão está correto:

while (self.len == 0 and !self.closed) {
    self.not_empty.wait(&self.mutex);
}

Este parece equivalente, mas está errado:

if (self.len == 0 and !self.closed) {
    self.not_empty.wait(&self.mutex);
}

Quando wait é chamado, a condition libera o mutex e coloca a thread para dormir de forma coordenada. Ao acordar, a thread precisa recuperar o mutex antes de retornar. Nesse intervalo, outra consumidora pode ter acordado primeiro e retirado o único item disponível. Se usarmos if, a segunda consumidora continua como se ainda houvesse trabalho. Com while, ela verifica novamente len e volta a esperar.

O loop também protege contra despertares espúrios, nos quais uma thread pode retornar de wait sem que uma operação lógica correspondente tenha ocorrido. A condition é um mecanismo de notificação; a fonte da verdade continua sendo o predicado sob o mutex.

signal para o fluxo normal, broadcast para shutdown

Depois de um push, existe pelo menos um item a mais. Acordar uma consumidora com not_empty.signal() costuma ser suficiente. Depois de um pop, existe pelo menos uma vaga a mais, então not_full.signal() acorda um produtor.

No encerramento, a situação muda. Pode haver vários produtores esperando por espaço e vários consumidores esperando por trabalho. Nenhum novo item ou vaga precisa aparecer: todos devem acordar apenas para observar closed. Por isso close chama:

self.not_empty.broadcast();
self.not_full.broadcast();

Sem os dois broadcasts, o programa pode travar no join: uma thread continua dormindo para sempre, embora o restante da aplicação já tenha decidido encerrar.

Exemplo com produtores e workers

Considere um serviço que recebe caminhos de arquivos e calcula relatórios. A fila guarda apenas descritores pequenos; cada worker faz o trabalho caro depois de retirar o item.

const std = @import("std");

const Job = struct {
    id: u64,
    path: []const u8,
};

const JobQueue = BoundedQueue(Job, 64);

fn worker(queue: *JobQueue, worker_id: usize) void {
    while (queue.pop()) |job| {
        // O mutex da fila já foi liberado aqui.
        processFile(job) catch |err| {
            std.debug.print(
                "worker {d}: job {d} falhou: {s}\n",
                .{ worker_id, job.id, @errorName(err) },
            );
        };
    }
}

fn processFile(job: Job) !void {
    std.debug.print("processando {d}: {s}\n", .{ job.id, job.path });
    // Abrir, ler, validar e persistir o resultado.
}

A propriedade mais importante aparece no comentário: o processamento acontece fora do mutex. O pop segura o lock apenas pelo tempo necessário para copiar um item e atualizar três inteiros. Se o worker mantivesse o mutex enquanto abre arquivo, consulta banco ou faz HTTP, a fila serializaria o sistema inteiro.

Uma sequência de ciclo de vida segura é:

1. inicializar a fila;
2. iniciar os workers;
3. iniciar ou executar os produtores;
4. parar de aceitar novas tarefas;
5. aguardar os produtores terminarem;
6. chamar queue.close();
7. deixar os workers drenarem os itens restantes;
8. fazer join em todos os workers.

Fechar a fila antes de os produtores terminarem faz push retornar error.QueueClosed. Isso pode ser exatamente o desejado durante cancelamento imediato, mas não é o shutdown gracioso que preserva trabalho já aceito. Para serviços que recebem SIGTERM, combine essa ordem com o padrão de graceful shutdown em Zig.

O que colocar dentro da fila

Guardar um []const u8 na fila não copia os bytes apontados. A fila copia apenas a fatia — ponteiro e tamanho. Portanto, a memória do texto precisa continuar válida até o worker terminar.

Há três estratégias comuns:

Evite enfileirar uma fatia que aponta para um buffer local reutilizado pelo produtor. O código pode passar em testes leves e corromper dados quando os workers atrasarem. O contrato de ownership precisa dizer quem aloca, quem libera e até quando o ponteiro permanece válido — a mesma disciplina do guia de alocação de memória em Zig.

Backpressure não significa necessariamente bloquear

O push deste exemplo bloqueia enquanto a fila está cheia. Isso é adequado para pipelines internos em que o produtor pode desacelerar. Em outros sistemas, bloquear é uma política ruim.

Um servidor HTTP talvez deva responder 503 Service Unavailable quando sua fila chega ao limite. Um coletor de métricas pode descartar amostras de baixa prioridade. Uma aplicação interativa pode esperar apenas alguns milissegundos. Um pipeline financeiro pode persistir a tarefa antes de reconhecer o recebimento.

Na prática, vale oferecer operações distintas:

• push: espera até haver espaço ou a fila fechar;
• tryPush: retorna imediatamente error.QueueFull;
• pushTimeout: espera até um prazo;
• pop: espera trabalho ou fechamento;
• tryPop: retorna imediatamente quando estiver vazia.

A estrutura de dados é a mesma; o que muda é o contrato operacional. Não esconda essa decisão. “Fila cheia” é um estado normal de sobrecarga, não uma exceção impossível.

Como escolher a capacidade

Não escolha capacity = 10_000 apenas para fazer o problema desaparecer em testes. Uma fila enorme troca backpressure por latência e memória acumulada.

Comece estimando:

capacidade ≈ taxa de pico tolerada × tempo de absorção desejado

Se entram 200 tarefas por segundo, os workers processam 180, e você quer absorver um pico por cinco segundos, uma reserva de aproximadamente 100 itens cobre a diferença inicial. Depois valide com carga real.

Observe pelo menos quatro métricas:

• tamanho atual da fila;
• percentual de tempo perto da capacidade;
• tempo que cada item passa esperando;
• quantidade de rejeições ou timeouts no produtor.

Fila sempre vazia pode indicar workers sobrando. Fila sempre cheia indica consumidores insuficientes, tarefa lenta ou capacidade usada para mascarar saturação. O guia de observabilidade em Zig mostra como expor contadores e histogramas sem poluir a regra de negócio.

Testes que encontram bugs de concorrência

Um teste útil não verifica apenas se um item entra e sai. Ele força os estados de espera e encerramento.

test "fila preserva ordem em uma unica thread" {
    var queue: BoundedQueue(u32, 3) = .{};

    try queue.push(10);
    try queue.push(20);
    try queue.push(30);

    try std.testing.expectEqual(@as(?u32, 10), queue.pop());
    try std.testing.expectEqual(@as(?u32, 20), queue.pop());
    try std.testing.expectEqual(@as(?u32, 30), queue.pop());

    queue.close();
    try std.testing.expectEqual(@as(?u32, null), queue.pop());
}

test "push falha depois do fechamento" {
    var queue: BoundedQueue(u8, 1) = .{};
    queue.close();

    try std.testing.expectError(error.QueueClosed, queue.push(1));
}

Depois adicione um teste concorrente com vários produtores, vários consumidores e uma soma verificável. Gere IDs únicos, conte quantos foram consumidos e confirme que nenhum sumiu ou apareceu duas vezes. Repita o teste centenas de vezes em CI quando ele for barato.

Também teste o cenário que costuma travar produção: fila vazia, vários workers bloqueados e close chamado sem adicionar outro item. Todos devem acordar e terminar. Veja mais padrões em testes completos em Zig e use builds Debug ou ReleaseSafe durante investigação.

Armadilhas comuns

Usar if em vez de while

É a falha clássica. O programa funciona com uma thread e quebra quando duas disputam a mesma notificação.

Processar dentro do lock

Copie ou mova o item, atualize a fila, sinalize e solte o mutex. Trabalho caro não pertence à região crítica.

Ler len sem sincronização

Até uma métrica precisa adquirir o mutex ou usar uma representação atômica separada. Uma leitura “só para log” ainda é acesso concorrente ao estado.

Destruir a fila antes dos joins

A fila contém mutexes, conditions e memória observada pelas threads. Seu tempo de vida precisa superar o de todos os produtores e consumidores.

Fechar sem acordar os dois lados

Consumidores esperam em not_empty; produtores, em not_full. O shutdown precisa acordar ambos.

Tentar corrigir contenção com uma fila lock-free cedo demais

Estruturas lock-free exigem raciocínio sobre ordem de memória, ABA, reclamation e arquitetura de CPU. Elas podem ser necessárias em um hot path medido, mas são uma troca de complexidade, não uma evolução automática.

Mutex, canal ou estrutura lock-free?

Para a maioria dos serviços e ferramentas, comece com mutex e condition. Meça tempo bloqueado e throughput. Se a contenção for relevante, primeiro reduza a região crítica, use lotes ou particione filas. Só depois avalie lock-free.

Perguntas frequentes

Quando devo usar uma fila limitada em Zig?

Quando a produção pode superar temporariamente o consumo e você precisa limitar memória, latência acumulada ou trabalho pendente. É um bom padrão para processamento de arquivos, pools de workers, servidores e pipelines.

Por que Condition.wait precisa de um loop?

Porque a notificação não reserva o item para a thread acordada. Quando ela recupera o mutex, outra thread pode ter alterado o estado. O loop revalida o predicado e cobre despertares espúrios.

signal ou broadcast?

Use signal no fluxo normal, quando uma operação disponibiliza um item ou uma vaga. Use broadcast no shutdown, porque todas as threads bloqueadas precisam observar que a fila foi fechada.

Como evitar perda de tarefas ao encerrar?

Pare os produtores, espere que terminem, feche a fila, deixe os consumidores drenarem os itens restantes e só então faça join. pop deve retornar null apenas quando a fila estiver fechada e vazia.

Devo usar uma fila lock-free?

Somente depois de medir contenção relevante e confirmar que particionamento, lotes ou uma região crítica menor não resolvem. Mutex e condition entregam um contrato muito mais fácil de revisar.

Quantos workers devo criar?

Para trabalho intensivo de CPU, comece perto do número de núcleos disponíveis. Para I/O bloqueante, pode haver mais workers, mas o limite deve vir de medição e dos recursos externos, como conexões, descritores e capacidade do banco.

Conclusão

Uma fila limitada com std.Thread.Mutex e std.Thread.Condition resolve três problemas ao mesmo tempo: coordena produtores e consumidores, limita o uso de memória e transforma saturação em uma política explícita de backpressure.

O padrão seguro é pequeno: proteja todo o estado com um mutex, espere conditions dentro de while, faça o mínimo possível sob o lock, use signal no fluxo normal e broadcast no fechamento. Depois trate ownership, capacidade, métricas e ordem de shutdown como partes do design — não como detalhes posteriores.

Essa base já é suficiente para um pool de workers confiável. O próximo passo é adicionar cancelamento, timeouts e métricas conforme a necessidade real do sistema, mantendo a fila simples e testável. Para aprofundar, continue por processos filhos com timeout em Zig e error handling em Zig.