En parallèle foreach avec asynchrone lambda

Je voudrais gérer une collection en parallèle, mais je vais avoir de la difficulté à le mettre en œuvre et je suis donc en espérant de l'aide.

Le problème se pose de savoir 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 avec le comte de 0, parce que tous les threads créés sont effectivement juste threads d'arrière-plan et la Parallel.ForEach d'appel de ne pas attendre 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;

Il fonctionne, mais il désactive complètement l'attendent à l'astuce et je dois faire quelques manuel de gestion des exceptions.. (Supprimé par souci de concision).

Comment puis-je mettre en œuvre un Parallel.ForEach boucle, qui utilise le mot clé await dans le lambda? Est-il possible?

Le prototype de la Parallèle.ForEach méthode prend un Action<T> comme paramètre, mais je veux attendre que mon asynchrone lambda.

  • Je suppose que vous vouliez supprimer await de await GetData(item) dans votre deuxième bloc de code comme il produirait une erreur de compilation comme-est.
  • Double Possible de Nidification attendent en Parallèle.ForEach
InformationsquelleAutor clausndk | 2013-02-28
  1. 129

    Si vous voulez juste simple parallélisme, 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 de quelque chose de plus complexe, découvrez Stephen Toub est ForEachAsync post.

    • Probablement un mécanisme de limitation est nécessaire. Cela va immédiatement créer autant de tâches qu'il existe des éléments qui pourraient se retrouver dans les 10k demandes de réseau et de ces.
    • Le dernier exemple en Stephen Toub de l'article de que.
    • Je me creusais la tête de plus que le dernier échantillon. Il me semble qu'il vient de lots d'une charge de tâches pour créer des tâches plus à moi, mais ils ont tous commencé à en-masse.
    • Il crée dop tâches et chacun d'eux traite ensuite des sous-ensembles de la collection d'entrée dans la série.
    • Limitation: stackoverflow.com/questions/10806951/...
    • J'ai été le tester votre code en passant une liste de nombres(de 1 à 10) et de l'écriture de chaque nombre dans la console. Tous les numéros imprimés ont été dans l'ordre, cependant quand j'ai changé attendent GetData à la Tâche.Exécuter(async () => attendent GetData (x)) les numéros imprimés ont été mélangées. Aucune idée de pourquoi ça ne fonctionne pas parallèle à la première tentative?
    • Si vous appelez Task.Run sans awaiting le résultat, alors que juste jetant le feu et oublier le travail sur le pool de threads. C'est presque toujours une erreur.

    InformationsquelleAutor Stephen Cleary
  2. 38

    Vous pouvez utiliser le ParallelForEachAsync méthode d'extension de AsyncEnumerator Package NuGet:

    using System.Collections.Async;

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;
    • Ceci est votre forfait? Je vous ai vu ce post dans quelques endroits aujourd'hui? 😀 Oh wait.. votre nom figure sur l'emballage 😀 +1
    • oui, c'est le mien. J'ai vu ce problème, encore et encore, il a donc décidé de le résoudre de la manière la plus simple possible et de libérer les autres de mal que de bien 🙂
    • Merci.. il fait vraiment bon sens et m'a aidé beaucoup de temps!
    • vous avez une faute de frappe: maxDegreeOfParallelism > maxDegreeOfParalellism
    • Je suis assez sûr qu'il est correctement orthographié
    • L'orthographe correcte est en effet maxDegreeOfParallelism, cependant il y a quelque chose dans @ShiranDror commentaire - dans votre forfait que vous avez appelé la variable maxDegreeOfParalellism par erreur (et donc de votre cité de code ne compile pas jusqu'à ce que vous le changer..)

    InformationsquelleAutor Serge Semenov
  3. 1

    J'ai créé une méthode d'extension pour ce qui rend l'utilisation de SemaphoreSlim et permet également de définir le 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);
    • 1. Liste<T> n'est pas thread-safe. 2. Pas optimale l'utilisation de la mémoire - à la fin tasksWithThrottler contiendra autant de tâches que les éléments en énumérable. 3. En cas d'exception sémaphore n'est pas libéré. La boucle sera bloqué pour toujours.
    • Ici, nous n'avons pas besoin de thread-safe Liste tasksWithThrottler n'est pas accessible à partir de plusieurs threads. Aussi, le sémaphore est déclarée avec l'aide de Lire la documentation de l'aide. en utilisant automatiquement les appels en disposer quand vous sortez de l'instruction à l'aide du champ - de sorte qu'en cas d'exception, il sera automatiquement libéré et rien ne sera bloqué.
    • "à l'aide" ne va pas aider. boucle foreach seront en attente pour semaphone indéfiniment. Juste essayer ce code simple qui reproduit le problème: attendre Énumérable.Range(1, 4).ForEachAsyncConcurrent(async (i) => { Console.WriteLine(i); throw new Exception("test d'exception"); }, maxDegreeOfParallelism: 2);
    • vous avez raison à propos de #2. Que la mémoire de problème peut être résolu par l'ajout d'tasksWithThrottler.RemoveAll(x => x.IsCompleted);
    InformationsquelleAutor Jay Shah
  4. 1

    Mon léger de la mise en œuvre de ParallelForEach async.

    Caractéristiques:

    1. D'étranglement (max degree of parallelism).
    2. La gestion des exceptions (agrégation exception sera levée à l'achèvement).
    3. Efficace en terme de mémoire (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);
    InformationsquelleAutor nicolay.anykienko