翻譯自一篇博文，原文：The performance characteristics of async methods in C#

異步系列

• 剖析C#中的異步方法
• 擴展C#中的異步方法
• C#中異步方法的性能特點
• 用一個用戶場景來說明需要註意的問題

在前兩篇中，我們介紹了C#中異步方法的內部原理，以及C#編譯器提供的可擴展性從而自定義異步方法的行為。今天我們將探討異步方法的性能特點。

正如第一篇所述，編譯器進行了大量的轉換，使異步編程體驗非常類似於同步編程。但要做到這一點，編譯器會創建一個狀態機實例，將其傳遞給異步方法的builder，然後這個builder會調用task awaiter等。很明顯，所有這些邏輯都需要成本，但是需要付出多少呢？

在TPL問世之前，異步操作通常是粗細粒度的，因此異步操作的開銷很可能可以忽略不計。但如今，即使是相對簡單的應用程序，每秒也可能有數百次或數千次異步操作。TPL在設計時考慮了這樣的工作負載，但它沒那麽神，它會有一些開銷。

要度量異步方法的開銷，我們將使用第一篇文章中使用過的例子，並加以適當修改：

public class StockPrices
{
    private const int Count = 100;
    private List<(string name, decimal price)> _stockPricesCache;
 
    // 異步版本
    public async Task<decimal> GetStockPriceForAsync(string companyId)
    {
        await InitializeMapIfNeededAsync();
        return DoGetPriceFromCache(companyId);
    }
 
    // 調用init方法的同步版本
    public decimal GetStockPriceFor(string companyId)
    {
        InitializeMapIfNeededAsync().GetAwaiter().GetResult();
        return DoGetPriceFromCache(companyId);
    }
 
    // 純同步版本
    public decimal GetPriceFromCacheFor(string companyId)
    {
        InitializeMapIfNeeded();
        return DoGetPriceFromCache(companyId);
    }
 
    private decimal DoGetPriceFromCache(string name)
    {
        foreach (var kvp in _stockPricesCache)
        {
            if (kvp.name == name)
            {
                return kvp.price;
            }
        }
 
        throw new InvalidOperationException($"Can‘t find price for ‘{name}‘.");
    }
 
    [MethodImpl(MethodImplOptions.NoInlining)]
    private void InitializeMapIfNeeded()
    {
        // 類似的初始化邏輯
    }
 
    private async Task InitializeMapIfNeededAsync()
    {
        if (_stockPricesCache != null)
        {
            return;
        }
 
        await Task.Delay(42);
 
        // 從外部數據源得到股價
        // 生成1000個元素，使緩存命中略顯昂貴
        _stockPricesCache = Enumerable.Range(1, Count)
            .Select(n => (name: n.ToString(), price: (decimal)n))
            .ToList();
        _stockPricesCache.Add((name: "MSFT", price: 42));
    }
}
 

StockPrices這個類使用來自外部數據源的股票價格來填充緩存，並提供用於查詢的API。和第一篇中的例子主要的不同就是從價格的dictionary變成了價格的list。為了度量不同形式的異步方法與同步方法的開銷，操作本身應該至少做一些工作，比如對_stockPricesCache的線性搜索。

DoGetPriceFromCache使用一個循環完成，從而避免任何對象分配。

同步 vs. 基於Task的異步版本

在第一次基準測試中，我們比較1.調用了異步初始化方法的異步方法（GetStockPriceForAsync），2.調用了異步初始化方法的同步方法（GetStockPriceFor），3.調用了同步初始化方法的同步方法。

private readonly StockPrices _stockPrices = new StockPrices();
 
public SyncVsAsyncBenchmark()
{
    // 初始化_stockPricesCache
    _stockPrices.GetStockPriceForAsync("MSFT").GetAwaiter().GetResult();
}
 
[Benchmark]
public decimal GetPricesDirectlyFromCache()
{
    return _stockPrices.GetPriceFromCacheFor("MSFT");
}
 
[Benchmark(Baseline = true)]
public decimal GetStockPriceFor()
{
    return _stockPrices.GetStockPriceFor("MSFT");
}
 
[Benchmark]
public decimal GetStockPriceForAsync()
{
    return _stockPrices.GetStockPriceForAsync("MSFT").GetAwaiter().GetResult();
}

結果如下：

                     Method |     Mean | Scaled |  Gen 0 | Allocated |
--------------------------- |---------:|-------:|-------:|----------:|
 GetPricesDirectlyFromCache | 2.177 us |   0.96 |      - |       0 B |
           GetStockPriceFor | 2.268 us |   1.00 |      - |       0 B |
      GetStockPriceForAsync | 2.523 us |   1.11 | 0.0267 |      88 B |

結果很有趣：

• 異步方法很快。GetPricesForAsync在本次測試中同步地執行完畢，比純同步方法慢了15%。
• 調用了InitializeMapIfNeededAsync的同步方法GetPricesFor的開銷甚至更小，但最奇妙的是它根本沒有任何（managed heap上的）分配（上面的結果表中的Allocated列對GetPricesDirectlyFromCache和GetStockPriceFor都為0）。

當然，你也不能說異步機制的開銷對於所有異步方法同步執行的情況都是15%。這個百分比與方法所做的工作量非常相關。如果測量一個啥都不做的異步方法和啥都不做的同步方法的開銷對比就會顯示出很大的差異。這個基準測試是想顯示執行相對較少量工作的異步方法的開銷是適度的。

為什麽InitializeMapIfNeededAsync的調用沒有任何分配？我在第一篇文章中提到過，異步方法必須在managed heap上至少分配一個對象——Task實例本身。下面我們來探索一下這個問題：

優化 #1. 可能地緩存Task實例

前面的問題的答案非常簡單：AsyncMethodBuilder對每一個成功完成的異步操作都使用同一個task實例。一個返回Task的異步方法依賴於AsyncMethodBuilder在SetResult方法中做如下邏輯的處理：

// AsyncMethodBuilder.cs from mscorlib
public void SetResult()
{
    // I.e. the resulting task for all successfully completed
    // methods is the same -- s_cachedCompleted.
            
    m_builder.SetResult(s_cachedCompleted);
}

只有對於每一個成功完成的異步方法，SetResult方法會被調用，所以每一個基於Task的方法的成功結果都可以被共享。我們可以通過下面的測試看到這一點：

[Test]
public void AsyncVoidBuilderCachesResultingTask()
{
    var t1 = Foo();
    var t2 = Foo();
 
    Assert.AreSame(t1, t2);
            
    async Task Foo() { }
}

但這並不是唯一的可能發生的優化。AsyncTaskMethodBuilder<T>做了一個類似的優化：它緩存了Task<bool>以及其他一些primitive type（原始類型）的task。比如，它緩存了整數類型的所有默認值，而且對於Task<int>還緩存了在[-1; 9)這個範圍內的值（詳見AsyncTaskMethodBuilder<T>.GetTaskForResult()）。

下面的測試證明了確實如此：

[Test]
public void AsyncTaskBuilderCachesResultingTask()
{
    // These values are cached
    Assert.AreSame(Foo(-1), Foo(-1));
    Assert.AreSame(Foo(8), Foo(8));
 
    // But these are not
    Assert.AreNotSame(Foo(9), Foo(9));
    Assert.AreNotSame(Foo(int.MaxValue), Foo(int.MaxValue));
 
    async Task<int> Foo(int n) => n;
}

你不應該過分依賴於這種行為，但是知道語言和框架的作者盡可能以各種可能的方式來優化性能總歸是好的。緩存一個任務是一種常見的優化模式，在其他地方也使用這種模式。例如，在corefx倉庫中的新的Socket實現就嚴重地依賴於這種優化，並盡可能地使用緩存任務。

優化 #2: 使用ValueTask

上面的優化只在某些情況下有用。與其依賴於它，我們還可以使用ValueTask<T>：一個特殊的“類task”的類型，如果方法是同步地執行完畢，那麽就不會有額外的分配。

我們其實可以把ValueTask<T>看作T和Task<T>的聯和：如果“value task”已經完成，那麽底層的value就會被使用。如果底層的任務還沒有完成，那麽一個Task實例就會被分配。

當操作同步地執行完畢時，這個特殊類型能幫助避免不必要的分配。要使用ValueTask<T>，我們只需要把GetStockPriceForAsync的返回結果從Task<decimal改為ValueTask<decimal>：

public async ValueTask<decimal> GetStockPriceForAsync(string companyId)
{
    await InitializeMapIfNeededAsync();
    return DoGetPriceFromCache(companyId);
}

然後我們就可以用一個額外的基準測試來衡量差異：

[Benchmark]
public decimal GetStockPriceWithValueTaskAsync_Await()
{
    return _stockPricesThatYield.GetStockPriceValueTaskForAsync("MSFT").GetAwaiter().GetResult();
}

                          Method |     Mean | Scaled |  Gen 0 | Allocated |
-------------------------------- |---------:|-------:|-------:|----------:|
      GetPricesDirectlyFromCache | 1.260 us |   0.90 |      - |       0 B |
                GetStockPriceFor | 1.399 us |   1.00 |      - |       0 B |
           GetStockPriceForAsync | 1.552 us |   1.11 | 0.0267 |      88 B |
 GetStockPriceWithValueTaskAsync | 1.519 us |   1.09 |      - |       0 B |

你可以看到，返回ValueTask的方法比返回Task的方法稍快一點。主要的差別在於避免了堆上的內存分配。我們稍後將討論是否值得進行這樣的轉換，但在此之前，我想介紹一種技巧性的優化。

優化 #3: 在一個通常的路徑上避免異步機制（avoid async machinery on a common path）

如果你有一個非常廣泛使用的異步方法，並且希望進一步減少開銷，也許你可以考慮下面的優化：你可以去掉async修飾符，在方法中檢查task的狀態，並且將整個操作同步執行，從而完全不需要用到異步機制。

聽起來很復雜？來看一個例子：

public ValueTask<decimal> GetStockPriceWithValueTaskAsync_Optimized(string companyId)
{
    var task = InitializeMapIfNeededAsync();
 
    // Optimizing for a common case: no async machinery involved.
    if (task.IsCompleted)
    {
        return new ValueTask<decimal>(DoGetPriceFromCache(companyId));
    }
 
    return DoGetStockPricesForAsync(task, companyId);
 
    async ValueTask<decimal> DoGetStockPricesForAsync(Task initializeTask, string localCompanyId)
    {
        await initializeTask;
        return DoGetPriceFromCache(localCompanyId);
    }
}

在這個例子中，GetStockPriceWithValueTaskAsync_Optimized方法沒有async修飾符，它從InitializeMapIfNeededAsync方法中得到一個task的時候，檢查這個task是否已完成，如果已經完成，它就調用DoGetPriceFromCache直接立刻得到結果。但如果這個task還沒有完成，它就調用一個本地函數（local function，從C# 7.0開始支持），然後等待結果。

使用本地函數不是唯一的選擇但是是最簡單的。但有個需要註意的，就是本地函數的最自然的實現會捕獲一個閉包狀態：局部變量和參數：

public ValueTask<decimal> GetStockPriceWithValueTaskAsync_Optimized2(string companyId)
{
    // Oops! This will lead to a closure allocation at the beginning of the method!
    var task = InitializeMapIfNeededAsync();
 
    // Optimizing for acommon case: no async machinery involved.
    if (task.IsCompleted)
    {
        return new ValueTask<decimal>(DoGetPriceFromCache(companyId));
    }
 
    return DoGetStockPricesForAsync();
 
    async ValueTask<decimal> DoGetStockPricesForAsync() // 註意這次捕獲了外部的局部變量
    {
        await task;
        return DoGetPriceFromCache(companyId);
    }
}

但很不幸，由於一個編譯器bug，這段代碼即使是從通常的路徑上（即if字句中）完成的，依然會分配一個閉包（closure）。下面是這個方法被編譯器轉換後的樣子：

public ValueTask<decimal> GetStockPriceWithValueTaskAsync_Optimized(string companyId)
{
    var closure = new __DisplayClass0_0()
    {
        __this = this,
        companyId = companyId,
        task = InitializeMapIfNeededAsync()
    };
 
    if (closure.task.IsCompleted)
    {
        return ...
    }
 
    // The rest of the code
}

編譯器為給定範圍中的所有局部變量/參數使用一個共享的閉包實例。所以上面的代碼雖然看起來是有道理的，但是它使堆分配（heap allocation）的避免變得不可能。

提示：這種優化技巧性非常強。好處非常小，而且即使你寫的本地函數是沒問題的，在未來你也很可能進行修改，然後意外地捕獲了外部變量，於是造成堆分配。如果你在寫一個像BCL那樣的高度可復用的類庫，你依然可以用這個技巧來優化那些肯定會被用在熱路徑（hot path）上的方法。

等待一個task的開銷

到目前為止我們只討論了一個特殊情況：一個同步地執行完畢的異步方法的開銷。這是故意的。異步方法越小，其總體的性能開銷就越明顯。細粒度異步方法做的事相對來說較少，更容易同步地完成。我們也會相對更加頻繁地調用他們。

但我們也應該知道當一個方法等待一個未完成的task時的異步機制的性能開銷。為了度量這個開銷，我們將InitializeMapIfNeededAsync修改為調用Task.Yield()：

private async Task InitializeMapIfNeededAsync()
{
    if (_stockPricesCache != null)
    {
        await Task.Yield();
        return;
    }
 
    // Old initialization logic
}

讓我們為我們的性能基準測試添加以下的幾個方法：

[Benchmark]
public decimal GetStockPriceFor_Await()
{
    return _stockPricesThatYield.GetStockPriceFor("MSFT");
}
 
[Benchmark]
public decimal GetStockPriceForAsync_Await()
{
    return _stockPricesThatYield.GetStockPriceForAsync("MSFT").GetAwaiter().GetResult();
}
 
[Benchmark]
public decimal GetStockPriceWithValueTaskAsync_Await()
{
    return _stockPricesThatYield.GetStockPriceValueTaskForAsync("MSFT").GetAwaiter().GetResult();
}

                                    Method |      Mean | Scaled |  Gen 0 |  Gen 1 | Allocated |
------------------------------------------ |----------:|-------:|-------:|-------:|----------:|
                          GetStockPriceFor |  2.332 us |   1.00 |      - |      - |       0 B |
                     GetStockPriceForAsync |  2.505 us |   1.07 | 0.0267 |      - |      88 B |
           GetStockPriceWithValueTaskAsync |  2.625 us |   1.13 |      - |      - |       0 B |
                    GetStockPriceFor_Await |  6.441 us |   2.76 | 0.0839 | 0.0076 |     296 B |
               GetStockPriceForAsync_Await | 10.439 us |   4.48 | 0.1577 | 0.0122 |     553 B |
     GetStockPriceWithValueTaskAsync_Await | 10.455 us |   4.48 | 0.1678 | 0.0153 |     577 B |

正如我們所見，在速度和內存方面，差異都是顯而易見的。下面是對結果的簡短解釋。

• 每一個對未完成的task的“await”操作大概需要4us並且每次調用分配了約300B（依賴於平臺（x64 vs. x86 ），以及異步方法中的局部變量或參數）的內存。這解釋了為什麽GetStockPriceFor約為GetStockPriceForAsync的兩倍快，並分配更少的內存。
• 當異步方法不是同步地執行完畢時，基於ValueTask的異步方法比基於Task的稍慢。因為基於ValueTask的異步方法的狀態機需要保存更多數據。

異步方法性能的總結

• 如果異步方法同步地執行完畢，額外的開銷相當小。
• 如果異步方法同步地執行完畢，以下內存開銷會發生：對async Task來說沒有額外開銷，對async Task<T來說，每個異步操作導致88 bytes的開銷（x64平臺）。
• 對於同步執行完畢的異步方法，ValueTask<T>可以消除上一條中的額外開銷。
• 如果方法是同步執行完畢的，那麽一個基於ValueTask<T>的異步方法比基於Task<T>的方法稍快；如果是異步執行完畢的，則稍慢。
• 等待未完成的task的異步方法的性能開銷相對大得多（在x64平臺，每個操作需要300 bytes）。

一如既往地，記得先進行性能測試。如果你發現異步操作造成了性能問題，你可以從ValueTask<T>切換到ValueTask<T>，緩存一個task或是新增一個通常的執行路徑（如果可能的話）。但你也可以嘗試將異步操作粗粒度化。這可以提高性能，簡化調試，並使代碼更好理解。並不是每一小段代碼都必須是異步的。

其他參考資料

• Dissecting the async methods in C#
• Extending the async methods in C#
• Stephen Toub‘s comment about ValueTask‘s usage scenarios
• "Dissecting the local functions in C#"

[翻譯]C#中異步方法的性能特點