Task.WhenAny -避免O(N平方公里)问题的替代办法

Question

我一直在努力提高对async代码在C#中的理解和使用，特别是如何将它集成到现有的同步代码中。

我有下面的测试程序，它基本上是来自https://learn.microsoft.com/en-us/dotnet/csharp/programming-guide/concepts/async/start-multiple-async-tasks-and-process-them-as-they-complete?pivots=dotnet-6-0的一个同步调用程序和一个LinqPad可运行包装器的测试。

void Main()
{
    var a = new A();
    
    List<string> urls = new List<string>() 
        {
            "https://learn.microsoft.com/dotnet",
            "https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.task.whenall?view=net-6.0",
            "https://stackoverflow.com/questions/11836325/await-operator-can-only-be-used-within-an-async-method"
        };
        
    a.GetUrlContentLengths(urls).Dump();
}

public class A
{   
    public int GetUrlContentLengths(IEnumerable<string> urls)
    {
        return Task.Run<int>(async() => await GetUrlContentLengthsAsync(urls)).Result;
    }
    
public async Task<int> GetUrlContentLengthsAsync(IEnumerable<string> urls)
{
    System.Net.Http.HttpClient client = new System.Net.Http.HttpClient();

    IEnumerable<Task<int>> downloadTasksQuery = urls.Select(x => ProcessUrlAsync(x, client));

    var downloadTasks = downloadTasksQuery.ToList();

    int total = 0;
    
    while (downloadTasks.Any())
    {
        Task<int> finishedTask = await Task.WhenAny(downloadTasks);
        downloadTasks.Remove(finishedTask);
        total += await finishedTask;
    }
    
    return total;
}
        
    
    public  async Task<int> ProcessUrlAsync(string url, System.Net.Http.HttpClient client)
    {
        byte[] content = await client.GetByteArrayAsync(url);
        Console.WriteLine($"{url,-60} {content.Length,10:#,#}");

        return content.Length;
    }
}

这个链接的文件描述了这样的O(n平方)问题：

我们在这里有效地创建了一个O(N2)算法:对于每个任务，搜索任务列表以删除它，这是一个O(N) 操作，我们向每个任务注册一个延续，这也是一个O(N)操作。

那么，这个对Dictionary的小更改会修复这个问题并将整个事情作为O(n)操作处理吗？

public async Task<int> GetUrlContentLengthsAsync(IEnumerable<string> urls)
    {
        System.Net.Http.HttpClient client = new System.Net.Http.HttpClient();

        IEnumerable<Task<int>> downloadTasksQuery = urls.Select(x => ProcessUrlAsync(x, client));

        var downloadTasks = downloadTasksQuery.ToDictionary(xk => xk.GetHashCode(), xv => xv);

        int total = 0;
        
        while (downloadTasks.Any())
        {
            Task<int> finishedTask = await Task.WhenAny(downloadTasks.Values);
            downloadTasks.Remove(finishedTask.GetHashCode());
            total += await finishedTask;
        }
        
        return total;
    }

Answer 1

那么，这个对Dictionary的小更改会修复这个问题并将整个事情作为O(n)操作处理吗？

不是的。搜索一个List<T>确实是一个O(n)操作，但是消除这个操作并不能消除在while循环中发生的所有O(n)操作。在Task.WhenAny方法中还隐藏了一个O(n)操作，它在减慢代码速度方面比在列表中搜索具有更大的影响(开销)。隐藏操作是将连续附加到downloadTasks集合中的所有不完整任务上，然后在任何任务完成时分离这些连续性。这是很多工作要做的，因为它涉及内存分配和同步开销，而避免它的唯一方法是避免使用WhenAny-a-循环反模式。下面是算法的另一个O(n)实现。它是O(n)，因为通过Task.WhenAll方法，每个任务只附加了一个延续：

public async Task<int> GetUrlContentLengthsAsync(IEnumerable<string> urls)
{
    HttpClient client = new();

    int total = 0;

    Task<int>[] higherOrderTasks = urls.Select(async url =>
    {
        int result = await ProcessUrlAsync(url, client).ConfigureAwait(false);
        Interlocked.Add(ref total, result);
        return result;
    }).ToArray();

    await Task.WhenAll(higherOrderTasks);

    return total;
}

为每个ProcessUrlAsync任务创建了一个更高级的任务，它封装了该任务，并合并了任务完成时应该运行的代码。await ProcessUrlAsync之后的延续可能会并发地彼此运行，因此您可能不得不同步对任何共享状态的访问，就像上面示例中的total变量一样。除非您确信您的代码将运行在将同步延续的SynchronizationContext上，否则您还应该删除.ConfigureAwait(false)。在这种特殊情况下，实际上可以完全摆脱高阶任务和共享状态，如下所示：

public async Task<int> GetUrlContentLengthsAsync(IEnumerable<string> urls)
{
    HttpClient client = new();

    Task<int>[] tasks = urls
        .Select(url => ProcessUrlAsync(url, client))
        .ToArray();

    int[] results = await Task.WhenAll(tasks);

    return results.Sum();
}