Assincronia Avançada: Muitas Tarefas, Cancelamento e Fluxos Assíncronos

Esperar uma coisa de cada vez é fácil. Os sistemas reais esperam dezenas de coisas ao mesmo tempo, desistem das que demoram demais e processam dados que chegam aos poucos.

Em "async e await" a Livraria fez sua primeira consulta a um serviço na internet, e sobrepusemos duas esperas iniciando as tarefas antes de aguardá-las. Terminamos com três perguntas em aberto: como esperar dez consultas de uma vez, como desistir de uma operação que o usuário não quer mais e como processar resultados que chegam aos poucos, linha a linha ou página a página. Esta aula responde às três, com as ferramentas que transformam o await básico num modelo completo para sistemas que conversam com o mundo.

Task.WhenAll: aguardar muitas tarefas

Para aguardar um conjunto de tarefas, Task.WhenAll devolve uma única tarefa que termina quando todas terminarem. Se as tarefas produzem valores, o resultado é um array, na mesma ordem das tarefas:

Cep[] ceps = [new("01001-000"), new("80010-000"), new("20040-020")];

Task<EnderecoViaCep?>[] tarefas = ceps.Select(cep => consulta.BuscarAsync(cep)).ToArray();
EnderecoViaCep?[] enderecos = await Task.WhenAll(tarefas);

for (int i = 0; i < ceps.Length; i++)
    Console.WriteLine($"{ceps[i]}: {enderecos[i]?.Localidade ?? "não encontrado"}");

O Select inicia as três consultas imediatamente, e o ToArray() materializa a sequência, garantindo que elas sejam iniciadas uma única vez. Lembre-se da execução adiada de "LINQ, Parte II": sem ele, percorrer a sequência de novo iniciaria as consultas de novo. O tempo total fica próximo da consulta mais lenta, e não da soma das três.

Quando uma ou mais tarefas falham, o comportamento merece atenção. O await sobre o WhenAll relança apenas a primeira exceção. As demais ficam guardadas na própria tarefa combinada, na propriedade Exception, um AggregateException que agrupa todas:

Task<EnderecoViaCep?[]> todas = Task.WhenAll(tarefas);
try
{
    await todas;
}
catch
{
    foreach (Exception erro in todas.Exception!.InnerExceptions)
        Console.WriteLine($"Falha: {erro.Message}");
}

Guardar a tarefa numa variável antes do await é o que permite inspecionar todas as falhas depois. Com await Task.WhenAll(...) direto, você só conheceria a primeira.

Task.WhenAny e limites de tempo

Task.WhenAny termina assim que qualquer uma das tarefas terminar, e devolve qual foi. Um uso clássico é consultar dois servidores equivalentes e ficar com a resposta mais rápida. Durante anos ele também foi usado para impor limites de tempo, mas hoje existe uma forma direta, WaitAsync:

try
{
    EnderecoViaCep? endereco = await consulta.BuscarAsync(cep).WaitAsync(TimeSpan.FromSeconds(3));
}
catch (TimeoutException)
{
    Console.WriteLine("Consulta demorou mais de 3 segundos.");
}

Há um detalhe importante aqui, e ele prepara a próxima seção: WaitAsync desiste de esperar, mas não interrompe a operação. A requisição HTTP continua em andamento, consumindo conexão e banda, e o resultado, quando chegar, será simplesmente ignorado. Para interromper a operação de verdade, é preciso pedir a ela que pare.

Cancelamento cooperativo

O .NET não mata tarefas à força. Interromper código no meio de uma operação qualquer poderia deixar arquivos pela metade, travas presas e estruturas corrompidas. Em vez disso, o cancelamento é cooperativo: quem quer cancelar sinaliza o pedido, e quem executa verifica esse sinal nos pontos em que é seguro parar.

Duas peças formam o mecanismo. O CancellationTokenSource é o controle remoto, que fica com quem pode decidir cancelar. O CancellationToken é o sinal, que é passado adiante para quem executa o trabalho:

using var cts = new CancellationTokenSource();
cts.CancelAfter(TimeSpan.FromSeconds(5));   // cancela sozinho depois de 5 segundos

try
{
    string conteudo = await File.ReadAllTextAsync(caminho, cts.Token);
}
catch (OperationCanceledException)
{
    Console.WriteLine("Operação cancelada.");
}

Praticamente todo método assíncrono da biblioteca aceita um CancellationToken como último parâmetro. Quando o sinal é acionado, eles interrompem o que estão fazendo e lançam OperationCanceledException (ou sua derivada, TaskCanceledException). Nos seus métodos, a convenção é a mesma: receber o token por último, com valor padrão, e repassá-lo a tudo o que for chamado:

public async Task<EnderecoViaCep?> BuscarAsync(Cep cep, CancellationToken cancellationToken = default)
{
    EnderecoViaCep? endereco = await http.GetFromJsonAsync<EnderecoViaCep>(
        $"ws/{cep.Valor}/json/", cancellationToken);
    return endereco?.Localidade is null ? null : endereco;
}

Em trabalho de processador, que não passa por nenhuma API assíncrona, você mesmo verifica o sinal de tempos em tempos:

for (int i = 0; i < itens.Count; i++)
{
    if (i % 1000 == 0)
        cancellationToken.ThrowIfCancellationRequested();
    Processar(itens[i]);
}

Verificar a cada mil itens, e não a cada item, equilibra a rapidez da resposta ao cancelamento com o custo da verificação.

Tokens podem ser combinados. CancellationTokenSource.CreateLinkedTokenSource(tokenA, tokenB) cria um sinal que dispara quando qualquer um dos dois disparar: por exemplo, "o usuário cancelou" ou "passou o tempo limite desta etapa". E num programa de console, o gesto natural de cancelar é o Ctrl+C, que pode ser ligado a um token:

using var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) =>
{
    e.Cancel = true;   // não encerra o processo imediatamente...
    cts.Cancel();      // ...em vez disso, pede às operações que parem com ordem
};

e.Cancel = true impede o encerramento abrupto do processo. Assim, as operações em andamento recebem o sinal, param num ponto seguro, os using e finally executam, os arquivos são fechados, e o programa termina de forma limpa. É o encerramento gracioso que o curso de C faz com sinais Unix, aqui com a mesma intenção e sem código de baixo nível.

Limitando a concorrência

Iniciar todas as tarefas de uma vez é ótimo para três consultas. Para três mil, é um problema: o serviço externo vai recusar ou bloquear seu endereço por excesso de requisições, e o próprio programa pode esgotar conexões. É preciso limitar quantas operações rodam ao mesmo tempo. A ferramenta clássica é o SemaphoreSlim, um contador de vagas:

static async Task<(Cep Cep, EnderecoViaCep? Endereco)[]> BuscarVariosAsync(
    ConsultaCep consulta, IEnumerable<Cep> ceps, int maximoSimultaneo, CancellationToken ct)
{
    using var vagas = new SemaphoreSlim(maximoSimultaneo);

    var tarefas = ceps.Select(async cep =>
    {
        await vagas.WaitAsync(ct);          // espera uma vaga, sem bloquear a thread
        try
        {
            return (cep, await consulta.BuscarAsync(cep, ct));
        }
        finally
        {
            vagas.Release();                // devolve a vaga, mesmo em caso de erro
        }
    });

    return await Task.WhenAll(tarefas);
}

Todas as tarefas são criadas, mas cada uma espera por uma vaga antes de fazer a requisição. Com maximoSimultaneo = 4, nunca há mais de quatro consultas em andamento. O finally garante a devolução da vaga: sem ele, uma consulta que falhasse levaria a vaga consigo, e o programa acabaria travado esperando vagas que nunca voltam.

O .NET também oferece uma forma pronta para o mesmo padrão, Parallel.ForEachAsync, com a opção MaxDegreeOfParallelism. Ela é conveniente quando você não precisa dos resultados em ordem.

Fluxos assíncronos: IAsyncEnumerable<T>

Em "Sequências Sob Demanda" escrevemos iteradores com yield return, que produzem elementos sob demanda. E se produzir cada elemento exigir esperar, por exemplo a próxima linha do disco ou a próxima página de uma API? Um iterador comum não pode usar await. Para isso existe o iterador assíncrono (C# 8), que combina as duas ideias:

static async IAsyncEnumerable<int> ContarDevagarAsync(int ate)
{
    for (int i = 1; i <= ate; i++)
    {
        await Task.Delay(200);   // simula uma espera por cada elemento
        yield return i;
    }
}

await foreach (int n in ContarDevagarAsync(5))
    Console.Write($"{n} ");

O método devolve IAsyncEnumerable<T> e pode ter tanto await quanto yield return. Quem consome usa await foreach, que aguarda cada elemento sem bloquear. É o encontro das duas máquinas de estados do compilador, a dos iteradores e a do async, num único recurso.

A biblioteca já oferece fluxos assíncronos prontos. O mais útil para nós é File.ReadLinesAsync, a versão assíncrona do ReadLines: lê um arquivo linha a linha, esperando o disco sem bloquear e respeitando cancelamento. E, a partir do .NET 10, os operadores do LINQ também estão disponíveis diretamente para IAsyncEnumerable<T> (Where, Select, Take e os demais), sem pacote extra.

Para que um iterador assíncrono aceite cancelamento, o parâmetro do token recebe o atributo [EnumeratorCancellation]. Assim, um token passado por quem consome, com .WithCancellation(token), chega até o iterador. Veremos isso no exemplo final.

Uma palavra sobre ValueTask

Você vai encontrar métodos que devolvem ValueTask<T> em vez de Task<T>. A diferença é de desempenho: uma Task é um objeto no heap, e um método chamado milhões de vezes, que quase sempre termina sem esperar (porque o dado já estava num cache, por exemplo), alocaria milhões de tarefas desnecessárias. ValueTask é uma struct que evita essa alocação no caso síncrono. O preço é um conjunto de regras: uma ValueTask só pode ser aguardada uma vez e não deve ser guardada para uso posterior. Na dúvida, use Task. Reserve ValueTask para caminhos muito frequentes, depois de medir.

Juntando tudo: um importador de vendas que pode ser interrompido

A Livraria agora importa o arquivo de vendas de forma assíncrona. As linhas são lidas sob demanda, interpretadas com o TentarLerVenda de "Span<T> e Memory<T>", e o processo inteiro pode ser interrompido com Ctrl+C sem deixar nada pela metade:

using System.Runtime.CompilerServices;

static async IAsyncEnumerable<Venda> LerVendasAsync(
    string caminho,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    int numeroDaLinha = 0;
    await foreach (string linha in File.ReadLinesAsync(caminho, cancellationToken))
    {
        numeroDaLinha++;
        if (string.IsNullOrWhiteSpace(linha))
            continue;

        if (TentarLerVenda(linha, out Venda? venda))
            yield return venda;
        else
            Console.Error.WriteLine($"Linha {numeroDaLinha} ignorada: {linha}");
    }
}

Repare que a interpretação com spans acontece dentro do laço, de forma síncrona, entre dois pontos de espera. É exatamente o cuidado exigido pela regra de "Span<T> e Memory<T>": um span nunca atravessa um await. A linha chega como string, é convertida para span apenas durante a chamada a TentarLerVenda, e o span deixa de existir antes da próxima espera.

O programa:

using var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };

var unidadesPorProduto = new Dictionary<int, int>();
int vendasLidas = 0;

try
{
    await foreach (Venda venda in LerVendasAsync("vendas.csv", cts.Token))
    {
        vendasLidas++;
        unidadesPorProduto[venda.ProdutoId] = unidadesPorProduto.GetValueOrDefault(venda.ProdutoId) + venda.Quantidade;

        if (vendasLidas % 100_000 == 0)
            Console.WriteLine($"{vendasLidas:N0} vendas processadas...");
    }
    Console.WriteLine($"Importação concluída: {vendasLidas:N0} vendas de {unidadesPorProduto.Count} produtos.");
}
catch (OperationCanceledException)
{
    Console.WriteLine($"Importação interrompida pelo usuário após {vendasLidas:N0} vendas.");
}

Um arquivo de milhões de linhas é processado com memória para uma linha por vez. O progresso aparece a cada cem mil vendas. E o Ctrl+C, em vez de matar o processo, faz o ReadLinesAsync parar na próxima leitura, o await foreach descartar o leitor de arquivo e o programa relatar até onde chegou.

O importador lê e processa as vendas uma de cada vez, numa única sequência. Com milhões de linhas, o disco entrega os dados mais rápido do que um único núcleo do processador consegue interpretá-los e agregá-los, e o computador tem vários núcleos parados. Distribuir o trabalho entre eles parece fácil: basta várias tarefas processando partes diferentes ao mesmo tempo. O problema é o dicionário unidadesPorProduto, que todas elas precisariam atualizar. Quando várias threads alteram o mesmo dado ao mesmo tempo, coisas estranhas acontecem, e elas não aparecem nos testes, aparecem em produção.

Fontes e leituras recomendadas

• Task.WhenAll (referência da API): https://learn.microsoft.com/dotnet/api/system.threading.tasks.task.whenall. Documenta o comportamento com resultados e com exceções múltiplas.
• Cancelamento em threads gerenciadas: https://learn.microsoft.com/dotnet/standard/threading/cancellation-in-managed-threads. Explica o modelo cooperativo, tokens combinados e boas práticas.
• Gerar e consumir fluxos assíncronos: https://learn.microsoft.com/dotnet/csharp/asynchronous-programming/generate-consume-asynchronous-stream. É o tutorial oficial de IAsyncEnumerable<T> e await foreach.
• SemaphoreSlim (referência da API): https://learn.microsoft.com/dotnet/api/system.threading.semaphoreslim. Traz o semáforo leve usado para limitar a concorrência.
• Stephen Toub, "Understanding the Whys, Whats, and Whens of ValueTask": https://devblogs.microsoft.com/dotnet/understanding-the-whys-whats-and-whens-of-valuetask/. É a explicação definitiva de quando ValueTask compensa.
• Stephen Cleary, Concurrency in C# Cookbook, 2ª edição (O'Reilly, 2019): receitas diretas para cada padrão desta aula e da próxima.

Exercícios

Exercício 1

Escreva um método que receba vários caminhos de arquivo e devolva o total de linhas de todos eles, lendo os arquivos ao mesmo tempo com Task.WhenAll.

static async Task<int> TotalDeLinhasAsync(IEnumerable<string> caminhos, CancellationToken ct = default)
{
    Task<string[]>[] leituras = caminhos
        .Select(c => File.ReadAllLinesAsync(c, ct))
        .ToArray();

    string[][] conteudos = await Task.WhenAll(leituras);
    return conteudos.Sum(linhas => linhas.Length);
}

Exercício 2

Qual a diferença entre limitar o tempo de uma consulta com WaitAsync(TimeSpan.FromSeconds(3)) e com um CancellationTokenSource configurado com CancelAfter(TimeSpan.FromSeconds(3)) e passado à consulta? Em que situação cada um é aceitável?

Exercício 3

Escreva um método que some os quadrados de 1 a 200 milhões e que possa ser cancelado, verificando o token a cada milhão de iterações. Chame-o com um token que cancela após 100 ms.

static long SomaDosQuadrados(long limite, CancellationToken ct)
{
    long soma = 0;
    for (long i = 1; i <= limite; i++)
    {
        if (i % 1_000_000 == 0)
            ct.ThrowIfCancellationRequested();
        soma += i * i % 1_000_007;   // módulo para não estourar o long
    }
    return soma;
}

using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(100));
try
{
    long resultado = await Task.Run(() => SomaDosQuadrados(200_000_000, cts.Token), cts.Token);
    Console.WriteLine(resultado);
}
catch (OperationCanceledException)
{
    Console.WriteLine("Cálculo cancelado.");
}

Exercício 4

Consuma o iterador ContarDevagarAsync(100) da aula de duas formas: parando depois de 3 elementos com break, e parando com um token cancelado depois de 1 segundo. O que acontece com o iterador em cada caso?

// Forma 1: break
int recebidos = 0;
await foreach (int n in ContarDevagarAsync(100))
{
    Console.Write($"{n} ");
    if (++recebidos == 3) break;
}

// Forma 2: token
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(1));
try
{
    await foreach (int n in ContarDevagarAsync(100).WithCancellation(cts.Token))
        Console.Write($"{n} ");
}
catch (OperationCanceledException)
{
    Console.WriteLine("(cancelado)");
}

Exercício 5

Três tarefas são aguardadas com Task.WhenAll, e duas delas falham com mensagens diferentes. Mostre o que um try/catch simples captura e como obter as duas mensagens.

Task a = Task.Run(() => throw new InvalidOperationException("falha A"));
Task b = Task.Delay(100);
Task c = Task.Run(() => throw new IOException("falha C"));

try
{
    await Task.WhenAll(a, b, c);
}
catch (Exception ex)
{
    Console.WriteLine($"Capturada: {ex.Message}");   // apenas uma: "falha A"
}

Task todas = Task.WhenAll(a, b, c);
try
{
    await todas;
}
catch
{
    foreach (Exception erro in todas.Exception!.InnerExceptions)
        Console.WriteLine($"Falha: {erro.GetType().Name}: {erro.Message}");   // as duas
}