为什么 Task.WhenAny 在针对同一个 TaskCompletionSource 多次调用时如此缓慢？答案

【问题标题】：Why is Task.WhenAny so slow when called many times against the same TaskCompletionSource?为什么 Task.WhenAny 在针对同一个 TaskCompletionSource 多次调用时如此缓慢？
【发布时间】：2017-07-22 12:47:41
【问题描述】：

如果一个类的成员 TaskCompletionSource<TResult> m_tcs 的生命周期很长，并且如果调用 Task.WhenAny 时使用 m_tcs.Task 作为其参数之一，那么当调用次数超过 50,000 次左右时，性能似乎会呈指数级下降。

为什么在这种情况下它这么慢？是否有替代方案运行速度更快但不使用 4 倍以上的内存？

我的想法是，Task.WhenAny 可能会在m_tcs.Task 之间添加和删除如此多的延续，并且在其中某处会导致 O(N²) 的复杂性。

通过将 TCS 包装在等待 m_tcs.Task 的异步函数中，我找到了一个性能更高的替代方案。它使用大约 4 倍的内存，但在超过 20,000 次迭代后运行速度快得多。

下面的示例代码（为了获得准确的结果，请直接编译并运行 .exe，无需附加调试器）。请注意，WhenAnyMemberTcsDirect 存在性能问题，WhenAnyMemberTcsIndirect 是更快的替代方案，WhenAnyLocalTcs 是用于比较的基准：

using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading.Tasks;

public class WithTcs
{
    // long-lived TaskCompletionSource
    private readonly TaskCompletionSource<bool> m_tcs = new TaskCompletionSource<bool>();

    // this has performance issues for large N - O(N^2)
    public async Task WhenAnyMemberTcsDirectAsync(Task task)
    {
        await await Task.WhenAny(task, m_tcs.Task).ConfigureAwait(false);
    }

    // performs faster - O(N), but uses 4x memory
    public async Task WhenAnyMemberTcsIndirectAsync(Task task)
    {
        await await Task.WhenAny(task, AwaitTcsTaskAsync(m_tcs)).ConfigureAwait(false);
    }

    private async Task<TResult> AwaitTcsTaskAsync<TResult>(TaskCompletionSource<TResult> tcs)
    {
        return await tcs.Task.ConfigureAwait(false);
    }

    // baseline for comparison using short-lived TCS
    public async Task WhenAnyLocalTcsAsync(Task task)
    {
        var tcs = new TaskCompletionSource<bool>();
        await await Task.WhenAny(task, tcs.Task).ConfigureAwait(false);
    }
}

class Program
{
    static void Main(string[] args)
    {
        show_warning_if_debugger_attached();

        MainAsync().GetAwaiter().GetResult();

        show_warning_if_debugger_attached();
        Console.ReadLine();
    }

    static async Task MainAsync()
    {
        const int n = 100000;

        Console.WriteLine("Running Task.WhenAny tests ({0:#,0} iterations)", n);
        Console.WriteLine();

        await WhenAnyLocalTcs(n).ConfigureAwait(false);

        await Task.Delay(1000).ConfigureAwait(false);

        await WhenAnyMemberTcsIndirect(n).ConfigureAwait(false);

        await Task.Delay(1000).ConfigureAwait(false);

        await WhenAnyMemberTcsDirect(n).ConfigureAwait(false);
    }

    static Task WhenAnyLocalTcs(int n)
    {
        Func<WithTcs, Task, Task> function =
            (instance, task) => instance.WhenAnyLocalTcsAsync(task);

        return RunTestAsync(n, function);
    }

    static Task WhenAnyMemberTcsIndirect(int n)
    {
        Func<WithTcs, Task, Task> function =
            (instance, task) => instance.WhenAnyMemberTcsIndirectAsync(task);

        return RunTestAsync(n, function);
    }

    static Task WhenAnyMemberTcsDirect(int n)
    {
        Func<WithTcs, Task, Task> function =
            (instance, task) => instance.WhenAnyMemberTcsDirectAsync(task);

        return RunTestAsync(n, function);
    }

    static async Task RunTestAsync(int n, Func<WithTcs, Task, Task> function, [CallerMemberName] string name = "")
    {
        Console.WriteLine(name);

        var tasks = new Task[n];
        var sw = new Stopwatch();
        var startBytes = GC.GetTotalMemory(true);
        sw.Start();

        var instance = new WithTcs();
        var step = n / 78;
        for (int i = 0; i < n; i++)
        {
            var iTemp = i;
            Task primaryTask = Task.Run(() => { if (iTemp % step == 0) Console.Write("."); });
            tasks[i] = function(instance, primaryTask);
        }

        await Task.WhenAll(tasks).ConfigureAwait(false);
        Console.WriteLine();

        var endBytes = GC.GetTotalMemory(true);
        sw.Stop();
        GC.KeepAlive(instance);
        GC.KeepAlive(tasks);

        Console.WriteLine("  Time: {0,7:#,0} ms, Memory: {1,10:#,0} bytes", sw.ElapsedMilliseconds, endBytes - startBytes);
        Console.WriteLine();
    }

    static void show_warning_if_debugger_attached()
    {
        if (Debugger.IsAttached)
            Console.WriteLine("WARNING: running with the debugger attached may result in inaccurate results\r\n".ToUpper());
    }
}

示例结果：

迭代 | WhenAny* 方法 |时间（毫秒）|内存（字节） ---------: | ------------------ | --------: | -------------: 1,000 |本地Tcs | 21 | 58,248 1,000 | MemberTcs间接 | 54 | 217,268 1,000 |会员TcsDirect | 21 | 52,496 10,000 |本地Tcs | 91 | 545,836 10,000 | MemberTcs间接 | 98 | 2,141,836 10,000 |会员TcsDirect | 140 | 545,640 100,000 |本地Tcs | 210 | 4,898,512 100,000 | MemberTcs间接 | 502 | 21,426,316 100,000 |会员TcsDirect | 14,090 | 5,085,396 200,000 |本地Tcs | 366 | 9,630,872 200,000 | MemberTcs间接 |第659章41,450,916 200,000 |会员TcsDirect | 42,599 | 10,069,248 500,000 |本地Tcs | 808 | 23,670,492 500,000 | MemberTcs间接 | 1,906 | 97,339,192 500,000 |会员TcsDirect | 288,373 | 24,968,436 1,000,000 |本地Tcs | 1,642 | 47,272,744 1,000,000 | MemberTcs间接 | 3,149 | 200,480,888 1,000,000 |会员TcsDirect | 1,268,030 | 48,064,772

注意：针对 .NET 4.6.2 版本（任何 CPU），在 Windows 7 SP1 64 位、Intel Core i7-4770 上测试。

【问题讨论】：

请注意，我也可以在 .NET 4.7 和 windows 10 creators edition 机器上的 netcoreapp1.1 项目上重新创建相同的行为。
执行环境是什么？有同步上下文吗？ @ScottChamberlain，你的情况有吗？
@MikeHenry，您的测试模型很可能受到the thread pool stuttering issue 的影响，请尝试增加和固定线程池的大小（例如，ThreadPool.SetMaxThreads(100, 100); ThreadPool.SetMinThreads(100, 100);）。它会改变控制台应用程序的结果吗？
@Noseratio，感谢您的建议。为线程池设置最小/最大线程确实提高了控制台应用程序的性能，因此它接近 WinForms 中的 WhenAnyMemberTcsDirect 的性能。（使用 ConfigureAwait(false) 进行 500K 迭代时为 68 秒，使用 ConfigureAwait(true) 进行 500K 迭代时为 73 秒）
@Noseratio ThreadPool 口吃听起来确实是一个因素。使用 GC.{TryStart|End}NoGCRegion 似乎会稍微影响性能，但后来我的测试已经在使用 GC.KeepAlive。此外，我必须针对 x64 而不是 Any CPU 才能使用 NoGCRegion 进行这么多次迭代。

标签： .net performance async-await task-parallel-library

【解决方案1】：

我找到了一个似乎运行速度很快（O(N) 时间）和大约运行时间的解决方案。相同的内存空间，通过在TaskCompletionSource 旁边使用成员CancellationTokenSource m_cts。任何以前设置m_tcs 取消/故障/结果的调用都需要附有m_cts.Cancel()。这当然可以抽象出来。

解决办法：

public class WithTcs
{
    // ... same as above, plus below

    private readonly CancellationTokenSource m_cts = new CancellationTokenSource();

    public async Task WhenAnyMemberCtsAsync(Task task)
    {
        var ct = m_cts.Token;
        var tcs = new TaskCompletionSource<bool>();
        using (ct.Register(() => tcs.TrySetFrom(m_tcs)))
            await await Task.WhenAny(task, tcs.Task).ConfigureAwait(false);
    }
}

public static class TcsExtensions
{
    public static bool TrySetFrom<TResult>(this TaskCompletionSource<TResult> dest, TaskCompletionSource<TResult> source)
    {
        switch (source.Task.Status)
        {
            case TaskStatus.Canceled:
                return dest.TrySetCanceled();
            case TaskStatus.Faulted:
                return dest.TrySetException(source.Task.Exception.InnerExceptions);
            case TaskStatus.RanToCompletion:
                return dest.TrySetResult(source.Task.Result);
            default:
                return false; // TCS has not yet completed
        }
    }
}

这回答了是否存在内存高效的快速替代方案的问题。我仍然很好奇 WhenAnyMemberTcsDirect 的幕后发生了什么导致 O(N²) 问题。

【讨论】：