Foreach parallèle avec lambda asynchrone

138

J'aimerais gérer une collection en parallèle, mais j'ai du mal à l'implémenter et j'espère donc avoir de l'aide.

Le problème survient si je veux appeler une méthode marquée async en C #, dans le lambda de la boucle parallèle. Par exemple:

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;

Le problème se produit lorsque le nombre est égal à 0, car tous les threads créés ne sont en fait que des threads d'arrière-plan et l' Parallel.ForEachappel n'attend pas la fin. Si je supprime le mot-clé async, la méthode ressemble à ceci:

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;

Cela fonctionne, mais cela désactive complètement l'intelligence d'attente et je dois faire un peu de gestion manuelle des exceptions. (Supprimé par souci de concision).

Comment puis-je implémenter une Parallel.ForEachboucle, qui utilise le mot clé await dans le lambda? C'est possible?

Le prototype de la méthode Parallel.ForEach prend un Action<T>paramètre as, mais je veux qu'il attende mon lambda asynchrone.

Clausndk
la source
1
Je suppose que vous vouliez supprimer awaitde await GetData(item)votre deuxième bloc de code car cela produirait une erreur de compilation en l'état.
Josh M.
2
Possibilité de duplication de Nesting en parallèle.ForEach
Vitaliy Ulantikov

Réponses:

188

Si vous voulez juste un parallélisme simple, vous pouvez le faire:

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;

Si vous avez besoin quelque chose de plus complexe, consultez Stephen Toub ForEachAsyncposte .

Stephen Cleary
la source
46
Un mécanisme d'étranglement est probablement nécessaire. Cela créera immédiatement autant de tâches qu'il y a d'éléments qui pourraient aboutir à des requêtes réseau de 10k et autres.
usr
10
@usr Le dernier exemple de l'article de Stephen Toub répond à cela.
svick
@svick J'étais perplexe sur ce dernier échantillon. Il me semble que cela ne fait que regrouper une charge de tâches pour créer plus de tâches pour moi, mais elles commencent toutes en masse.
Luke Puplett
2
@LukePuplett Il crée des doptâches et chacune d'elles traite ensuite un sous-ensemble de la collection d'entrée en série.
svick
4
@Afshin_Zavvar: Si vous appelez Task.Runsans obtenir awaitle résultat, cela ne fait que lancer un travail d'incendie et d'oubli sur le pool de threads. C'est presque toujours une erreur.
Stephen Cleary
74

Vous pouvez utiliser la ParallelForEachAsyncméthode d'extension du package NuGet 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
la source
1
Ceci est votre forfait? Je vous ai vu publier ceci dans quelques endroits maintenant? : D Oh attendez ... votre nom est sur l'emballage: D +1
Piotr Kula
17
@ppumkin, oui, c'est à moi. J'ai vu ce problème encore et encore, alors j'ai décidé de le résoudre de la manière la plus simple possible et de libérer les autres de la difficulté également :)
Serge Semenov
Merci ... cela a vraiment du sens et m'a beaucoup aidé!
Piotr Kula
2
vous avez une faute de frappe: maxDegreeOfParallelism>maxDegreeOfParalellism
Shiran Dror
3
L'orthographe correcte est en effet maxDegreeOfParallelism, mais il y a quelque chose dans le commentaire de @ ShiranDror - dans votre package, vous avez appelé la variable maxDegreeOfParalellism par erreur (et donc votre code cité ne se compilera pas tant que vous ne l'aurez pas modifié ..)
BornToCode
17

Avec SemaphoreSlimvous pouvez obtenir un contrôle de parallélisme.

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
la source
3

Mon implémentation légère de ParallelForEach async.

Fonctionnalités:

  1. Throttling (degré maximum de parallélisme).
  2. Gestion des exceptions (l'exception d'agrégation sera levée à la fin).
  3. Mémoire efficace (pas besoin de stocker la liste des tâches).

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);
        }
    }
}

Exemple d'utilisation:

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

J'ai créé une méthode d'extension pour cela qui utilise SemaphoreSlim et permet également de définir un degré maximal de parallélisme

    /// <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)));
        }
    }

Exemple d'utilisation:

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);
Jay Shah
la source
«utiliser» n'aidera pas. la boucle foreach attendra indéfiniment le sémaphone. Essayez simplement ce code simple qui reproduit le problème: wait Enumerable.Range (1, 4) .ForEachAsyncConcurrent (async (i) => {Console.WriteLine (i); throw new Exception ("test exception");}, maxDegreeOfParallelism: 2);
nicolay.anykienko
@ nicolay.anykienko vous avez raison sur # 2. Ce problème de mémoire peut être résolu en ajoutant tasksWithThrottler.RemoveAll (x => x.IsCompleted);
askids
1
Je l'ai essayé dans mon code et si je maxDegreeOfParallelism n'est pas nul, les blocages de code. Ici vous pouvez voir tout le code à reproduire: stackoverflow.com/questions/58793118/…
Massimo Savazzi