Идиома Process Tasks By Completion

При работе с тасками часто возникает такая задача: у нас есть набор входных данных и для обработки каждого элемента используется длительная операция.

Можно подойти к этой задаче в лоб. Крутим цикл, запускаем таски, обрабатываем результаты по одному:private Task<,Weather>, GetWeatherForAsync(string city)
{
Console.WriteLine('[{1}]: Getting the weather for '{0}'', city,
DateTime.Now.ToLongTimeString()),
return WeatherService.GetWeatherAsync(city),
}
[
Test]
public async Task
ProcessOneByOneNaive()
{
var cities = new List<,string>, { 'Moscow', 'Seattle', 'New York' },
var tasks =
from city in
cities
select new { City = city, WeatherTask = GetWeatherForAsync(city) },
foreach (var entry in tasks)
{
var wheather = await entry.WeatherTask,
ProcessWeather(entry
.City, wheather),
}
}
private void ProcessWeather(string city, Weather
weather)
{
Console.WriteLine('[{2}]: Processing weather for '{0}': '{1}'', city, weather,
DateTime.Now.ToLongTimeString()),
}

Здесь мы обращаемся к некоторому сервису погоды, для получения температуры в каждом городе, затем обрабатываем полученные результаты путем вывода города и температуры на экран.

Подход рабочий, но есть одна проблема: новая задача будет запущена лишь после завершения предыдущей. Тут можно принудительно дернуть все задачи и вызвать ToList() на LINQ-запросе, но и в этом случае задачи будут обрабатываться в порядке городов, а не в порядке доступности результатов (предположим, что для получения погоды для первого города уйдет втрое больше времени, чем для других, в этом случае мы будем ждать результат по первому городу, хотя результаты по двум другим уже доступны).

Решение заключается в использовании идиомы Process Tasks by Completion (можете называть это паттерном, если хотите), которая заключается в следующем: задачи должны обрабатываться не в порядке их запуска, а в порядке их завершения.

Вот как это будет выглядеть:[Test]
public async Task
ManualProcessByCompletion()
{
var cities = new List<,string>, { 'Moscow', 'Seattle', 'New York' },
var tasks = (from city in cities
let result = new { City = city, WeatherTask = GetWeatherForAsync(city) }
select TaskEx.FromTask(result, r =>, r.WeatherTask)).ToList(),
while (tasks.Count != 0)
{
var completedTask = await Task.WhenAny(tasks),
tasks
.Remove(completedTask),
var result = completedTask.Result,
ProcessWeather(result
.City, result.WeatherTask.Result),
}
}
ПРИМЕЧАНИЕ
В одном из курсов Pluralsight данная идиома называется Process tasks one by one, но, ИМХО, это еще менее понятное название, чем у меня. Так что если есть мысли, какое название будет лучше передавать ее суть буду рад выслушать варианты!

Метод ManualProcessByCompletion, запускает получение погоды с помощью GetWeatherForAsync для всех городов, оборачивает объект анонимного типа в таску (позже объясню, почему это нужно). Затем, внутри цикла while мы зовем Task.WhenAny и получаем первую завершенную задачу. Получается, что задачи обрабатываются по мере завершения, а не по порядку их запуска. Но в данном случае, таска содержит ответ на вопрос (какая погода в городе?), но не содержит самого вопроса (имени города). Нам нужно как-то объединить результат и контекст исполнения. Для этого используется метод TaskEx.FromTask:public static Task<,T>, FromTask<,T, U>,(T result, Func<,T, Task<,U>,>, taskSelector)
{
Contract.Requires(taskSelector != null),
var tcs = new TaskCompletionSource<,T>,(),
var task = taskSelector(result),
task
.ContinueWith(t =>,
{
if (t.IsFaulted)
tcs
.SetException(t.Exception),
else if (t.IsCanceled)
tcs
.SetCanceled(),
else
tcs.
SetResult(result),
}),
return tcs.Task,
}

Метод TaskEx.FromResult, создает прокси-таску, которая завершится при завершении оригинальной задачи. А делегат taskSelector позволяет извлечь задачу из основного объекта, что позволяет удобно использовать этот подход совместно с анонимными типами.

Новый подход работает лучше оригинального, но выглядит более громоздким. Есть смысл сделать небольшую обертку, которая позволит использовать его повторно.public static IEnumerable<,Task<,TElement>,>, OrderByCompletion<,TElement, TTaskResult>,(
this IEnumerable<,TElement>, sequence, Func<,TElement, Task<,TTaskResult>,>, taskSelector)
{
Contract.Requires(sequence != null),
Contract.Requires(taskSelector != null),
var tasks = (from element in sequence
let pair = new {Element = element, Task = taskSelector(element)}
select FromTask(pair, p =>, p.Task)).ToList(),
while (tasks.Count != 0)
{
var tcs = new TaskCompletionSource<,TElement>,(),
// Getting the first finished task
Task.WhenAny(tasks).ContinueWith(tsk =>,
{
var finishedTask = tsk.Result,
tasks
.Remove(finishedTask),
tcs
.FromTask(finishedTask, arg =>, arg.Element),
}),
yield return tcs.Task,
}
}

Полный код класса доступен на гитхабе, но смысл его такой. Метод преобразует последовательность задач в другую последовательность задач, порядок которых определяется порядком завершения задач, а не порядком в исходной последовательности.

И вот как выглядит пример использования:[Test]
public async Task
ProcessByCompletion()
{
var cities = new List<,string>, { 'Moscow', 'Seattle', 'New York' },
var tasks =
from city in
cities
select new {City = city, WeatherTask = GetWeatherForAsync(city)},
foreach (var task in tasks.OrderByCompletion(t =>, t.WeatherTask))
{
var taskResult = await task,
// taskResult is an object of anonymous type with City and WeatherTask
ProcessWeather(taskResult.City, taskResult.WeatherTask.Result),
}
}

Мы снова вернулись к исходному варианту с точки зрения синтаксиса, но оставили новое поведение. Вот результат исполнения, который показывает, что все задачи запускаются одновременно, а обработка происходит по мере поступления результатов:

[12:54:35 PM]: Getting the weather for 'Moscow'
[12:54:35 PM]: Getting the weather for 'Seattle'
[12:54:35 PM]: Getting the weather for 'New York'
[12:54:36 PM]: Processing weather for 'Seattle': 'Temp: 7C'
Got the weather for 'Moscow'
[12:54:39 PM]: Processing weather for 'Moscow': 'Temp: 6C'
Got the weather for 'New York'
[12:54:40 PM]: Processing weather for 'New York': 'Temp: 8C'

UPDATE:

Отказ от Query Comprehension Syntax-а позволит упростить последний пример еще немного:

[Test]
public async Task
ProcessByCompletion()
{
var cities = new List<,string>, { 'Moscow', 'Seattle', 'New York' },
var tasks = cities.Select(async city =>,
{
return new {City = city, Weather = await GetWeatherForAsync(city)},
}),
foreach (var task in tasks.OrderByCompletion())
{
var taskResult = await task,
// taskResult is an object of anonymous type with City and WeatherTask
ProcessWeather(taskResult.City, taskResult.Weather),
}
}

И вариант, предложенный @hazzik на основе Rx-ов:

[Test]
public void
ProcessOneUsingRx()
{
var cities = new[] { 'Moscow', 'Seattle', 'New York' },
var objs = cities.Select(async city =>, new
{
City
= city,
Weather
= await GetWeatherForAsync(city)
})
.Select(task =>, task.ToObservable()).Merge().ToEnumerable(),
foreach (var obj in objs)
{
ProcessWeather(obj
.City, obj.Weather),
}
}

Он работает следующим образом: вначале мы берем последовательность задач, конвертим ее в последовательность IEnumerable<,IObservable<,T>,>,, которая, затем мерджится в одну последовательность IObservable<,T>,.

Особенность этого подхода в том, что в данном случае вызов MoveNext в цикле foreach является блокирующим.

З.Ы. Код залит в новый репо на гитхабе - https://github.com/SergeyTeplyakov/TplTipsAndTricks