11

I am trying to update my toolset with the new tools offered by C# 8, and one method that seems particularly useful is a version of Task.WhenAll that returns an IAsyncEnumerable. This method should stream the task results as soon as they become available, so naming it WhenAll doesn't make much sense. WhenEach sounds more appropriate. The signature of the method is:

public static IAsyncEnumerable<TResult> WhenEach<TResult>(Task<TResult>[] tasks);

This method could be used like this:

var tasks = new Task<int>[]
{
    ProcessAsync(1, 300),
    ProcessAsync(2, 500),
    ProcessAsync(3, 400),
    ProcessAsync(4, 200),
    ProcessAsync(5, 100),
};

await foreach (int result in WhenEach(tasks))
{
    Console.WriteLine($"Processed: {result}");
}

static async Task<int> ProcessAsync(int result, int delay)
{
    await Task.Delay(delay);
    return result;
}

Expected output:

Processed: 5
Processed: 4
Processed: 1
Processed: 3
Processed: 2

I managed to write a basic implementation using the method Task.WhenAny in a loop, but there is a problem with this approach:

public static async IAsyncEnumerable<TResult> WhenEach<TResult>(
    Task<TResult>[] tasks)
{
    var hashSet = new HashSet<Task<TResult>>(tasks);
    while (hashSet.Count > 0)
    {
        var task = await Task.WhenAny(hashSet).ConfigureAwait(false);
        yield return await task.ConfigureAwait(false);
        hashSet.Remove(task);
    }
}

The problem is the performance. The Task.WhenAny method has to watch for the completion of all the supplied tasks, and it does so by attaching and detaching continuations, so calling it repeatedly in a loop results in O(n²) computational complexity. My naive implementation struggles to process 10,000 tasks. The overhead is nearly 10 sec in my machine. I would like the method to be nearly as performant as the build-in Task.WhenAll, that can handle hundreds of thousands of tasks with ease. How could I improve the WhenEach method to make it perform decently?

Theodor Zoulias
  • 34,835
  • 7
  • 69
  • 104
  • 1
    Maybe this can be of some use to you: https://devblogs.microsoft.com/pfxteam/processing-tasks-as-they-complete/ About halfway down the article you will see a performance version. – JohanP Oct 02 '19 at 01:46
  • @JohanP interesting article, thanks! The technique of divide-and conquer (apply the `Task.WhenAny` in subsequences) passed through my mind as possible solution, but it is complex and may still not be optimal. The other technique with `ContinueWith` seems more promising, but I have a hard time visualizing how it can be combined with an `IAsyncEnumerable` as return value. – Theodor Zoulias Oct 02 '19 at 01:58
  • You wont be able to yield inside an anonymous method unfortunately, so ContinueWith is out as far as i cant tell. – TheGeneral Oct 02 '19 at 02:02
  • @TheodorZoulias You can do the `foreach(var bucket in Interleaved(tasks))` inside your `WhenEach` and then `yield return await (await bucket)` or something along those lines – JohanP Oct 02 '19 at 02:07
  • @TheGeneral yeap, I can't think of a way to overpass this limitation with the `ContinueWith` approach. – Theodor Zoulias Oct 02 '19 at 02:08
  • You could possibly use the `ContinueWith` to signal a Async like AutoResetEvent to *repoll* the list and *yield* in that, might give you a slight performance boost over `WhenAny`, though i am not sure – TheGeneral Oct 02 '19 at 02:16
  • @TheGeneral [`AsyncAutoResetEvent`](https://learn.microsoft.com/en-us/dotnet/api/microsoft.visualstudio.threading.asyncautoresetevent)! [`Microsoft.VisualStudio.Threading`](https://www.nuget.org/packages/Microsoft.VisualStudio.Threading/)! New stuff for me. :-) – Theodor Zoulias Oct 02 '19 at 02:29
  • @TheodorZoulias https://devblogs.microsoft.com/pfxteam/building-async-coordination-primitives-part-2-asyncautoresetevent/ also https://github.com/StephenCleary/AsyncEx/wiki/AsyncAutoResetEvent – TheGeneral Oct 02 '19 at 02:29

5 Answers5

8

By using code from this article, you can implement the following:

public static Task<Task<T>>[] Interleaved<T>(IEnumerable<Task<T>> tasks)
{
   var inputTasks = tasks.ToList();

   var buckets = new TaskCompletionSource<Task<T>>[inputTasks.Count];
   var results = new Task<Task<T>>[buckets.Length];
   for (int i = 0; i < buckets.Length; i++)
   {
       buckets[i] = new TaskCompletionSource<Task<T>>();
       results[i] = buckets[i].Task;
   }

   int nextTaskIndex = -1;
   Action<Task<T>> continuation = completed =>
   {
       var bucket = buckets[Interlocked.Increment(ref nextTaskIndex)];
       bucket.TrySetResult(completed);
   };

   foreach (var inputTask in inputTasks)
       inputTask.ContinueWith(continuation, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);

   return results;
}

Then change your WhenEach to call the Interleaved code

public static async IAsyncEnumerable<TResult> WhenEach<TResult>(Task<TResult>[] tasks)
{
    foreach (var bucket in Interleaved(tasks))
    {
        var t = await bucket;
        yield return await t;
    }
}

Then you can call your WhenEach as per usual

await foreach (int result in WhenEach(tasks))
{
    Console.WriteLine($"Processed: {result}");
}

I did some rudimentary benchmarking with 10k tasks and performed 5 times better in terms of speed.

JohanP
  • 5,252
  • 2
  • 24
  • 34
  • I am accepting this answer because it is very efficient, it runs everywhere, and it doesn't depend on external packages! – Theodor Zoulias Oct 03 '19 at 07:11
  • I realize this is an old post. I actually found Stephen Toub's article first and although his article clearly shows timestamps being produced at different points as tasks are being completed the Interleaved method does not yield results using IAsyncEnumerable. It seemingly returns the entire result array at once. Is this trickery of the returned nested task? – mrUlrik Jul 13 '23 at 12:41
6

You can use a Channel as an async queue. Each task can write to the channel when it completes. Items in the channel will be returned as an IAsyncEnumerable through ChannelReader.ReadAllAsync. 

IAsyncEnumerable<T> ToAsyncEnumerable<T>(IEnumerable<Task<T>> inputTasks)
{
    var channel=Channel.CreateUnbounded<T>();
    var writer=channel.Writer;
    var continuations=inputTasks.Select(t=>t.ContinueWith(x=>
                                           writer.TryWrite(x.Result)));
    _ = Task.WhenAll(continuations)
            .ContinueWith(t=>writer.Complete(t.Exception));

    return channel.Reader.ReadAllAsync();
}

When all tasks complete writer.Complete() is called to close the channel.

To test this, this code produces tasks with decreasing delays. This should return the indexes in reverse order :

var tasks=Enumerable.Range(1,4)
                    .Select(async i=>
                    { 
                      await Task.Delay(300*(5-i));
                      return i;
                    });

await foreach(var i in Interleave(tasks))
{
     Console.WriteLine(i);

}

Produces :

4
3
2
1
Panagiotis Kanavos
  • 120,703
  • 13
  • 188
  • 236
  • Thanks Panagiotis for the great answer! Your solution performs equally well with JohanP's solution, and is superior at memory allocation. It handles exceptions differently though. Your solution delays the propagation of all exceptions until the end of the stream, while JohanP's solution throws immediately at the first task's failure. I am not sure which behavior is more useful. The drawback of your solution is that it doesn't compile on .NET Framework, because the `Reader.ReadAllAsync` method is .NET Core specific. Is there any way to make it .NET Framework-friendly? – Theodor Zoulias Oct 02 '19 at 18:25
  • @TheodorZoulias I also wanted immediate exception propagation, so I added a solution that builds off of this one to allow for that: https://stackoverflow.com/a/62204126/1428743. Although it still uses `Reader.ReadAllAsync`, so I'm not sure it'll work for you if you still have a requirement for .NET Framework support. – PseudoPsyche Jun 04 '20 at 21:07
2

Just for the fun of it, using System.Reactive and System.Interactive.Async:

public static async IAsyncEnumerable<TResult> WhenEach<TResult>(
    Task<TResult>[] tasks)
    => Observable.Merge(tasks.Select(t => t.ToObservable())).ToAsyncEnumerable()
Paulo Morgado
  • 14,111
  • 3
  • 31
  • 59
1

I really liked the solution provided by Panagiotis, but still wanted to get exceptions raised as they happen like in JohanP's solution.

To achieve that we can slightly modify that to try closing the channel in the continuations when a task fails:

public IAsyncEnumerable<T> ToAsyncEnumerable<T>(IEnumerable<Task<T>> inputTasks)
{
    if (inputTasks == null)
    {
        throw new ArgumentNullException(nameof(inputTasks), "Task list must not be null.");
    }

    var channel = Channel.CreateUnbounded<T>();
    var channelWriter = channel.Writer;
    var inputTaskContinuations = inputTasks.Select(inputTask => inputTask.ContinueWith(completedInputTask =>
    {
        // Check whether the task succeeded or not
        if (completedInputTask.Status == TaskStatus.RanToCompletion)
        {
            // Write the result to the channel on successful completion
            channelWriter.TryWrite(completedInputTask.Result);
        }
        else
        {
            // Complete the channel on failure to immediately communicate the failure to the caller and prevent additional results from being returned
            var taskException = completedInputTask.Exception?.InnerException ?? completedInputTask?.Exception;
            channelWriter.TryComplete(taskException);
        }
    }));

    // Ensure the writer is closed after the tasks are all complete, and propagate any exceptions from the continuations
    _ = Task.WhenAll(inputTaskContinuations).ContinueWith(completedInputTaskContinuationsTask => channelWriter.TryComplete(completedInputTaskContinuationsTask.Exception));

    // Return the async enumerator of the channel so results are yielded to the caller as they're available
    return channel.Reader.ReadAllAsync();
}

The obvious downside to this is that the first error encountered will end enumeration and prevent any other, possibly successful, results from being returned. This is a tradeoff that's acceptable for my use case, but may not be for others.

PseudoPsyche
  • 4,332
  • 5
  • 37
  • 58
  • Thanks @PseudoPsyche for the answer! I noticed that it behaves strangely when it encounters a canceled task. The resulting `IAsyncEnumerable` completes successfully immediately after the completion of the first canceled task. – Theodor Zoulias Jun 05 '20 at 00:50
  • Ah, yeah, I see that now. Should be easy enough to handle by adding a check for `TaskStatus.Canceled` so it doesn't close the channel in that case. Didn't notice that since my scenario isn't using task cancelation. – PseudoPsyche Jun 05 '20 at 00:55
0

I am adding one more answer to this question, because there are a couple of issues that need to be addressed.

  1. It is recommended that methods creating async-enumerable sequences should have a CancellationToken parameter. This enables the WithCancellation configuration in await foreach loops.
  2. It is recommended that when an asynchronous operation attaches continuations to tasks, these continuations should be cleaned up when the operation completes. So if for example the caller of the WhenEach method decide to exit prematurely the await foreach loop (using break, return etc), or if the loop terminates prematurely because of an exception, we don't want to leave a bunch of dead continuations hanging around, attached to the tasks. This can be particularly important if the WhenEach is called repeatedly in a loop (as part of a Retry functionality for example).

The implementation below addresses these two issues. It is based on a Channel<Task<TResult>>. Now the channels have become an integral part of the .NET platform, so there is no reason to avoid them in favor of more complex TaskCompletionSource-based solutions.

public async static IAsyncEnumerable<TResult> WhenEach<TResult>(
    Task<TResult>[] tasks,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    if (tasks == null) throw new ArgumentNullException(nameof(tasks));
    var channel = Channel.CreateUnbounded<Task<TResult>>();
    using var completionCts = new CancellationTokenSource();
    var continuations = new List<Task>(tasks.Length);
    try
    {
        int pendingCount = tasks.Length;
        foreach (var task in tasks)
        {
            if (task == null) throw new ArgumentException(
                $"The tasks argument included a null value.", nameof(tasks));
            continuations.Add(task.ContinueWith(t =>
            {
                bool accepted = channel.Writer.TryWrite(t);
                Debug.Assert(accepted);
                if (Interlocked.Decrement(ref pendingCount) == 0)
                    channel.Writer.Complete();
            }, completionCts.Token, TaskContinuationOptions.ExecuteSynchronously |
                TaskContinuationOptions.DenyChildAttach, TaskScheduler.Default));
        }

        await foreach (var task in channel.Reader.ReadAllAsync(cancellationToken)
            .ConfigureAwait(false))
        {
            yield return await task.ConfigureAwait(false);
            cancellationToken.ThrowIfCancellationRequested();
        }
    }
    finally
    {
        completionCts.Cancel();
        try { await Task.WhenAll(continuations).ConfigureAwait(false); }
        catch (OperationCanceledException) { } // Ignore
    }
}

The finally block takes care of cancelling the attached continuations, and awaiting them to complete before exiting.

The ThrowIfCancellationRequested inside the await foreach loop might seem redundant, but it is actually required because of a by-design behavior of the ReadAllAsync method, that is explained here.

Note: The OperationCanceledException in the finally block is suppressed by an inefficient try/catch block. Catching exceptions is expensive. A more efficient implementation would suppress the error by awaiting the continuations with a specialized SuppressException awaiter, like the one featured in this answer, and special-handling the IsCanceled case. For the purpose of this answer, fixing this inefficiency is probably overkill. It's unlikely that the WhenEach method will be ever used in a tight loop.

Theodor Zoulias
  • 34,835
  • 7
  • 69
  • 104