Jak uzyskiwać z zadań równoległych w .NET 4.5

Question

Chciałbym użyć iteratora .NET z równoległymi zadaniami/poczekać ?. Coś takiego:

IEnumerable<TDst> Foo<TSrc, TDest>(IEnumerable<TSrc> source) 
{ 
    Parallel.ForEach(
     source, 
     s=> 
     { 
      // Ordering is NOT important 
      // items can be yielded as soon as they are done     
      yield return ExecuteOrDownloadSomething(s); 
     } 
} 

Niestety .NET nie może natywnie sobie z tym poradzić. Najlepsza odpowiedź do tej pory przez @svick - użyj AsParallel().

BONUS: Każdy prosty kod asynchroniczny/oczekujący, który implementuje wielu wydawców i jednego subskrybenta? Subskrybent wyda, a puby będą przetwarzać. (tylko biblioteki rdzeniowe)

Answer 1

To wydaje się zadanie dla PLINQ:

return source.AsParallel().Select(s => ExecuteOrDownloadSomething(s)); 

Spowoduje to wykonanie delegata równolegle przy użyciu ograniczonej liczby wątków, zwracając każdy wynik zaraz po jego zakończeniu.

Jeśli metoda ExecuteOrDownloadSomething() IO jest związany (np faktycznie pobiera coś) i nie chcesz tracić wątki, a następnie za pomocą async - await może mieć sens, ale byłoby bardziej skomplikowane.

Jeśli chcesz w pełni skorzystać z async, nie powinieneś zwracać numeru IEnumerable, ponieważ jest on synchroniczny (tzn. Blokuje, jeśli nie są dostępne żadne elementy). Co jest potrzebne jest jakieś kolekcji asynchronicznym, a można użyć ISourceBlock (konkretnie TransformBlock) z OC dataflow za to:

ISourceBlock<TDst> Foo<TSrc, TDest>(IEnumerable<TSrc> source) 
{ 
    var block = new TransformBlock<TSrc, TDest>(
     async s => await ExecuteOrDownloadSomethingAsync(s), 
     new ExecutionDataflowBlockOptions 
     { 
      MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded 
     }); 

    foreach (var item in source) 
     block.Post(item); 

    block.Complete(); 

    return block; 
} 

Jeśli źródłem jest „wolna” (tj chcesz rozpocząć przetwarzanie wyników Foo() przed zakończeniem powtarzania source), możesz chcieć przenieść wywołanie foreach i Complete() do osobnego Task. Jeszcze lepszym rozwiązaniem byłoby wprowadzenie source do ISourceBlock<TSrc>.

Answer 2

W bibliotece asynchronicznej wykonanej przez zespół robotyki MS, posiadały prymitywy współbieżności, które pozwoliły na użycie iteratora do wygenerowania kodu asynchronicznego.

Biblioteka (CCR) jest bezpłatna (nie była wolna). Miły artykuł wprowadzający można znaleźć tutaj: Concurrent affairs

Być może można użyć tej biblioteki obok biblioteki .NET zadania, lub będzie ona inspiracją do „roll własne”

Answer 3

Wygląda na to, że naprawdę chcesz, aby zamówić sekwencję zadań w oparciu o czas ukończenia. To nie jest strasznie skomplikowane:

public static IEnumerable<Task<T>> Order<T>(this IEnumerable<Task<T>> tasks) 
{ 
    var input = tasks.ToList(); 

    var output = input.Select(task => new TaskCompletionSource<T>()); 
    var collection = new BlockingCollection<TaskCompletionSource<T>>(); 
    foreach (var tcs in output) 
     collection.Add(tcs); 

    foreach (var task in input) 
    { 
     task.ContinueWith(t => 
     { 
      var tcs = collection.Take(); 
      switch (task.Status) 
      { 
       case TaskStatus.Canceled: 
        tcs.TrySetCanceled(); 
        break; 
       case TaskStatus.Faulted: 
        tcs.TrySetException(task.Exception.InnerExceptions); 
        break; 
       case TaskStatus.RanToCompletion: 
        tcs.TrySetResult(task.Result); 
        break; 
      } 
     } 
     , CancellationToken.None 
     , TaskContinuationOptions.ExecuteSynchronously 
     , TaskScheduler.Default); 
    } 

    return output.Select(tcs => tcs.Task); 
} 

Więc tworzymy TaskCompletionSource dla każdego zadania wejściowego, a następnie przejść przez każde z zadań i ustaw kontynuację który chwyta następną źródło uzupełniania z BlockingCollection i ustawia to wynik. Pierwsze ukończone zadanie pobiera pierwsze zwrócone tcs, drugie wykonane zadanie otrzymuje drugie zwrócone tcs, i tak dalej.

Teraz kod staje się dość prosta:

var tasks = collection.Select(item => LongRunningOperationThatReturnsTask(item)) 
    .Order(); 
foreach(var task in tasks) 
{ 
    var result = task.Result;//or you could `await` each result 
    //.... 
}