Coopera Sharp - Chamadas em Sequência

Olá!

Este é mais um post da seção CooperaSharp e, desta vez, vamos falar um pouco sobre paralelismo e desempenho.

Vamos lá!

O Problema

O desafio propõe a melhoria de um conjunto de chamadas a uma API para a obtenção de certos dados.

O código do problema está replicado abaixo:

public class IntegradorService
{
    private readonly IApiClienteA _clienteA;
    private readonly IApiClienteB _clienteB;
    private readonly IApiClienteC _clienteC;

    public IntegradorService(
        IApiClienteA clienteA,
        IApiClienteB clienteB,
        IApiClienteC clienteC)
    {
        _clienteA = clienteA;
        _clienteB = clienteB;
        _clienteC = clienteC;
    }

    public async Task<List<Dado>> ObterDados()
    {
        var dados = new List<Dado>();

        var dadosA = await _clienteA.BuscarDados(); // chamada pode demorar
        dados.AddRange(dadosA);

        var dadosB = await _clienteB.BuscarDados(); // chamada pode falhar
        dados.AddRange(dadosB);

        var dadosC = await _clienteC.BuscarDados(); // chamada pode ser lenta
        dados.AddRange(dadosC);

        return dados;
    }
}

Repare no seguinte: o código acima faz uma chamada para uma dada API, obtém seu resultado, adiciona-o a uma lista e faz a chamada seguinte.

Ou seja, o tempo total da execução do método ObterDados vai ser igual à soma do tempo de execução de cada chamada externa. Portanto, se cada chamada levar 1s para ser executada, teremos 3s de tempo total de execução.

Ruim. Certo?

Há mais um detalhe: o código acima é tremendamente otimista! Se alguma exceção ocorrer no meio do caminho ela não apenas não é tratada como é lançada para o método chamador, o que comprometeria todo o processo de obtenção dos dados.

A Solução: Paralelismo!

Algo que podemos fazer em caso semelhante é paralelizar as chamadas às API. Para fazer isso existem duas opções, que exploraremos abaixo:

Task.WhenAll

A forma mais simples de fazer diversas chamadas e aguardar sua execução é criando novas threads, uma para cada chamada, unindo-as ao final de suas respectivas execuções para extrair o resultado desejado.

Para isso, vamos transformar as chamadas em Task e realizar todas as chamadas simultaneamente, desta forma:

(...)
var requests = new Task<IEnumerable<WeatherForecast>>[] { serverErrorClient.GetWeatherForecasts(),
                                                          serverSlowClient.GetWeatherForecasts(),
                                                          serverTimeoutClient.GetWeatherForecasts() };

try
{
    var results = await Task.WhenAll(requests);
    var data = results.SelectMany(i => i);

    return Ok(new { Data = data });
}
catch (AggregateException ex)
{
    //Handle inner exceptions on ex
    foreach (exception in ex.InnerExceptions)
    (...)
}
(...)

Repare que acima criamos um array de tasks para abrigar as chamadas e que, para obter seus resultados invocamos o método Task.WhenAll(...).

Mas o que este método faz?

A ideia é que, a partir de uma lista de tarefas, aqui representada pelo nosso array, as mesmas sejam chamadas em paralelo, limitando o tempo de espera por aquela que mais demorar. Esse método também é interessante porque, caso haja exceções lançadas pelas tasks as mesmas serão agregadas em um tipo chamado AggregateException, integrando sua propriedade InnerExceptions. Desta forma é possível registrá-las em log ou manipulá-las, decidindo como o fluxo seguirá.

Parallel.ForEachAsync

Existe uma outra forma de obter nossos dados fazendo chamadas em paralelo para as API, o Parallel.ForEachAsync. O código seria o seguinte, nos baseando no código acima:

(...)
var requests = new Task<IEnumerable<WeatherForecast>>[] { serverErrorClient.GetWeatherForecasts(),
                                                          serverSlowClient.GetWeatherForecasts(),
                                                          serverTimeoutClient.GetWeatherForecasts() };

try
{
    ConcurrentBag<IEnumerable<WeatherForecast>> results = [];
    var cts = new CancellationTokenSource();

    await Parallel.ForEachAsync(requests, 
                                new ParallelOptions 
                                { 
                                    MaxDegreeOfParallelism = requests.Lenght,
                                    CancellationToken = cts.Token;
                                },
                                async (request, ct) => 
                                {
                                    ct.ThrowIfCancellationRequested();

                                    var result = await request;
                                    results.Add(result);
                                }).WithAggregateException();

    var data = results.SelectMany(i => i);

    return Ok(new { Elapsed = stopwatch.ElapsedMilliseconds, Data = data });
}
catch (Exception ex)
{
    //Handle ex

    return StatusCode(StatusCodes.Status500InternalServerError);
}
(...)

Repare que aqui temos um capacidade adicional em relação ao Task.WhenAll, que é a possibilidade de estabelecer o número máximo de cores que serão usados, em vez de threads do threadpool, via ParallelOptions { MaxDegreeOfParallelism = requests.Lengh }. O tamanho da nossa lista de requisições não foi informado à toa, ele indica exatamente o número de cores -- repare, cores, não threads! -- que esperamos abrir, solicitadas para tornar as chamadas possíveis.

Ao final de cada requisição adicionaremos o resultado a uma ConcurrentBag, uma coleção threadsafe que armazenará todos os resultados.

Perceba que o comportamento é um tanto diferente to Task.WhenAll. Enquanto este aguarda a execução de todas as tasks para retornar o resultado de seu processamento, o Parallel.ForEachAsync executa a função recebida como parâmetro tão logo uma tarefa seja executada. Ainda assim, a instrução seguinte ao Parallel.ForEachAsync será executada apenas quando todas as tarefas forem concluídas.

Uma outra diferença está na opção WithAggregateException. Enquanto para o Task.WhenAll a lança uma AggregateException por padrão, o Parallel.ForEachAsync precisa ser informado. Não informar vai fazer com que a primeira exceção ocorrida no processamento paralelo seja lançada e o fluxo seja interrompido.

Qual opção usar?

A recomendação é que o Task.WhenAll seja utilizado como primeira escolha sempre que possível. É um mecanismo suficiente para a maioria dos casos e uma solução pouco verbosa.
Já o Parallel.ForEachAsync é uma escolha para quando houver precupação com o controle do uso de recursos ou houver cores disponíveis para executar as tarefas em paralelo, o que é mais performático que alternar entre threads em um mesmo core, sendo uma solução mais completa e um tanto mais performática, ainda que ao custo de maior verbosidade.

Conclusão

Processamento paralelo é um recurso muito interessante para casos como o do desafio, onde é importante processar diversas tarefas no menor tempo possível.

É uma bela alternativa para utlizar melhor certos recursos computacionais, como o threadpool e os cores do processador, de modo a aumentar o desempenho de sua aplicação.

Gostou? Me deixe saber pelos indicadores ou por minhas redes sociais.

Restou alguma dúvida? Fique à vontade para deixá-la nos comentários.

Muito obrigado por ler até aqui, e até a próxima!