如何使用谓词实现 Task.WhenAny()

Question

我想同时执行几个异步任务。每个任务都会运行一个可以成功完成或抛出异常的 HTTP 请求。我需要await 直到第一个任务成功完成，或者直到所有任务都失败。

如何实现接受谓词的Task.WhenAny 方法的重载，以便排除未成功完成的任务？

Answer 1

等待任何任务，如果条件满足则返回任务。否则再次等待其他任务，直到没有其他任务等待。

public static async Task<Task> WhenAny( IEnumerable<Task> tasks, Predicate<Task> condition )
{
    var tasklist = tasks.ToList();
    while ( tasklist.Count > 0 )
    {
        var task = await Task.WhenAny( tasklist );
        if ( condition( task ) )
            return task;
        tasklist.Remove( task );
    }
    return null;
}

简单检查

var tasks = new List<Task> {
    Task.FromException( new Exception() ),
    Task.FromException( new Exception() ),
    Task.FromException( new Exception() ),
    Task.CompletedTask, };

var completedTask = WhenAny( tasks, t => t.Status == TaskStatus.RanToCompletion ).Result;

if ( tasks.IndexOf( completedTask ) != 3 )
    throw new Exception( "not expected" );

Answer 2

public static Task<Task<T>> WhenFirst<T>(IEnumerable<Task<T>> tasks, Func<Task<T>, bool> predicate)
{
    if (tasks == null) throw new ArgumentNullException(nameof(tasks));
    if (predicate == null) throw new ArgumentNullException(nameof(predicate));

    var tasksArray = (tasks as IReadOnlyList<Task<T>>) ?? tasks.ToArray();
    if (tasksArray.Count == 0) throw new ArgumentException("Empty task list", nameof(tasks));
    if (tasksArray.Any(t => t == null)) throw new ArgumentException("Tasks contains a null reference", nameof(tasks));

    var tcs = new TaskCompletionSource<Task<T>>();
    var count = tasksArray.Count;

    Action<Task<T>> continuation = t =>
        {
            if (predicate(t))
            {
                tcs.TrySetResult(t);
            }
            if (Interlocked.Decrement(ref count) == 0)
            {
                tcs.TrySetResult(null);
            }
        };

    foreach (var task in tasksArray)
    {
        task.ContinueWith(continuation);
    }

    return tcs.Task;
}

示例用法：

var task = await WhenFirst(tasks, t => t.Status == TaskStatus.RanToCompletion);

if (task != null)
    var value = await task;

请注意，这不会传播失败任务的异常（就像 WhenAny 不会传播一样）。

您还可以为非通用 Task 创建一个此版本。

Answer 3

public static Task<T> GetFirstResult<T>(
    ICollection<Func<CancellationToken, Task<T>>> taskFactories, 
    Predicate<T> predicate) where T : class
{
    var tcs = new TaskCompletionSource<T>();
    var cts = new CancellationTokenSource();

    int completedCount = 0;
    // in case you have a lot of tasks you might need to throttle them 
    //(e.g. so you don't try to send 99999999 requests at the same time)
    // see: http://stackoverflow.com/a/25877042/67824
    foreach (var taskFactory in taskFactories)
    {
        taskFactory(cts.Token).ContinueWith(t => 
        {
            if (t.Exception != null)
            {
                Console.WriteLine($"Task completed with exception: {t.Exception}");
            }
            else if (predicate(t.Result))
            {
                cts.Cancel();
                tcs.TrySetResult(t.Result);
            }

            if (Interlocked.Increment(ref completedCount) == taskFactories.Count)
            {
                tcs.SetException(new InvalidOperationException("All tasks failed"));
            }

        }, cts.Token);
    }

    return tcs.Task;
}

示例用法：

using System.Net.Http;
var client = new HttpClient();
var response = await GetFirstResult(
    new Func<CancellationToken, Task<HttpResponseMessage>>[] 
    {
        ct => client.GetAsync("http://microsoft123456.com", ct),
        ct => client.GetAsync("http://microsoft123456.com", ct),
        ct => client.GetAsync("http://microsoft123456.com", ct),
        ct => client.GetAsync("http://microsoft123456.com", ct),
        ct => client.GetAsync("http://microsoft123456.com", ct),
        ct => client.GetAsync("http://microsoft123456.com", ct),
        ct => client.GetAsync("http://microsoft123456.com", ct),
        ct => client.GetAsync("http://microsoft.com", ct),
        ct => client.GetAsync("http://microsoft123456.com", ct),
        ct => client.GetAsync("http://microsoft123456.com", ct),
    }, 
    rm => rm.IsSuccessStatusCode);
Console.WriteLine($"Successful response: {response}");

Answer 4

这是对优秀 Eli Arbel 的answer 的尝试改进。这些是改进点：

1. predicate 中的异常作为返回任务的故障传播。
2. 在任务被接受为结果后不会调用predicate。
3. predicate 在原来的SynchronizationContext 中执行。这使得访问 UI 元素成为可能（如果从 UI 线程调用 WhenFirst 方法）
4. 直接枚举源IEnumerable<Task<T>>，无需先转换为数组。

public static Task<Task<T>> WhenFirst<T>(IEnumerable<Task<T>> tasks,
    Func<Task<T>, bool> predicate)
{
    if (tasks == null) throw new ArgumentNullException(nameof(tasks));
    if (predicate == null) throw new ArgumentNullException(nameof(predicate));

    var tcs = new TaskCompletionSource<Task<T>>(
        TaskCreationOptions.RunContinuationsAsynchronously);
    var pendingCount = 1; // The initial 1 represents the enumeration itself
    foreach (var task in tasks)
    {
        if (task == null) throw new ArgumentException($"The {nameof(tasks)}" +
            " argument included a null value.", nameof(tasks));
        Interlocked.Increment(ref pendingCount);
        HandleTaskCompletion(task);
    }
    if (Interlocked.Decrement(ref pendingCount) == 0) tcs.TrySetResult(null);
    return tcs.Task;

    async void HandleTaskCompletion(Task<T> task)
    {
        try
        {
            await task; // Continue on the captured context
        }
        catch { } // Ignore exception

        if (tcs.Task.IsCompleted) return;

        try
        {
            if (predicate(task))
                tcs.TrySetResult(task);
            else
                if (Interlocked.Decrement(ref pendingCount) == 0)
                    tcs.TrySetResult(null);
        }
        catch (Exception ex)
        {
            tcs.TrySetException(ex);
        }
    }
}

Answer 5

这样做的另一种方式，与Sir Rufo's answer 非常相似，但使用AsyncEnumerable 和Ix.NET

实现一个小助手方法，以便在任何任务完成后立即对其进行流式传输：

static IAsyncEnumerable<Task<T>> WhenCompleted<T>(IEnumerable<Task<T>> source) =>
    AsyncEnumerable.Create(_ =>
    {
        var tasks = source.ToList();
        Task<T> current = null;
        return AsyncEnumerator.Create(
            async () => tasks.Any() && tasks.Remove(current = await Task.WhenAny(tasks)), 
            () => current,
            async () => { });
    });
}

然后可以按完成顺序处理任务，例如根据要求返回第一个匹配的：

await WhenCompleted(tasks).FirstOrDefault(t => t.Status == TaskStatus.RanToCompletion)

Answer 6

只是想补充一些使用 List.Remove 的 @Peebo 和 @SirRufo 的答案（因为我还不能发表评论）

我会考虑使用：

var tasks = source.ToHashSet();

代替：

var tasks = source.ToList();

所以删除会更有效