Parallel foreach mit asynchronem Lambda

136

Ich möchte eine Sammlung parallel bearbeiten, habe jedoch Probleme bei der Implementierung und hoffe daher auf Hilfe.

Das Problem tritt auf, wenn ich eine in C # als asynchron gekennzeichnete Methode innerhalb des Lambda der Parallelschleife aufrufen möchte. Beispielsweise:

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;

Das Problem tritt auf, wenn die Anzahl 0 ist, da alle erstellten Threads praktisch nur Hintergrund-Threads sind und der Parallel.ForEachAufruf nicht auf den Abschluss wartet. Wenn ich das asynchrone Schlüsselwort entferne, sieht die Methode folgendermaßen aus:

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;

Es funktioniert, aber es deaktiviert die erwartete Klugheit vollständig und ich muss einige manuelle Ausnahmebehandlungen durchführen. (Der Kürze halber entfernt).

Wie kann ich eine Parallel.ForEachSchleife implementieren , die das Schlüsselwort await im Lambda verwendet? Ist es möglich?

Der Prototyp der Parallel.ForEach-Methode verwendet einen Action<T>as-Parameter, möchte jedoch, dass er auf mein asynchrones Lambda wartet.

clausndk
quelle
1
Ich nehme an, Sie entfernen , sollten awaitaus await GetData(item)in Ihrem zweiten Codeblock als würde es einen Übersetzungsfehler erzeugen , wie sie ist.
Josh M.
2
Mögliches Duplikat von Nesting erwartet parallel.ForEach
Vitaliy Ulantikov

Antworten:

186

Wenn Sie nur einfache Parallelität wünschen, können Sie dies tun:

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;

Wenn Sie etwas Komplexeres benötigen, lesen Sie den ForEachAsyncBeitrag von Stephen Toub .

Stephen Cleary
quelle
46
Wahrscheinlich wird ein Drosselmechanismus benötigt. Dadurch werden sofort so viele Aufgaben erstellt, wie Elemente vorhanden sind, die möglicherweise in 10.000 Netzwerkanforderungen und dergleichen enden.
usr
10
@usr Das letzte Beispiel in Stephen Toubs Artikel befasst sich damit.
Svick
@svick Ich habe über das letzte Beispiel gerätselt. Es sieht für mich so aus, als würde es nur eine Menge Aufgaben stapeln, um mehr Aufgaben für mich zu erstellen, aber alle werden massenhaft gestartet.
Luke Puplett
2
@LukePuplett Es werden dopAufgaben erstellt, und jede von ihnen verarbeitet dann eine Teilmenge der Eingabesammlung in Reihe.
Svick
4
@Afshin_Zavvar: Wenn Sie aufrufen, Task.Runohne awaitdas Ergebnis zu erhalten, dann wirft das nur Feuer-und-Vergessen-Arbeit auf den Thread-Pool. Das ist fast immer ein Fehler.
Stephen Cleary
73

Sie können die ParallelForEachAsyncErweiterungsmethode aus dem AsyncEnumerator NuGet-Paket verwenden :

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
quelle
1
Das ist dein Paket? Ich habe gesehen, dass du das jetzt an einigen Stellen postest? : D Oh warte .. dein Name ist auf dem Paket: D +1
Piotr Kula
17
@ppumkin, ja, es ist meins. Ich habe dieses Problem immer und immer wieder gesehen, also habe ich beschlossen, es auf einfachste Weise zu lösen und andere vom Kämpfen zu befreien :)
Serge Semenov
Danke .. es macht definitiv Sinn und hat mir sehr geholfen!
Piotr Kula
2
Sie haben einen Tippfehler: maxDegreeOfParallelism>maxDegreeOfParalellism
Shiran Dror
3
Die korrekte Schreibweise ist in der Tat maxDegreeOfParallelism, aber es gibt etwas in @ ShiranDror's Kommentar - in Ihrem Paket haben Sie versehentlich die Variable maxDegreeOfParalellism aufgerufen (und daher wird Ihr zitierter Code erst kompiliert, wenn Sie ihn ändern ..)
BornToCode
17

Mit können SemaphoreSlimSie Parallelitätskontrolle erreichen.

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
quelle
3

Meine einfache Implementierung von ParallelForEach async.

Eigenschaften:

  1. Drosselung (maximaler Parallelitätsgrad).
  2. Ausnahmebehandlung (Aggregationsausnahme wird nach Abschluss ausgelöst).
  3. Speichereffizient (die Liste der Aufgaben muss nicht gespeichert werden).

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

Anwendungsbeispiel:

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

Ich habe hierfür eine Erweiterungsmethode erstellt, die SemaphoreSlim verwendet und es ermöglicht, einen maximalen Grad an Parallelität festzulegen

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

Beispielnutzung:

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);
Jay Shah
quelle
'using' hilft nicht. Die foreach-Schleife wartet auf unbestimmte Zeit auf das Semaphon. Versuchen Sie einfach diesen einfachen Code, der das Problem reproduziert: Warten Sie auf Enumerable.Range (1, 4) .ForEachAsyncConcurrent (async (i) => {Console.WriteLine (i); werfen Sie eine neue Ausnahme ("Testausnahme");}, maxDegreeOfParallelism: 2);
nicolay.anykienko
@ nicolay.anykienko Sie haben Recht mit # 2. Dieses Speicherproblem kann durch Hinzufügen von TaskWithThrottler.RemoveAll (x => x.IsCompleted) gelöst werden.
Askids
1
Ich habe es in meinem Code versucht und wenn ich maxDegreeOfParallelism nicht null ist, blockiert der Code. Hier können Sie den gesamten zu reproduzierenden Code sehen: stackoverflow.com/questions/58793118/…
Massimo Savazzi