Die Verschachtelung wartet parallel

182

In einer Metro-App muss ich eine Reihe von WCF-Aufrufen ausführen. Es muss eine erhebliche Anzahl von Anrufen getätigt werden, daher muss ich sie in einer parallelen Schleife ausführen. Das Problem ist, dass die parallele Schleife beendet wird, bevor alle WCF-Aufrufe abgeschlossen sind.

Wie würden Sie dies umgestalten, damit es wie erwartet funktioniert?

var ids = new List<string>() { "1", "2", "3", "4", "5", "6", "7", "8", "9", "10" };
var customers = new  System.Collections.Concurrent.BlockingCollection<Customer>();

Parallel.ForEach(ids, async i =>
{
    ICustomerRepo repo = new CustomerRepo();
    var cust = await repo.GetCustomer(i);
    customers.Add(cust);
});

foreach ( var customer in customers )
{
    Console.WriteLine(customer.ID);
}

Console.ReadKey();
Darthg8r
quelle

Antworten:

171

Die ganze Idee dahinter Parallel.ForEach()ist, dass Sie eine Reihe von Threads haben und jeder Thread einen Teil der Sammlung verarbeitet. Wie Sie bemerkt haben, funktioniert dies nicht mit async- await, wo Sie den Thread für die Dauer des asynchronen Aufrufs freigeben möchten.

Sie könnten das "beheben", indem Sie die ForEach()Threads blockieren , aber das macht den ganzen Punkt von async- zunichte await.

Sie können stattdessen TPL Dataflow verwenden Parallel.ForEach(), das asynchrone TaskDaten gut unterstützt.

Insbesondere könnte Ihr Code mit einem geschrieben werden TransformBlock, der jede ID Customermit dem asyncLambda in einen umwandelt . Dieser Block kann so konfiguriert werden, dass er parallel ausgeführt wird. Sie würden diesen Block mit einem verknüpfen, ActionBlockder jeden Customerin die Konsole schreibt . Nachdem Sie das Blocknetzwerk eingerichtet haben, können Sie Post()jede ID dem TransformBlock.

In Code:

var ids = new List<string> { "1", "2", "3", "4", "5", "6", "7", "8", "9", "10" };

var getCustomerBlock = new TransformBlock<string, Customer>(
    async i =>
    {
        ICustomerRepo repo = new CustomerRepo();
        return await repo.GetCustomer(i);
    }, new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
    });
var writeCustomerBlock = new ActionBlock<Customer>(c => Console.WriteLine(c.ID));
getCustomerBlock.LinkTo(
    writeCustomerBlock, new DataflowLinkOptions
    {
        PropagateCompletion = true
    });

foreach (var id in ids)
    getCustomerBlock.Post(id);

getCustomerBlock.Complete();
writeCustomerBlock.Completion.Wait();

Obwohl Sie wahrscheinlich die Parallelität der TransformBlockauf eine kleine Konstante beschränken möchten . Sie können auch die Kapazität von einschränken TransformBlockund die Elemente asynchron hinzufügen SendAsync(), z. B. wenn die Sammlung zu groß ist.

Ein zusätzlicher Vorteil im Vergleich zu Ihrem Code (falls er funktioniert hat) ist, dass das Schreiben beginnt, sobald ein einzelnes Element fertig ist, und nicht wartet, bis die gesamte Verarbeitung abgeschlossen ist.

svick
quelle
2
Ein sehr kurzer Überblick über asynchrone, reaktive Erweiterungen, TPL und TPL DataFlow - vantsuyoshi.wordpress.com/2012/01/05/… für diejenigen wie mich, die möglicherweise Klarheit benötigen.
Norman H
1
Ich bin mir ziemlich sicher, dass diese Antwort die Verarbeitung NICHT parallelisiert. Ich glaube, Sie müssen eine Parallel.ForEach über die IDs machen und diese an den getCustomerBlock senden. Zumindest habe ich das gefunden, als ich diesen Vorschlag getestet habe.
JasonLind
4
@ JasonLind Das tut es wirklich. Die parallele Verwendung Parallel.ForEach()von Post()Elementen sollte keine wirklichen Auswirkungen haben.
Svick
1
@svick Ok, ich habe es gefunden, der ActionBlock muss auch parallel sein. Ich habe es etwas anders gemacht, ich brauchte keine Transformation, also habe ich nur einen Pufferblock verwendet und meine Arbeit im ActionBlock gemacht. Ich war verwirrt von einer anderen Antwort in den Interwebs.
JasonLind
2
Damit meine ich die Angabe von MaxDegreeOfParallelism auf dem ActionBlock wie auf dem TransformBlock in Ihrem Beispiel
JasonLind
125

Die Antwort von svick ist (wie immer) ausgezeichnet.

Ich finde Dataflow jedoch nützlicher, wenn Sie tatsächlich große Datenmengen übertragen müssen. Oder wenn Sie eine asynckompatible Warteschlange benötigen .

In Ihrem Fall besteht eine einfachere Lösung darin, nur die asyncParallelität im Stil zu verwenden:

var ids = new List<string>() { "1", "2", "3", "4", "5", "6", "7", "8", "9", "10" };

var customerTasks = ids.Select(i =>
  {
    ICustomerRepo repo = new CustomerRepo();
    return repo.GetCustomer(i);
  });
var customers = await Task.WhenAll(customerTasks);

foreach (var customer in customers)
{
  Console.WriteLine(customer.ID);
}

Console.ReadKey();
Stephen Cleary
quelle
13
Wenn Sie die Parallelität manuell begrenzen möchten (was Sie in diesem Fall höchstwahrscheinlich tun), wäre dies komplizierter.
Svick
1
Aber Sie haben Recht, dass der Datenfluss ziemlich kompliziert sein kann (zum Beispiel im Vergleich zu Parallel.ForEach()). Aber ich denke, es ist derzeit die beste Option, um fast jede asyncArbeit mit Sammlungen zu erledigen .
Svick
1
@ JamesManning Wie wird ParallelOptionses helfen? Es gilt nur für Parallel.For/ForEach/Invoke, die, wie das OP festgelegt hat, hier keinen Nutzen haben.
Ohad Schneider
1
@StephenCleary Wenn die GetCustomerMethode a zurückgibt Task<T>, sollte man verwenden Select(async i => { await repo.GetCustomer(i);});?
Shyju
5
@ Batmaci: Parallel.ForEachunterstützt nicht async.
Stephen Cleary
79

Die Verwendung von DataFlow, wie von svick vorgeschlagen, kann übertrieben sein, und Stephens Antwort bietet nicht die Möglichkeit, die Parallelität des Vorgangs zu steuern. Dies kann jedoch ziemlich einfach erreicht werden:

public static async Task RunWithMaxDegreeOfConcurrency<T>(
     int maxDegreeOfConcurrency, IEnumerable<T> collection, Func<T, Task> taskFactory)
{
    var activeTasks = new List<Task>(maxDegreeOfConcurrency);
    foreach (var task in collection.Select(taskFactory))
    {
        activeTasks.Add(task);
        if (activeTasks.Count == maxDegreeOfConcurrency)
        {
            await Task.WhenAny(activeTasks.ToArray());
            //observe exceptions here
            activeTasks.RemoveAll(t => t.IsCompleted); 
        }
    }
    await Task.WhenAll(activeTasks.ToArray()).ContinueWith(t => 
    {
        //observe exceptions in a manner consistent with the above   
    });
}

Die ToArray()Aufrufe können optimiert werden, indem ein Array anstelle einer Liste verwendet und abgeschlossene Aufgaben ersetzt werden. Ich bezweifle jedoch, dass dies in den meisten Szenarien einen großen Unterschied bewirken würde. Beispielnutzung gemäß der Frage des OP:

RunWithMaxDegreeOfConcurrency(10, ids, async i =>
{
    ICustomerRepo repo = new CustomerRepo();
    var cust = await repo.GetCustomer(i);
    customers.Add(cust);
});

EDIT Fellow SO-Benutzer und TPL-Assistent Eli Arbel hat mich auf einen verwandten Artikel von Stephen Toub hingewiesen . Wie immer ist seine Implementierung sowohl elegant als auch effizient:

public static Task ForEachAsync<T>(
      this IEnumerable<T> source, int dop, Func<T, Task> body) 
{ 
    return Task.WhenAll( 
        from partition in Partitioner.Create(source).GetPartitions(dop) 
        select Task.Run(async delegate { 
            using (partition) 
                while (partition.MoveNext()) 
                    await body(partition.Current).ContinueWith(t => 
                          {
                              //observe exceptions
                          });

        })); 
}
Ohad Schneider
quelle
1
@RichardPierre Diese Überlastung Partitioner.Createverwendet tatsächlich die Chunk-Partitionierung, die Elemente dynamisch für die verschiedenen Aufgaben bereitstellt, sodass das von Ihnen beschriebene Szenario nicht stattfindet. Beachten Sie auch, dass die statische (vorgegebene) Partitionierung in einigen Fällen aufgrund des geringeren Overheads (insbesondere der Synchronisierung) schneller sein kann. Weitere Informationen finden Sie unter: msdn.microsoft.com/en-us/library/dd997411(v=vs.110).aspx .
Ohad Schneider
1
@OhadSchneider In den // Ausnahmen beobachten, wenn dies eine Ausnahme auslöst, sprudelt sie dann zum Anrufer? Wenn ich zum Beispiel wollte, dass die gesamte Aufzählung die Verarbeitung beendet / fehlschlägt, wenn ein Teil davon fehlschlägt?
Terry
3
@Terry sprudelt zum Anrufer in dem Sinne, dass die oberste Aufgabe (erstellt von Task.WhenAll) die Ausnahme (innerhalb eines AggregateException) enthält. Wenn der Anrufer verwendet wird await, wird folglich eine Ausnahme in der Anrufsite ausgelöst. Wartet Task.WhenAlljedoch weiterhin auf den Abschluss aller Aufgaben und weist GetPartitionsElemente beim partition.MoveNextAufruf dynamisch zu, bis keine weiteren Elemente mehr verarbeitet werden müssen. Dies bedeutet, dass dies nicht von alleine CancellationTokengeschieht , wenn Sie keinen eigenen Mechanismus zum Stoppen der Verarbeitung hinzufügen (z. B. ).
Ohad Schneider
1
@gibbocool Ich bin mir immer noch nicht sicher, ob ich folge. Angenommen, Sie haben insgesamt 7 Aufgaben mit den Parametern, die Sie in Ihrem Kommentar angegeben haben. Angenommen, die erste Charge übernimmt gelegentlich 5-Sekunden-Aufgaben und drei 1-Sekunden-Aufgaben. Nach ungefähr einer Sekunde wird die 5-Sekunden-Aufgabe weiterhin ausgeführt, während die drei 1-Sekunden-Aufgaben abgeschlossen sind. Zu diesem Zeitpunkt werden die verbleibenden drei 1-Sekunden-Tasks ausgeführt (sie werden vom Partitionierer an die drei "freien" Threads übergeben).
Ohad Schneider
1
@MichaelFreidgeim kannst du so etwas wie var current = partition.Currentvorher machen await bodyund dann currentin der Fortsetzung ( ContinueWith(t => { ... }) verwenden.
Ohad Schneider
41

Mit dem neuen AsyncEnumerator NuGet-Paket , das vor 4 Jahren nicht existierte, als die Frage ursprünglich gestellt wurde, können Sie Aufwand sparen . Hiermit können Sie den Grad der Parallelität steuern:

using System.Collections.Async;
...

await ids.ParallelForEachAsync(async i =>
{
    ICustomerRepo repo = new CustomerRepo();
    var cust = await repo.GetCustomer(i);
    customers.Add(cust);
},
maxDegreeOfParallelism: 10);

Haftungsausschluss: Ich bin der Autor der AsyncEnumerator-Bibliothek, die Open Source ist und unter MIT lizenziert ist, und ich poste diese Nachricht nur, um der Community zu helfen.

Serge Semenov
quelle
11
Sergey, Sie sollten offenlegen, dass Sie ein Autor der Bibliothek sind
Michael Freidgeim
5
ok, fügte den Haftungsausschluss hinzu. Ich suche keinen Nutzen aus der Werbung, möchte nur Menschen helfen;)
Serge Semenov
Ihre Bibliothek ist nicht mit .NET Core kompatibel.
Corniel Nobel
2
@CornielNobel, es ist kompatibel mit .NET Core - der Quellcode auf GitHub bietet eine Testabdeckung für .NET Framework und .NET Core.
Serge Semenov
1
@SergeSemenov Ich habe Ihre Bibliothek viel dafür benutzt AsyncStreamsund ich muss sagen, dass sie ausgezeichnet ist. Kann diese Bibliothek nicht genug empfehlen.
WBuck
16

Wickeln Sie das Parallel.Foreachin ein Task.Run()und anstelle des awaitSchlüsselworts verwenden[yourasyncmethod].Result

(Sie müssen die Task.Run-Aktion ausführen, um den UI-Thread nicht zu blockieren.)

Etwas wie das:

var yourForeachTask = Task.Run(() =>
        {
            Parallel.ForEach(ids, i =>
            {
                ICustomerRepo repo = new CustomerRepo();
                var cust = repo.GetCustomer(i).Result;
                customers.Add(cust);
            });
        });
await yourForeachTask;
ofcoursedude
quelle
3
Was ist das Problem damit? Ich hätte es genau so gemacht. Lassen Parallel.ForEachSie die parallele Arbeit, die blockiert, bis alle erledigt sind, und verschieben Sie das Ganze dann in einen Hintergrund-Thread, um eine reaktionsfähige Benutzeroberfläche zu erhalten. Irgendwelche Probleme damit? Vielleicht ist das ein schlafender Thread zu viel, aber es ist kurzer, lesbarer Code.
Ygoe
@LonelyPixel Mein einziges Problem ist, dass es anruft, Task.Runwann TaskCompletionSourcees vorzuziehen ist.
Gusdor
1
@ Gusdor Neugierig - warum ist TaskCompletionSourcevorzuziehen?
Seafish
@ Seafish Eine gute Frage, die ich gerne beantworten könnte. Muss ein harter Tag gewesen sein: D
Gusdor
Nur ein kurzes Update. Ich habe jetzt genau danach gesucht, nach unten gescrollt, um die einfachste Lösung zu finden, und wieder meinen eigenen Kommentar gefunden. Ich habe genau diesen Code verwendet und er funktioniert wie erwartet. Es wird nur davon ausgegangen, dass sich innerhalb der Schleife eine Synchronisierungsversion der ursprünglichen Async-Aufrufe befindet. awaitkann nach vorne verschoben werden, um den zusätzlichen Variablennamen zu speichern.
Ygoe
7

Dies sollte ziemlich effizient und einfacher sein, als den gesamten TPL-Datenfluss zum Laufen zu bringen:

var customers = await ids.SelectAsync(async i =>
{
    ICustomerRepo repo = new CustomerRepo();
    return await repo.GetCustomer(i);
});

...

public static async Task<IList<TResult>> SelectAsync<TSource, TResult>(this IEnumerable<TSource> source, Func<TSource, Task<TResult>> selector, int maxDegreesOfParallelism = 4)
{
    var results = new List<TResult>();

    var activeTasks = new HashSet<Task<TResult>>();
    foreach (var item in source)
    {
        activeTasks.Add(selector(item));
        if (activeTasks.Count >= maxDegreesOfParallelism)
        {
            var completed = await Task.WhenAny(activeTasks);
            activeTasks.Remove(completed);
            results.Add(completed.Result);
        }
    }

    results.AddRange(await Task.WhenAll(activeTasks));
    return results;
}
John Gietzen
quelle
Sollte das Verwendungsbeispiel nicht awaitwie folgt verwendet werden : var customers = await ids.SelectAsync(async i => { ... });?
Paccc
5

Ich bin etwas spät zum Feiern, aber vielleicht möchten Sie GetAwaiter.GetResult () verwenden, um Ihren asynchronen Code im Synchronisationskontext auszuführen, aber so parallel wie unten;

 Parallel.ForEach(ids, i =>
{
    ICustomerRepo repo = new CustomerRepo();
    // Run this in thread which Parallel library occupied.
    var cust = repo.GetCustomer(i).GetAwaiter().GetResult();
    customers.Add(cust);
});
Teoman Shipahi
quelle
5

Nachdem Sie eine Reihe von Hilfsmethoden eingeführt haben, können Sie parallele Abfragen mit dieser einfachen Syntax ausführen:

const int DegreeOfParallelism = 10;
IEnumerable<double> result = await Enumerable.Range(0, 1000000)
    .Split(DegreeOfParallelism)
    .SelectManyAsync(async i => await CalculateAsync(i).ConfigureAwait(false))
    .ConfigureAwait(false);

Was hier passiert, ist: Wir teilen die Quellensammlung in 10 Chunks ( .Split(DegreeOfParallelism)) auf, führen dann 10 Tasks aus, die jeweils ihre Elemente .SelectManyAsync(...)einzeln verarbeiten ( ), und führen diese wieder in einer einzigen Liste zusammen.

Erwähnenswert ist, dass es einen einfacheren Ansatz gibt:

double[] result2 = await Enumerable.Range(0, 1000000)
    .Select(async i => await CalculateAsync(i).ConfigureAwait(false))
    .WhenAll()
    .ConfigureAwait(false);

Aber es bedarf einer Vorsichtsmaßnahme : Wenn Sie über eine zu große Quellensammlung verfügen, wird Taskfür jeden Artikel sofort ein Zeitplan festgelegt , was zu erheblichen Leistungseinbußen führen kann.

Die in den obigen Beispielen verwendeten Erweiterungsmethoden sehen wie folgt aus:

public static class CollectionExtensions
{
    /// <summary>
    /// Splits collection into number of collections of nearly equal size.
    /// </summary>
    public static IEnumerable<List<T>> Split<T>(this IEnumerable<T> src, int slicesCount)
    {
        if (slicesCount <= 0) throw new ArgumentOutOfRangeException(nameof(slicesCount));

        List<T> source = src.ToList();
        var sourceIndex = 0;
        for (var targetIndex = 0; targetIndex < slicesCount; targetIndex++)
        {
            var list = new List<T>();
            int itemsLeft = source.Count - targetIndex;
            while (slicesCount * list.Count < itemsLeft)
            {
                list.Add(source[sourceIndex++]);
            }

            yield return list;
        }
    }

    /// <summary>
    /// Takes collection of collections, projects those in parallel and merges results.
    /// </summary>
    public static async Task<IEnumerable<TResult>> SelectManyAsync<T, TResult>(
        this IEnumerable<IEnumerable<T>> source,
        Func<T, Task<TResult>> func)
    {
        List<TResult>[] slices = await source
            .Select(async slice => await slice.SelectListAsync(func).ConfigureAwait(false))
            .WhenAll()
            .ConfigureAwait(false);
        return slices.SelectMany(s => s);
    }

    /// <summary>Runs selector and awaits results.</summary>
    public static async Task<List<TResult>> SelectListAsync<TSource, TResult>(this IEnumerable<TSource> source, Func<TSource, Task<TResult>> selector)
    {
        List<TResult> result = new List<TResult>();
        foreach (TSource source1 in source)
        {
            TResult result1 = await selector(source1).ConfigureAwait(false);
            result.Add(result1);
        }
        return result;
    }

    /// <summary>Wraps tasks with Task.WhenAll.</summary>
    public static Task<TResult[]> WhenAll<TResult>(this IEnumerable<Task<TResult>> source)
    {
        return Task.WhenAll<TResult>(source);
    }
}
Vitaliy Ulantikov
quelle
4

Eine Erweiterungsmethode hierfür, die SemaphoreSlim verwendet und es auch 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