Concorrência: Threads, Travas e Dados Compartilhados Sem Corrupção

Duas threads que vendem o último exemplar do mesmo livro ao mesmo tempo podem, as duas, conseguir: a concorrência transforma certezas do código sequencial em apostas.

Em "Assincronia Avançada" terminamos com um importador que processa milhões de vendas numa única sequência, enquanto o computador tem vários núcleos parados. Distribuir o trabalho entre eles parece simples, mas esbarra num problema: os dados compartilhados. Nesta aula enfrentamos a concorrência de verdade, com várias threads executando código ao mesmo tempo e mexendo nos mesmos objetos. Veremos por que o código que funciona sequencialmente pode falhar em paralelo, as ferramentas do .NET para impedir isso e as estruturas que tornam o compartilhamento seguro por construção. O curso de C deste site enfrenta os mesmos problemas no artigo "Threads com pthreads". Aqui eles reaparecem, com ferramentas de nível mais alto.

Assincronia, concorrência e paralelismo

Três palavras que costumam ser confundidas:

• Assincronia é não bloquear enquanto se espera. Foi o assunto das duas aulas anteriores, e ela funciona até com uma única thread.
• Concorrência é ter várias tarefas em andamento no mesmo período, que podem se intercalar.
• Paralelismo é executar várias coisas literalmente ao mesmo tempo, em núcleos diferentes do processador.

Assincronia resolve espera por entrada e saída. Paralelismo resolve trabalho pesado de processador. E toda vez que existe paralelismo sobre dados compartilhados, surgem os problemas desta aula.

Threads e o pool

Uma thread é uma linha de execução independente, com sua própria pilha. Criar threads diretamente com new Thread(...) é possível, mas raramente é o certo: cada uma é cara de criar e ocupa memória. O .NET mantém um pool de threads reaproveitáveis, e Task.Run agenda trabalho nele:

Console.WriteLine($"Núcleos disponíveis: {Environment.ProcessorCount}");

Task t1 = Task.Run(() => Console.WriteLine($"Tarefa 1 na thread {Environment.CurrentManagedThreadId}"));
Task t2 = Task.Run(() => Console.WriteLine($"Tarefa 2 na thread {Environment.CurrentManagedThreadId}"));
await Task.WhenAll(t1, t2);

As duas tarefas podem rodar em threads diferentes, ao mesmo tempo, e a ordem das mensagens pode mudar de uma execução para outra. Essa não determinância é a primeira lição da concorrência: você não controla a ordem em que threads diferentes executam.

A condição de corrida

Vejamos o problema na sua forma mais pura. Um milhão de incrementos de um contador, distribuídos entre vários núcleos com Parallel.For:

int contador = 0;
Parallel.For(0, 1_000_000, _ => contador++);
Console.WriteLine(contador);   // quase nunca 1.000.000

O resultado costuma ficar abaixo de um milhão, e muda a cada execução. O motivo é que contador++ não é uma operação única. São três passos: ler o valor da memória, somar um e gravar o resultado. Se duas threads leem o mesmo valor (digamos, 41) antes de qualquer uma gravar, as duas somam e gravam 42. Dois incrementos aconteceram, e o contador só avançou um. Isso é uma condição de corrida: o resultado depende da ordem imprevisível em que as threads se intercalam.

O mais perigoso das condições de corrida é que elas são intermitentes. Com poucos dados, ou numa máquina com menos núcleos, o erro pode simplesmente não aparecer. Os testes passam, e a falha surge em produção, sob carga, uma vez a cada mil pedidos, impossível de reproduzir. Não dá para testar até que ela suma: é preciso raciocinar sobre o código.

Interlocked: operações indivisíveis

Para operações simples sobre um único valor, o processador oferece instruções atômicas, que leem, alteram e gravam como um passo indivisível. O .NET as expõe na classe Interlocked:

int contador = 0;
Parallel.For(0, 1_000_000, _ => Interlocked.Increment(ref contador));
Console.WriteLine(contador);   // sempre 1.000.000

Interlocked.Increment, Add, Exchange e CompareExchange são rápidos e corretos, mas só servem quando o estado compartilhado é um único valor. Quando uma operação envolve verificar uma condição e alterar mais de um dado, é preciso algo mais forte.

lock: exclusão mútua

A instrução lock garante que apenas uma thread por vez execute um trecho de código, a chamada seção crítica. As outras esperam na entrada:

private readonly Lock _trava = new();
private decimal _saldo;

public void Debitar(decimal valor)
{
    lock (_trava)
    {
        if (valor > _saldo)
            throw new InvalidOperationException("Saldo insuficiente.");
        _saldo -= valor;
    }
}

Dentro do lock, verificar e alterar formam um bloco indivisível: nenhuma outra thread consegue entrar no meio. O tipo System.Threading.Lock, introduzido no .NET 9 junto com o C# 13, é o objeto de trava moderno. Em código anterior, você vai ver lock usado sobre um private readonly object _trava = new();, com o mesmo significado. Em ambos os casos, a trava deve ser privada: travar sobre this ou sobre um objeto público permite que código de fora use a mesma trava e provoque bloqueios inesperados.

Três regras mantêm travas seguras:

1. Seções críticas curtas. Enquanto uma thread segura a trava, todas as outras que precisam dela ficam paradas. Faça dentro do lock só o indispensável.
2. Nunca chame código externo segurando uma trava. Um evento disparado de dentro de um lock executa código de terceiros que você não controla e que pode tentar pegar outra trava, ou até a mesma por outro caminho.
3. Nunca espere dentro de um lock. O compilador proíbe await dentro de um lock (erro CS1996), e com razão: a thread que retomaria o método depois da espera pode ser outra, e a trava ficaria presa a uma thread que já seguiu adiante. Para exclusão mútua em código assíncrono, use SemaphoreSlim(1, 1) com await WaitAsync() e Release() num finally, como na aula anterior.

Deadlock

Travas resolvem corridas, mas criam um risco novo. Imagine duas contas e duas transferências simultâneas, uma de A para B e outra de B para A. A primeira trava A e tenta travar B. A segunda já travou B e tenta travar A. Cada uma espera a trava que a outra segura, para sempre. É o deadlock que vimos com .Result em "async e await", agora na sua forma original.

A defesa clássica é adquirir travas sempre na mesma ordem. Se toda transferência travar primeiro a conta de menor identificador, a espera circular fica impossível. Melhor ainda é evitar precisar de duas travas ao mesmo tempo, redesenhando o código para que cada operação toque um único dado protegido.

Coleções concorrentes

Em vez de envolver coleções comuns em travas, o .NET oferece coleções desenhadas para acesso simultâneo, no namespace System.Collections.Concurrent. A mais usada é o ConcurrentDictionary<TKey, TValue>:

using System.Collections.Concurrent;

var unidades = new ConcurrentDictionary<int, int>();
Parallel.ForEach(vendas, v =>
    unidades.AddOrUpdate(v.ProdutoId, v.Quantidade, (_, atual) => atual + v.Quantidade));

AddOrUpdate insere o valor se a chave não existir, ou aplica a função de atualização se existir, sem corrupção mesmo com várias threads. Há também GetOrAdd, TryRemove e TryUpdate. Um Dictionary comum usado assim poderia não só perder atualizações, mas corromper sua estrutura interna e entrar em estados impossíveis.

Outras coleções concorrentes: ConcurrentQueue<T> (fila), ConcurrentStack<T> e ConcurrentBag<T> (um saco sem ordem).

Channels: produtores e consumidores

Um padrão muito frequente em sistemas reais é o de produtor e consumidor: uma parte do sistema produz itens, como linhas lidas de um arquivo ou pedidos recebidos pela rede, e outras os processam. O System.Threading.Channels oferece uma fila assíncrona feita para isso:

using System.Threading.Channels;

var canal = Channel.CreateBounded<Venda>(capacity: 1_000);

// produtor
await canal.Writer.WriteAsync(venda);   // espera, sem bloquear, se o canal estiver cheio
canal.Writer.Complete();                // avisa que não virá mais nada

// consumidor
await foreach (Venda v in canal.Reader.ReadAllAsync())
    Processar(v);                       // termina sozinho quando o canal é completado e esvaziado

A capacidade limitada (bounded) é o detalhe mais importante. Se o produtor for mais rápido que os consumidores, a fila não cresce indefinidamente até esgotar a memória: quando ela enche, o WriteAsync espera até haver espaço. Isso se chama contrapressão (backpressure), e é o que mantém estável um sistema com partes de velocidades diferentes.

Paralelismo de dados: Parallel e PLINQ

Para trabalho pesado de processador sobre muitos itens independentes, Parallel.For e Parallel.ForEach distribuem as iterações entre os núcleos. O LINQ tem uma versão paralela, o PLINQ, ativada com AsParallel():

int primos = Enumerable.Range(2, 5_000_000)
    .AsParallel()
    .Count(EhPrimo);

Paralelismo não é gratuito. Dividir o trabalho, coordenar as threads e juntar os resultados custa tempo, e para itens baratos ou em pouca quantidade a versão paralela pode ser mais lenta que a sequencial. Ela compensa quando cada item exige trabalho real e os itens são independentes. E o ganho nunca é exatamente proporcional ao número de núcleos: há partes que não se dividem, e os núcleos disputam memória e cache. Sempre meça.

Imutabilidade: a trava que não precisa existir

Todos esses problemas têm uma causa comum: estado mutável compartilhado. Se os dados compartilhados nunca mudam, não há corrida possível, e nenhuma trava é necessária. É por isso que a imutabilidade, que perseguimos desde os records de "Classe, Struct ou Record" e as coleções imutáveis de "Sequências Sob Demanda", é uma ferramenta de concorrência. O Catalogo imutável que construímos pode ser lido por mil threads ao mesmo tempo sem nenhum cuidado. Quem quer "alterá-lo" cria um novo, e publicar a referência nova é uma única atribuição. Quando possível, prefira desenhar para não compartilhar estado mutável a administrar travas sobre ele.

Juntando tudo: estoque seguro e uma importação paralela

Até aqui, o Vender de LivroImpresso supunha uma única thread. Veja o que acontece quando mil clientes tentam comprar ao mesmo tempo o mesmo livro, que tem 100 exemplares:

var livro = new LivroImpresso("Dom Casmurro", "Machado de Assis", 59.90m, pesoGramas: 350, estoque: 100);
int vendidas = 0, recusadas = 0;

Parallel.For(0, 1_000, _ =>
{
    try
    {
        livro.Vender(1);
        Interlocked.Increment(ref vendidas);
    }
    catch (EstoqueInsuficienteException)
    {
        Interlocked.Increment(ref recusadas);
    }
});

Console.WriteLine($"Vendidas: {vendidas}, recusadas: {recusadas}, estoque final: {livro.Estoque}");

Com a versão da aula de eventos, o resultado é imprevisível: às vezes mais de 100 vendas "bem-sucedidas", às vezes estoque final diferente do esperado. O if (quantidade > Estoque) e o Estoque -= quantidade são dois passos separados, e duas threads podem passar pela verificação com o último exemplar antes que qualquer uma o desconte. É a condição de corrida do tipo "verificar e depois agir", e num sistema real ela significa vender um livro que não existe. A correção protege os dois passos juntos:

public sealed class LivroImpresso : Produto
{
    private readonly Lock _trava = new();

    // ... demais membros ...

    public override void Vender(int quantidade)
    {
        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(quantidade);

        int restante;
        lock (_trava)
        {
            if (quantidade > Estoque)
                throw new EstoqueInsuficienteException(Titulo, Estoque, quantidade);
            Estoque -= quantidade;
            restante = Estoque;
        }

        if (restante <= LimiteEstoqueBaixo)
            EstoqueBaixo?.Invoke(this, new EstoqueBaixoEventArgs(this, restante));   // fora da trava
    }

    public void Repor(int quantidade)
    {
        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(quantidade);
        lock (_trava)
            Estoque += quantidade;
    }
}

Agora o resultado é sempre o mesmo: 100 vendidas, 900 recusadas, estoque final zero. Repare no detalhe que segue a regra 2: o valor restante é copiado para uma variável local dentro da trava, e o evento é disparado fora dela. Os inscritos no evento, como o monitor de estoque da aula de memória, executam sem que a trava do livro esteja presa.

Por fim, a importação de vendas ganha paralelismo com um canal: uma tarefa lê o arquivo e escreve no canal, e quatro consumidores processam as vendas, acumulando num dicionário concorrente:

using System.Collections.Concurrent;
using System.Threading.Channels;

using var cts = new CancellationTokenSource();
var canal = Channel.CreateBounded<Venda>(capacity: 10_000);
var unidadesPorProduto = new ConcurrentDictionary<int, int>();

Task produtor = Task.Run(async () =>
{
    try
    {
        await foreach (Venda venda in LerVendasAsync("vendas.csv", cts.Token))
            await canal.Writer.WriteAsync(venda, cts.Token);
    }
    finally
    {
        canal.Writer.TryComplete();   // os consumidores precisam saber que acabou, mesmo em caso de erro
    }
});

Task[] consumidores = Enumerable.Range(0, 4).Select(_ => Task.Run(async () =>
{
    await foreach (Venda venda in canal.Reader.ReadAllAsync(cts.Token))
        unidadesPorProduto.AddOrUpdate(venda.ProdutoId, venda.Quantidade, (_, atual) => atual + venda.Quantidade);
})).ToArray();

await produtor;
await Task.WhenAll(consumidores);

foreach (var (produtoId, unidades) in unidadesPorProduto.OrderByDescending(p => p.Value).Take(5))
    Console.WriteLine($"Produto {produtoId}: {unidades} unidades");

O TryComplete no finally é essencial. Se o produtor falhasse sem completar o canal, os consumidores esperariam para sempre por itens que nunca viriam. O canal limitado impede que a leitura do disco, mais rápida, encha a memória com vendas ainda não processadas. E o dicionário concorrente garante que as somas não se percam quando dois consumidores atualizam o mesmo produto ao mesmo tempo.

Nesta aula, e em quase todo o curso até aqui, o código que escrevemos roda inteiramente dentro do mundo gerenciado do .NET. Mas há situações em que a melhor ferramenta para um trabalho já existe, testada e otimizada há décadas, escrita em C: bibliotecas de compressão, de criptografia, de processamento de imagens, as APIs do próprio sistema operacional e até o motor de banco de dados que usaremos na próxima fase. O C# consegue chamar esse código diretamente, e para isso precisa descer ao nível em que ponteiros, layouts de memória e a responsabilidade de liberar recursos voltam a ser seus.

Fontes e leituras recomendadas

• Práticas recomendadas de threading gerenciado (Microsoft Learn): https://learn.microsoft.com/dotnet/standard/threading/managed-threading-best-practices. Traz regras sobre travas, deadlocks e dados compartilhados.
• Instrução lock: https://learn.microsoft.com/dotnet/csharp/language-reference/statements/lock. É a referência da instrução e do tipo System.Threading.Lock.
• Coleções thread-safe: https://learn.microsoft.com/dotnet/standard/collections/thread-safe/. É a visão geral de ConcurrentDictionary e das demais coleções concorrentes.
• System.Threading.Channels: https://learn.microsoft.com/dotnet/core/extensions/channels. Explica canais limitados e ilimitados, produtores e consumidores e contrapressão.
• Programação paralela no .NET: https://learn.microsoft.com/dotnet/standard/parallel-programming/. Cobre Parallel, PLINQ e os critérios para decidir quando o paralelismo compensa.
• Joe Duffy, Concurrent Programming on Windows (Addison-Wesley, 2008): a referência mais profunda sobre concorrência na plataforma .NET, escrita por um dos arquitetos da biblioteca de tarefas.

Exercícios

Exercício 1

Rode o contador de um milhão de incrementos com Parallel.For três vezes, primeiro com contador++, depois com Interlocked.Increment e por fim com lock. Anote os resultados e os tempos, e explique as diferenças.

using System.Diagnostics;

int contador = 0;
var relogio = Stopwatch.StartNew();
Parallel.For(0, 1_000_000, _ => contador++);
Console.WriteLine($"Sem proteção: {contador} em {relogio.ElapsedMilliseconds} ms");

contador = 0; relogio.Restart();
Parallel.For(0, 1_000_000, _ => Interlocked.Increment(ref contador));
Console.WriteLine($"Interlocked:  {contador} em {relogio.ElapsedMilliseconds} ms");

contador = 0; relogio.Restart();
var trava = new Lock();
Parallel.For(0, 1_000_000, _ => { lock (trava) contador++; });
Console.WriteLine($"lock:         {contador} em {relogio.ElapsedMilliseconds} ms");

Exercício 2

O método abaixo transfere valores entre contas e trava em produção de vez em quando. Explique o deadlock e corrija-o.

static void Transferir(Conta origem, Conta destino, decimal valor)
{
    lock (origem.Trava)
    {
        lock (destino.Trava)
        {
            origem.Debitar(valor);
            destino.Creditar(valor);
        }
    }
}

static void Transferir(Conta origem, Conta destino, decimal valor)
{
    Conta primeira = origem.Id < destino.Id ? origem : destino;
    Conta segunda = origem.Id < destino.Id ? destino : origem;

    lock (primeira.Trava)
    {
        lock (segunda.Trava)
        {
            origem.Debitar(valor);
            destino.Creditar(valor);
        }
    }
}

Exercício 3

Por que o compilador recusa o código abaixo? Reescreva-o de forma correta.

lock (_trava)
{
    string conteudo = await File.ReadAllTextAsync(caminho);
    _cache[caminho] = conteudo;
}

private readonly SemaphoreSlim _acesso = new(1, 1);

await _acesso.WaitAsync();
try
{
    string conteudo = await File.ReadAllTextAsync(caminho);
    _cache[caminho] = conteudo;
}
finally
{
    _acesso.Release();
}

Exercício 4

Conte os números primos abaixo de 5 milhões de forma sequencial e com PLINQ, medindo o tempo de cada forma. O ganho é proporcional ao número de núcleos? Por quê?

using System.Diagnostics;

static bool EhPrimo(int n)
{
    if (n < 2) return false;
    if (n % 2 == 0) return n == 2;
    for (int d = 3; (long)d * d <= n; d += 2)
        if (n % d == 0) return false;
    return true;
}

var relogio = Stopwatch.StartNew();
int seq = Enumerable.Range(0, 5_000_000).Count(EhPrimo);
Console.WriteLine($"Sequencial: {seq} primos em {relogio.ElapsedMilliseconds} ms");

relogio.Restart();
int par = Enumerable.Range(0, 5_000_000).AsParallel().Count(EhPrimo);
Console.WriteLine($"PLINQ:      {par} primos em {relogio.ElapsedMilliseconds} ms");

Exercício 5

Um cache usa _cache.GetOrAdd(chave, k => CarregarDoBanco(k)) num ConcurrentDictionary. Sob carga, os registros mostram que CarregarDoBanco às vezes é chamado duas vezes para a mesma chave. Explique por que e proponha uma correção.

private readonly ConcurrentDictionary<string, Lazy<Produto>> _cache = new();

public Produto Obter(string chave) =>
    _cache.GetOrAdd(chave, k => new Lazy<Produto>(() => CarregarDoBanco(k))).Value;