How can I await an array of tasks and stop waiting on first exception?
Your best bet is to build your WhenAllFailFast
method using TaskCompletionSource
. You can .ContinueWith() every input task with a synchronous continuation that errors the TCS when the tasks end in the Faulted state (using the same exception object).
Perhaps something like (not fully tested):
using System;
using System.Threading;
using System.Threading.Tasks;
namespace stackoverflow
{
class Program
{
static async Task Main(string[] args)
{
var cts = new CancellationTokenSource();
cts.Cancel();
var arr = await WhenAllFastFail(
Task.FromResult(42),
Task.Delay(2000).ContinueWith<int>(t => throw new Exception("ouch")),
Task.FromCanceled<int>(cts.Token));
Console.WriteLine("Hello World!");
}
public static Task<TResult[]> WhenAllFastFail<TResult>(params Task<TResult>[] tasks)
{
if (tasks is null || tasks.Length == 0) return Task.FromResult(Array.Empty<TResult>());
// defensive copy.
var defensive = tasks.Clone() as Task<TResult>[];
var tcs = new TaskCompletionSource<TResult[]>();
var remaining = defensive.Length;
Action<Task> check = t =>
{
switch (t.Status)
{
case TaskStatus.Faulted:
// we 'try' as some other task may beat us to the punch.
tcs.TrySetException(t.Exception.InnerException);
break;
case TaskStatus.Canceled:
// we 'try' as some other task may beat us to the punch.
tcs.TrySetCanceled();
break;
default:
// we can safely set here as no other task remains to run.
if (Interlocked.Decrement(ref remaining) == 0)
{
// get the results into an array.
var results = new TResult[defensive.Length];
for (var i = 0; i < tasks.Length; ++i) results[i] = defensive[i].Result;
tcs.SetResult(results);
}
break;
}
};
foreach (var task in defensive)
{
task.ContinueWith(check, default, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}
return tcs.Task;
}
}
}
Edit: Unwraps AggregateException, Cancellation support, return array of results. Defend against array mutation, null and empty. Explicit TaskScheduler.
Your loop waits for each of the tasks in pseudo-serial, so that's why it waits for task1 to complete before checking if task2 failed.
You might find this article helpful on a pattern for aborting after the first failure: http://gigi.nullneuron.net/gigilabs/patterns-for-asynchronous-composite-tasks-in-c/
public static async Task<TResult[]> WhenAllFailFast<TResult>(
params Task<TResult>[] tasks)
{
var taskList = tasks.ToList();
while (taskList.Count > 0)
{
var task = await Task.WhenAny(taskList).ConfigureAwait(false);
if(task.Exception != null)
{
// Left as an exercise for the reader:
// properly unwrap the AggregateException;
// handle the exception(s);
// cancel the other running tasks.
throw task.Exception.InnerException;
}
taskList.Remove(task);
}
return await Task.WhenAll(tasks).ConfigureAwait(false);
}
I recently needed once again the WhenAllFailFast
method, and I revised @ZaldronGG's excellent solution to make it a bit more performant (and more in line with Stephen Cleary's recommendations). The implementation below handles around 3,500,000 tasks per second in my PC.
public static Task<TResult[]> WhenAllFailFast<TResult>(params Task<TResult>[] tasks)
{
if (tasks is null) throw new ArgumentNullException(nameof(tasks));
if (tasks.Length == 0) return Task.FromResult(new TResult[0]);
var results = new TResult[tasks.Length];
var remaining = tasks.Length;
var tcs = new TaskCompletionSource<TResult[]>(
TaskCreationOptions.RunContinuationsAsynchronously);
for (int i = 0; i < tasks.Length; i++)
{
var task = tasks[i];
if (task == null) throw new ArgumentException(
$"The {nameof(tasks)} argument included a null value.", nameof(tasks));
HandleCompletion(task, i);
}
return tcs.Task;
async void HandleCompletion(Task<TResult> task, int index)
{
try
{
var result = await task.ConfigureAwait(false);
results[index] = result;
if (Interlocked.Decrement(ref remaining) == 0)
{
tcs.TrySetResult(results);
}
}
catch (OperationCanceledException)
{
tcs.TrySetCanceled();
}
catch (Exception ex)
{
tcs.TrySetException(ex);
}
}
}