Waiting asynchronously for some tasks to complete (Task.WhenSome)

Question

I am writing a service that combines data from various internet sources, and generates a response on the fly. Speed is more important than completeness, so I would like to generate my response as soon as some (not all) of the internet sources have responded. Typically my service creates 10 concurrent web requests, and should stop waiting and start processing after 5 of them have completed. Neither the .NET Framework, nor any of the third-party libraries I am aware of are offering this functionality, so I 'll probably have to write it myself. The method I am trying to implement has the following signature:

public static Task<TResult[]> WhenSome<TResult>(int atLeast, params Task<TResult>[] tasks)
{
    // TODO
}

Contrary to how Task.WhenAny works, exceptions should be swallowed, provided that the required number of results have been acquired. If however, after completion of all tasks, there are not enough gathered results, then an AggregateException should be thrown propagating all exceptions.

Usage example:

var tasks = new Task<int>[]
{
    Task.Delay(100).ContinueWith<int>(_ => throw new ApplicationException("Oops!")),
    Task.Delay(200).ContinueWith(_ => 10),
    Task.Delay(Timeout.Infinite).ContinueWith(_ => 0,
        new CancellationTokenSource(300).Token),
    Task.Delay(400).ContinueWith(_ => 20),
    Task.Delay(500).ContinueWith(_ => 30),
};
var results = await WhenSome(2, tasks);
Console.WriteLine($"Results: {String.Join(", ", results)}");

Expected output:

Results: 10, 20

In this example the last task returning the value 30 should be ignored (not even awaited), because we have already acquired the number of results we want (2 results). The faulted and cancelled tasks should also be ignored, for the same reason.

Answer 1

I am adding one more solution to this problem, not because stuartd's answer in not sufficient, but just for the sake of variety. This implementation uses the Unwrap technique in order to return a Task that contains all the exceptions, in exactly the same way that the built-in Task.WhenAll method propagates all the exceptions.

public static Task<TResult[]> WhenSome<TResult>(int atLeast, params Task<TResult>[] tasks)
{
    if (tasks == null) throw new ArgumentNullException(nameof(tasks));
    if (atLeast < 1 || atLeast > tasks.Length)
        throw new ArgumentOutOfRangeException(nameof(atLeast));

    var cts = new CancellationTokenSource();
    int successfulCount = 0;
    var continuationAction = new Action<Task<TResult>>(task =>
    {
        if (task.IsCompletedSuccessfully)
            if (Interlocked.Increment(ref successfulCount) == atLeast) cts.Cancel();
    });
    var continuations = tasks.Select(task => task.ContinueWith(continuationAction,
        cts.Token, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default));

    return Task.WhenAll(continuations).ContinueWith(_ =>
    {
        cts.Dispose();
        if (successfulCount >= atLeast) // Success
            return Task.WhenAll(tasks.Where(task => task.IsCompletedSuccessfully));
        else
            return Task.WhenAll(tasks); // Failure
    }, TaskScheduler.Default).Unwrap();
}

The continuations do not propagate the results or the exceptions of the tasks. These are cancelable continuations, and they are canceled en masse when the specified number of successful tasks has been reached.

Note: This implementation might propagate more than atLeast results. If you want exactly this number of results, you can chain a .Take(atLeast) after the .Where LINQ operatgor.