Foreach paralelo com lambda assíncrono

138

Gostaria de lidar com uma coleção em paralelo, mas estou tendo problemas para implementá-la e, portanto, espero por alguma ajuda.

O problema surge se eu quiser chamar um método marcado como assíncrono em C #, dentro do lambda do loop paralelo. Por exemplo:

var bag = new ConcurrentBag<object>();
Parallel.ForEach(myCollection, async item =>
{
  // some pre stuff
  var response = await GetData(item);
  bag.Add(response);
  // some post stuff
}
var count = bag.Count;

O problema ocorre com a contagem sendo 0, porque todos os threads criados são efetivamente apenas threads de segundo plano e a Parallel.ForEachchamada não espera pela conclusão. Se eu remover a palavra-chave assíncrona, o método será assim:

var bag = new ConcurrentBag<object>();
Parallel.ForEach(myCollection, item =>
{
  // some pre stuff
  var responseTask = await GetData(item);
  responseTask.Wait();
  var response = responseTask.Result;
  bag.Add(response);
  // some post stuff
}
var count = bag.Count;

Funciona, mas desativa completamente a esperteza esperada e eu tenho que fazer um tratamento manual de exceções .. (Removido por questões de brevidade).

Como posso implementar um Parallel.ForEachloop, que usa a palavra-chave wait dentro do lambda? É possível?

O protótipo do método Parallel.ForEach usa um Action<T>parâmetro as, mas quero que ele aguarde minha lambda assíncrona.

clausndk
fonte
1
Eu suponho que você pretendia remover awaitdo await GetData(item)seu segundo bloco de código, pois isso produziria um erro de compilação como está.
Josh M.
2
Possível duplicação de aninhamento aguardam em Parallel.ForEach #
Vitaliy Ulantikov 02/17 /

Respostas:

186

Se você apenas deseja um paralelismo simples, pode fazer o seguinte:

var bag = new ConcurrentBag<object>();
var tasks = myCollection.Select(async item =>
{
  // some pre stuff
  var response = await GetData(item);
  bag.Add(response);
  // some post stuff
});
await Task.WhenAll(tasks);
var count = bag.Count;

Se você precisar de algo mais complexo, confira o ForEachAsyncpost de Stephen Toub .

Stephen Cleary
fonte
46
Provavelmente, é necessário um mecanismo de limitação. Isso criará imediatamente tantas tarefas quanto itens, que podem acabar em solicitações de rede de 10k e outras coisas.
usr
10
@usr O último exemplo do artigo de Stephen Toub trata disso.
svick
@svick Eu estava intrigado com a última amostra. Parece-me que apenas agrupa uma carga de tarefas para criar mais tarefas para mim, mas todas elas começam em massa.
Luke Puplett
2
@LukePuplett Cria doptarefas e cada uma delas processa algum subconjunto da coleção de entradas em série.
svick
4
@Afshin_Zavvar: Se você telefona Task.Runsem obter awaito resultado, isso é apenas um trabalho de disparar e esquecer no pool de threads. Isso quase sempre é um erro.
Stephen Cleary
74

Você pode usar o ParallelForEachAsyncmétodo de extensão do pacote NuGet do AsyncEnumerator :

using Dasync.Collections;

var bag = new ConcurrentBag<object>();
await myCollection.ParallelForEachAsync(async item =>
{
  // some pre stuff
  var response = await GetData(item);
  bag.Add(response);
  // some post stuff
}, maxDegreeOfParallelism: 10);
var count = bag.Count;
Serge Semenov
fonte
1
Este é o seu pacote? Eu já vi você postar isso em alguns lugares agora? : D Oh espera .. seu nome está no pacote: D +1
Piotr Kula
17
@ppumkin, sim, é meu. Eu vi esse problema e outra vez, por isso, decidiu resolvê-lo de maneira mais simples possíveis e gratuitos outras pessoas lutando também :)
Serge Semenov
Obrigado .. definitivamente faz sentido e me ajudou muito!
Piotr Kula
2
você tem um erro de digitação: maxDegreeOfParallelism>maxDegreeOfParalellism
Shiran Dror
3
A grafia correta é realmente maxDegreeOfParallelism, no entanto há algo no @ o comentário de ShiranDror - em seu pacote você chamou o maxDegreeOfParalellism variável por engano (e, portanto, seu código citado não irá compilar até ser alterada ..)
BornToCode
17

Com SemaphoreSlimvocê pode conseguir o controle do paralelismo.

var bag = new ConcurrentBag<object>();
var maxParallel = 20;
var throttler = new SemaphoreSlim(initialCount: maxParallel);
var tasks = myCollection.Select(async item =>
{
  try
  {
     await throttler.WaitAsync();
     var response = await GetData(item);
     bag.Add(response);
  }
  finally
  {
     throttler.Release();
  }
});
await Task.WhenAll(tasks);
var count = bag.Count;
Felipe l
fonte
3

Minha implementação leve do ParallelForEach assíncrono.

Recursos:

  1. Limitação (grau máximo de paralelismo).
  2. Tratamento de exceção (a exceção de agregação será lançada na conclusão).
  3. Memória eficiente (não há necessidade de armazenar a lista de tarefas).

public static class AsyncEx
{
    public static async Task ParallelForEachAsync<T>(this IEnumerable<T> source, Func<T, Task> asyncAction, int maxDegreeOfParallelism = 10)
    {
        var semaphoreSlim = new SemaphoreSlim(maxDegreeOfParallelism);
        var tcs = new TaskCompletionSource<object>();
        var exceptions = new ConcurrentBag<Exception>();
        bool addingCompleted = false;

        foreach (T item in source)
        {
            await semaphoreSlim.WaitAsync();
            asyncAction(item).ContinueWith(t =>
            {
                semaphoreSlim.Release();

                if (t.Exception != null)
                {
                    exceptions.Add(t.Exception);
                }

                if (Volatile.Read(ref addingCompleted) && semaphoreSlim.CurrentCount == maxDegreeOfParallelism)
                {
                    tcs.SetResult(null);
                }
            });
        }

        Volatile.Write(ref addingCompleted, true);
        await tcs.Task;
        if (exceptions.Count > 0)
        {
            throw new AggregateException(exceptions);
        }
    }
}

Exemplo de uso:

await Enumerable.Range(1, 10000).ParallelForEachAsync(async (i) =>
{
    var data = await GetData(i);
}, maxDegreeOfParallelism: 100);
nicolay.anykienko
fonte
2

Eu criei um método de extensão para isso, que usa o SemaphoreSlim e também permite definir o grau máximo de paralelismo

    /// <summary>
    /// Concurrently Executes async actions for each item of <see cref="IEnumerable<typeparamref name="T"/>
    /// </summary>
    /// <typeparam name="T">Type of IEnumerable</typeparam>
    /// <param name="enumerable">instance of <see cref="IEnumerable<typeparamref name="T"/>"/></param>
    /// <param name="action">an async <see cref="Action" /> to execute</param>
    /// <param name="maxDegreeOfParallelism">Optional, An integer that represents the maximum degree of parallelism,
    /// Must be grater than 0</param>
    /// <returns>A Task representing an async operation</returns>
    /// <exception cref="ArgumentOutOfRangeException">If the maxActionsToRunInParallel is less than 1</exception>
    public static async Task ForEachAsyncConcurrent<T>(
        this IEnumerable<T> enumerable,
        Func<T, Task> action,
        int? maxDegreeOfParallelism = null)
    {
        if (maxDegreeOfParallelism.HasValue)
        {
            using (var semaphoreSlim = new SemaphoreSlim(
                maxDegreeOfParallelism.Value, maxDegreeOfParallelism.Value))
            {
                var tasksWithThrottler = new List<Task>();

                foreach (var item in enumerable)
                {
                    // Increment the number of currently running tasks and wait if they are more than limit.
                    await semaphoreSlim.WaitAsync();

                    tasksWithThrottler.Add(Task.Run(async () =>
                    {
                        await action(item).ContinueWith(res =>
                        {
                            // action is completed, so decrement the number of currently running tasks
                            semaphoreSlim.Release();
                        });
                    }));
                }

                // Wait for all tasks to complete.
                await Task.WhenAll(tasksWithThrottler.ToArray());
            }
        }
        else
        {
            await Task.WhenAll(enumerable.Select(item => action(item)));
        }
    }

Uso da amostra:

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);
Jay Shah
fonte
'using' não ajudará. O loop foreach estará aguardando o sem fio indefinidamente. Tente este código simples que reproduz o problema: aguarde Enumerable.Range (1, 4) .ForEachAsyncConcurrent (async (i) => {Console.WriteLine (i); execute nova exceção ("exceção de teste");}, maxDegreeOfParallelism: 2);
Nicolay.anykienko
@ nicolay.anykienko você está certo sobre o # 2. Esse problema de memória pode ser resolvido adicionando taskWithThrottler.RemoveAll (x => x.IsCompleted);
askids
1
Eu tentei no meu código e, se maxDegreeOfParallelism não for nulo, os bloqueios de código. Aqui você pode ver todo o código a ser reproduzido: stackoverflow.com/questions/58793118/…
Massimo Savazzi