Zagnieżdżanie czeka w Parallel.ForEach

W aplikacji metra muszę wykonać wiele połączeń WCF. Istnieje wiele połączeń, które należy wykonać, więc muszę je wykonywać w pętli równoległej. Problem polega na tym, że pętla równoległa kończy się przed zakończeniem wszystkich wywołań WCF.

Jak zmieniłbyś to tak, aby działało zgodnie z oczekiwaniami?

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

Cała idea Parallel.ForEach()polega na tym, że masz zestaw wątków, a każdy wątek przetwarza część kolekcji. Jak zauważyłeś, nie działa to z async- await, gdzie chcesz zwolnić wątek na czas trwania wywołania asynchronicznego.

Możesz to „naprawić”, blokując ForEach()wątki, ale to niweczy cały punkt async- await.

Zamiast tego możesz użyć przepływu danych TPLParallel.ForEach() , który Taskdobrze obsługuje asynchroniczne s.

Konkretnie, twój kod może być napisany przy użyciu, TransformBlockktóry przekształca każdy identyfikator w Customerprzy użyciu asynclambda. Blok ten można skonfigurować do wykonywania równoległego. Łączyłbyś ten blok z pismem, ActionBlockktóry zapisuje każdy Customerdo konsoli. Po skonfigurowaniu sieci bloków możesz Post()przypisać każdy identyfikator do TransformBlock.

W kodzie:

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

Chociaż prawdopodobnie chcesz ograniczyć paralelizm tej TransformBlockmałej stałej. Ponadto możesz ograniczyć pojemność TransformBlocki dodać do niej elementy asynchronicznie SendAsync(), na przykład, jeśli kolekcja jest zbyt duża.

Dodatkową korzyścią w porównaniu z twoim kodem (jeśli zadziałał) jest to, że pisanie rozpocznie się, gdy tylko pojedynczy element zostanie zakończony, i nie czekaj, aż całe przetwarzanie zostanie zakończone.

Odpowiedź Svicka jest (jak zwykle) doskonała.

Uważam jednak, że przepływ danych jest bardziej przydatny, gdy faktycznie masz duże ilości danych do przesłania. Lub gdy potrzebujesz asynckolejki kompatybilnej.

W twoim przypadku prostszym rozwiązaniem jest po prostu użycie asyncrównoległości stylu:

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

Używanie DataFlow zgodnie z sugestią Svicka może być przesadą, a odpowiedź Stephena nie zapewnia środków do kontrolowania współbieżności operacji. Można to jednak osiągnąć po prostu:

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

Te ToArray()połączenia mogą być optymalizowane przy użyciu tablicę zamiast listy i zastąpienie wykonanych zadań, ale wątpię, by to zrobić wielkiej różnicy w większości scenariuszy. Przykładowe użycie według pytania PO:

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

EDIT Inny użytkownik SO i kreator TPL, Eli Arbel, wskazał mi powiązany artykuł Stephena Touba . Jak zwykle jego implementacja jest zarówno elegancka, jak i wydajna:

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

        })); 
}

Możesz zaoszczędzić wysiłek dzięki nowemu pakietowi NuGet AsyncEnumerator , który nie istniał 4 lata temu, kiedy pytanie zostało pierwotnie opublikowane. Pozwala kontrolować stopień równoległości:

using System.Collections.Async;
...

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

Oświadczenie: Jestem autorem biblioteki AsyncEnumerator, która jest oprogramowaniem typu open source i jest licencjonowana na MIT, i wysyłam tę wiadomość, aby pomóc społeczności.

Zawiń Parallel.Foreachw a Task.Run()i zamiast awaitsłowa kluczowego użyj[yourasyncmethod].Result

(musisz wykonać zadanie Task.Run, aby nie blokować wątku interfejsu użytkownika)

Coś takiego:

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

Powinno to być dość wydajne i łatwiejsze niż uruchomienie całego przepływu danych TPL:

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

Jestem trochę spóźniony na imprezę, ale możesz rozważyć użycie GetAwaiter.GetResult (), aby uruchomić kod asynchroniczny w kontekście synchronizacji, ale tak równolegle, jak poniżej;

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

Metoda rozszerzenia dla tego, która wykorzystuje SemaphoreSlim, a także pozwala ustawić maksymalny stopień równoległości

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

Przykładowe użycie:

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);

Po wprowadzeniu szeregu metod pomocniczych będziesz mógł wykonywać równoległe zapytania za pomocą tej prostej składni:

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

Dzieje się tak: dzielimy kolekcję źródeł na 10 części ( .Split(DegreeOfParallelism)), a następnie uruchamiamy 10 zadań, przetwarzając poszczególne elementy jeden po drugim ( .SelectManyAsync(...)) i scalamy je z powrotem w jedną listę.

Warto wspomnieć, że istnieje prostsze podejście:

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

Ale wymaga to ostrożności : jeśli masz zbyt dużą kolekcję źródeł, Taskod razu zaplanuje dla każdego elementu, co może spowodować znaczny spadek wydajności.

Metody rozszerzenia stosowane w powyższych przykładach wyglądają następująco:

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