C#：終於有人把 ValueTask、IValueTaskSource、ManualResetValueTaskSourceCore 說清楚了！

時間  2020-12-04
標籤 html git github 算法 api 服務器 網絡 併發 dom 異步 欄目 C# 简体版
原文   原文鏈接

目錄

最近 NCC 羣裏在討論 ValueTask/ ValueTask<TResult>，大帥(Natasha主要開發者)最近執着於搞算法和高性能計算，他這麼關注這個東西，說明有搞頭，揹着他偷偷學一下，省得沒話題🤣。

ValueTask/ValueTask<TResult> 出現時間其實比較早的了，以前一直沒有深刻，藉此機會好好學習一番。html

文章中說 ValueTask 時，爲了減小文字數量，通常包括其泛型版本 ValueTask<TRsult>；提到 Task，也包括其泛型版本；git

1，可用版本與參考資料

根據 Microsoft 官網的參考資料，如下版本的 .NET 程序(集)可使用 ValueTask/ValueTask<TResult>。github

版本類別 版本要求
.NET 5.0
.NET Core 2.一、3.0、3.1
.NET Standard 2.1

如下是筆者閱讀時的參考資料連接地址：算法

【1】 https://docs.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.valuetask?view=net-5.0api

【2]】 https://docs.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.valuetask-1?view=net-5.0服務器

【3】 https://www.infoworld.com/article/3565433/how-to-use-valuetask-in-csharp.html網絡

【4】 https://tooslowexception.com/implementing-custom-ivaluetasksource-async-without-allocations/併發

【5】 https://blog.marcgravell.com/2019/08/prefer-valuetask-to-task-always-and.htmldom

【6】 https://qiita.com/skitoy4321/items/31a97e03665bd7bcc8ca異步

【7】 https://neuecc.medium.com/valuetasksupplement-an-extensions-to-valuetask-4c247bc613ea

【8】https://github.com/dotnet/coreclr/blob/master/src/System.Private.CoreLib/shared/System/Threading/Tasks/Sources/ManualResetValueTaskSourceCore.cs

2，ValueTask<TResult> 和 Task

ValueTask<TResult> 存在於 System.Threading.Tasks 命名空間下，ValueTask<TResult> 的定義以下：

public struct ValueTask<TResult> : IEquatable<ValueTask<TResult>>
筆者注：IEquatable<T> 接口定義 Equals 方法，用於肯定兩個實例是否相等。

而 Task 的定義以下：

public class Task : IAsyncResult, IDisposable

從其繼承的接口和官方文檔來看，ValueTask<TResult> 複雜度應該不高。

根據文檔表面理解，這個類型，應該是 Task 的簡化版本，Task 是引用類型，所以從異步方法返回 Task 對象或者每次調用異步方法時，都會在託管堆中分配該對象。

根據比較，咱們應當知道：

  • Task 是引用類型，會在託管堆中分配內存；ValueTask 是值類型；

目前就只有這一點須要記住，下面咱們繼續比較二者的異同點。

這裏咱們嘗試一下使用這個類型對比 Task ，看看代碼如何。

public static async ValueTask<int> GetValueTaskAsync()
        {
            await Task.CompletedTask;	// 這裏別誤會，這是隨便找個地方 await 一下
            return 666;
        }

        public static async Task<int> GetTaskAsync()
        {
            await Task.CompletedTask;
            return 666;
        }

從代碼上看，二者在簡單代碼上使用的方法一致(CURD基本就是這樣)。

3，編譯器如何編譯

Task 在編譯時，由編譯器生成狀態機，會爲每一個方法生成一個繼承 IAsyncStateMachine 的類，而且出現大量的代碼包裝。

據筆者測試，ValueTask 也是生成相似的代碼。

如圖：

編譯後的Task

訪問 https://sharplab.io/#gist:ddf2a5e535a34883733196c7bf4c55b2 可在線閱讀以上代碼(Task)。

訪問 https://sharplab.io/#gist:7129478fc630a87c08ced38e7fd14cc0 在線閱讀 ValueTask 示例代碼。

你分別訪問這裏 URL，對比差別。

筆者將有差別的部分取出來了，讀者能夠認真看一下：

Task：

[AsyncStateMachine(typeof(<GetTaskAsync>d__0))]
    [DebuggerStepThrough]
    public static Task<int> GetTaskAsync()
    {
        <GetTaskAsync>d__0 stateMachine = new <GetTaskAsync>d__0();
        stateMachine.<>t__builder = AsyncTaskMethodBuilder<int>.Create();
        stateMachine.<>1__state = -1;
        AsyncTaskMethodBuilder<int> <>t__builder = stateMachine.<>t__builder;
        <>t__builder.Start(ref stateMachine);
        return stateMachine.<>t__builder.Task;
    }

ValueTask：

[AsyncStateMachine(typeof(<GetValueTaskAsync>d__0))]
    [DebuggerStepThrough]
    public static ValueTask<int> GetValueTaskAsync()
    {
        <GetValueTaskAsync>d__0 stateMachine = new <GetValueTaskAsync>d__0();
        stateMachine.<>t__builder = AsyncValueTaskMethodBuilder<int>.Create();
        stateMachine.<>1__state = -1;
        AsyncValueTaskMethodBuilder<int> <>t__builder = stateMachine.<>t__builder;
        <>t__builder.Start(ref stateMachine);
        return stateMachine.<>t__builder.Task;
    }

我是沒看出有啥區別。。。

不過這裏要提到第二點：

  • 若是這個方法的處理速度很快，或者你的代碼執行後當即可用等，使用異步並不會比同步快，反而有可能多消耗一下性能資源。

4，ValueTask 有什麼優點

從前面的內容可知，ValueTask 跟 Task 編譯後生成的狀態機代碼一致，那麼真正有區別的地方，就是 ValueTask 是值類型，Task 是引用類型。

從功能上看，ValueTask 是簡單的異步表示，而 Task 具備不少強大的方法，有各類各樣的騷操做。

ValueTask 由於不須要堆分配內存而提升了性能，這是 ValueTask 對 Task 有優點的地方。

要避免內存分配開銷，咱們可使用 ValueTask 包裝須要返回的結果。

public static ValueTask<int> GetValueTask()
        {
            return new ValueTask<int>(666);
        }

        public static async ValueTask<int> StartAsync()
        {
            return await GetValueTask();
        }

可是目前，咱們尚未進行任何性能測試，不足以說明 ValueTask 對提升性能的優點，筆者繼續講解一些基礎知識，待時機成熟後，會進行一些測試並放出示例代碼。

5，ValueTask 建立異步任務

咱們看一下 ValueTask 和 ValueTask<TResult> 的構造函數定義。

// ValueTask
        public ValueTask(Task task);
        public ValueTask(IValueTaskSource source, short token);

// ValueTask<TResult>
        public ValueTask(Task<TResult> task);
        public ValueTask(TResult result);
        public ValueTask(IValueTaskSource<TResult> source, short token);

若是經過 Task 建立任務，可使用 new Task() 、Task.Run() 等方式建立一個任務，而後就可使用 async/await 關鍵字 定義異步方法，開啓異步任務。那麼若是使用 ValueTask 呢？

第四小節咱們已經有了示例，使用了 ValueTask(TResult result) 構造函數，能夠本身 new ValueTask ，而後就可使用 await 關鍵字。

另外， ValueTask 的構造函數有多個，咱們能夠繼續挖掘一下。

經過 Task 轉換爲 ValueTask：

public static async ValueTask<int> StartAsync()
        {
            Task<int> task = Task.Run<int>(() => 666);
            return await new ValueTask<int>(task);
        }

剩下一個 IValueTaskSource 參數類型作構造函數的方法，咱們放到第 6 小節講。

ValueTask 實例僅可等待一次！必須記住這一點！

6，IValueTaskSource 和自定義包裝 ValueTask

關於 IValueTaskSource

IValueTaskSource 在 System.Threading.Tasks.Sources 命名空間中，其定義以下：

public interface IValueTaskSource
    {
        void GetResult(short token);

        ValueTaskSourceStatus GetStatus(short token);

        void OnCompleted(
            Action<object?> continuation, 
            object? state, 
            short token, 
            ValueTaskSourceOnCompletedFlags flags);
    }
方法名稱 做用
GetResult(Int16) 獲取 IValueTaskSource 的結果，僅在異步狀態機須要獲取操做結果時調用一次
GetStatus(Int16) 獲取當前操做的狀態，由異步狀態機調用以檢查操做狀態
OnCompleted(Action, Object, Int16, ValueTaskSourceOnCompletedFlags) 爲此 IValueTaskSource 計劃延續操做，開發者本身調用

在這個命名空間中，還有一些跟 ValueTask 相關的類型，可參考 微軟文檔。

在上述三個方法中，OnCompleted 用於延續任務，這個方法熟悉 Task 的讀者應該都清楚，這裏就再也不贅述。

前面咱們有一個示例：

public static ValueTask<int> GetValueTask()
        {
            return new ValueTask<int>(666);
        }

        public static async ValueTask<int> StartAsync()
        {
            return await GetValueTask();
        }

編譯器轉換後的簡化代碼：

public static int _StartAsync()
        {
            var awaiter = GetValueTask().GetAwaiter();
            if (!awaiter.IsCompleted)
            {
                // 一些莫名其妙的操做代碼
            }

            return awaiter.GetResult();
        }

基於這個代碼，咱們發現 ValueTask 能夠有狀態感知，那麼如何表達任務已經完成？裏面又有啥實現原理？

什麼是 IValueTaskSource

IValueTaskSource 是一種抽象，經過這種抽象咱們能夠將 任務/操做 的邏輯行爲和結果自己分開表示(狀態機)。

簡化示例：

IValueTaskSource<int> someSource = // ...
short token =                      // ...令牌
var vt = new ValueTask<int>(someSource, token);  // 建立任務
int value = await vt;						     // 等待任務完成

但從這段代碼來看，咱們沒法看到 如何實現 IValueTaskSource，ValueTask 內部又是如何使用 IValueTaskSource 的。在深刻其原理以前，筆者從其它博客、文檔等地方查閱到，爲了下降 Task(C#5.0引入) 的性能開銷，C# 7.0 出現了 ValueTask。ValueTask 的出現是爲了包裝返回結果，避免使用堆分配。

因此，須要使用 Task 轉換爲 ValueTask：

public ValueTask(Task task);		// ValueTask 構造函數

ValueTask 只是包裝 Task 的返回結果。

後來，爲了更高的性能，引入了 IValueTaskCource，ValueTask 便多增長了一個構造函數。

能夠經過實現 IValueTaskSource：

public ValueTask(IValueTaskSource source, short token);    // ValueTask 構造函數

這樣，能夠進一步消除 ValueTask 跟 Task 轉換的性能開銷。ValueTask 便擁有狀態「管理」能力，再也不依賴 Task 。

再說 ValueTask 優點

2019-8-22 的 coreclr 草案中，有個主題 「Make "async ValueTask/ValueTask " methods ammortized allocation-free」，深刻探討了 ValueTask 的性能影響以及後續改造計劃。

Issue 地址：https://github.com/dotnet/coreclr/pull/26310

裏面有各類各樣的性能指標比較，筆者十分推薦有興趣深刻研究的讀者看一下這個 Issue。

不要本身所有實現 IValueTaskSource

大多數人沒法完成這個接口，我我的看來不少次也沒有看懂，翻了好久，沒有找到合適的代碼示例。根據官方的文檔，我發現了 ManualResetValueTaskSourceCore，這個類型實現了 IValueTaskSource 接口，而且進行了封裝，所以咱們可使用 ManualResetValueTaskSourceCore 對本身的代碼進行包裝，更加輕鬆地實現 IValueTaskSource。

關於 ManualResetValueTaskSourceCore ，文章後面再給出使用方法和代碼示例。

ValueTaskSourceOnCompletedFlags

ValueTaskSourceOnCompletedFlags 是一個枚舉，用於表示延續的行爲，其枚舉說明以下：

枚舉 值 說明
FlowExecutionContext 2 OnCompleted 應捕獲當前 ExecutionContext 並用它來運行延續。
None 0 對延續的調用方式內有任何要求。
UseSchedulingContext 1 OnCompleted 應該捕獲當前調度上下文（SynchronizationContext），並在將延續加入執行隊列時使用。 若是未設置此標誌，實現能夠選擇執行任意位置的延續。

ValueTaskSourceStatus

ValueTaskSourceStatus 枚舉用於指示 指示 IValueTaskSource 或 IValueTaskSource 的狀態，其枚舉說明以下：

枚舉 值 說明
Canceled 3 操做因取消操做而完成。
Faulted 2 操做已完成但有錯誤。
Pending 0 操做還沒有完成。
Succeeded 1 操做已成功完成。

7，編寫 IValueTaskSource 實例

完整代碼：https://github.com/whuanle/RedisClientLearn/issues/1

假如咱們要設計一個 Redis 客戶端，而且實現異步，若是你有 Socket 開發經驗，會了解 Socket 並非 一發一收的。C# 中的 Socket 中也沒有直接的異步接口。

因此這裏咱們要實現一個異步的 Redis 客戶端。

使用 IValueTaskSource 編寫狀態機：

// 一個能夠將同步任務、不一樣線程同步操做，經過狀態機構建異步方法
    public class MyValueTaskSource<TRusult> : IValueTaskSource<TRusult>
    {
        // 存儲返回結果
        private TRusult _result;
        private ValueTaskSourceStatus status = ValueTaskSourceStatus.Pending;

        // 此任務有異常
        private Exception exception;

        #region 實現接口，告訴調用者，任務是否已經完成，以及是否有結果，是否有異常等
        // 獲取結果
        public TRusult GetResult(short token)
        {
            // 若是此任務有異常，那麼獲取結果時，從新彈出
            if (status == ValueTaskSourceStatus.Faulted)
                throw exception;
            // 若是任務被取消，也彈出一個異常
            else if (status == ValueTaskSourceStatus.Canceled)
                throw new TaskCanceledException("此任務已經被取消");

            return _result;
        }

        // 獲取狀態，這個示例中，用不到令牌 token
        public ValueTaskSourceStatus GetStatus(short token)
        {
            return status;
        }

        // 實現延續
        public void OnCompleted(Action<object> continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags)
        {
            // 不須要延續，不實現此接口
        }

        #endregion

        #region 實現狀態機，可以控制此任務是否已經完成，以及是否有異常

        // 以及完成任務，並給出結果
        public void SetResult(TRusult result)
        {
            status = ValueTaskSourceStatus.Succeeded;  // 此任務已經完成
            _result = result;
        }

        // 取消任務
        public void Cancel()
        {
            status = ValueTaskSourceStatus.Canceled;
        }

        // 要執行的任務出現異常
        public void SetException(Exception exception)
        {
            this.exception = exception;
            status = ValueTaskSourceStatus.Faulted;
        }

        #endregion

    }

假的 Socket：

public class 假的Socket
    {
        private bool IsHaveSend = false;

        // 模擬 Socket 向服務器發送數據
        public void Send(byte[] data)
        {
            new Thread(() =>
            {
                Thread.Sleep(100);
                IsHaveSend = true;
            }).Start();
        }

        // 同步阻塞等待服務器的響應
        public byte[] Receive()
        {
            // 模擬網絡傳輸的數據
            byte[] data = new byte[100];

            while (!IsHaveSend)
            {
                // 服務器沒有發送數據到客戶端時，一直空等待
            }

            // 模擬網絡接收數據耗時
            Thread.Sleep(new Random().Next(0, 100));
            new Random().NextBytes(data);
            IsHaveSend = false;
            return data;
        }
    }

實現 Redis 客戶端，而且實現

// Redis 客戶端
    public class RedisClient
    {
        // 隊列
        private readonly Queue<MyValueTaskSource<string>> queue = new Queue<MyValueTaskSource<string>>();

        private readonly 假的Socket _socket = new 假的Socket();  // 一個 socket 客戶端

        public RedisClient(string connectStr)
        {
            new Thread(() =>
            {
                while (true)
                {
                    byte[] data = _socket.Receive();
                    // 從隊列中拿出一個狀態機
                    if (queue.TryDequeue(out MyValueTaskSource<string> source))
                    {
                        // 設置此狀態機的結果
                        source.SetResult(Encoding.UTF8.GetString(data));
                    }
                }
            }).Start();
        }

        private void SendCommand(string command)
        {
            Console.WriteLine("客戶端發送了一個命令:" + command);
            _socket.Send(Encoding.UTF8.GetBytes(command));
        }

        public async ValueTask<string> GetStringAsync(string key)
        {
            // 自定義狀態機
            MyValueTaskSource<string> source = new MyValueTaskSource<string>();
            // 建立異步任務
            ValueTask<string> task = new ValueTask<string>(source, 0);

            // 加入隊列中
            queue.Enqueue(source);

            // 發送獲取值的命令
            SendCommand($"GET {key}");

            // 直接使用 await ，只會檢查移除狀態！一層必須在檢查以前完成任務，而後 await 後會陷入無限等待中！
            // return await task;

            // 要想真正實現這種異步，必須使用 SynchronizationContext 等複雜的結構邏輯！
            // 爲了不過多代碼，咱們可使用下面這種 無限 while 的方法！
            var awaiter = task.GetAwaiter();
            while (!awaiter.IsCompleted) { }

            // 返回結果
            return await task;
        }
    }

大概思路就是這樣。可是最後是沒法像 Task 那樣直接 await 的！ValueTask 只能 await 一次，而且 await 只能是最後的結果檢查！

若是咱們使用 TaskCompletionSource 寫 Task 狀態機，是能夠直接 await 的。

若是你要真正實現能夠 await 的 ValueTask，那麼編寫 IValueTasksource 時，必須實現 SynchronizationContext、TaskScheduler 等。

實現這些代碼，比較複雜，怎麼辦？微軟官方給出了一個ManualResetValueTaskSourceCore<TResult>，有了它，咱們能夠省去不少複雜的代碼！

ValueTask 是不可被取消的！

8，使用 ManualResetValueTaskSourceCore

接下來，咱們經過 ManualResetValueTaskSourceCore 改造以往的代碼，這樣咱們能夠直觀的感覺到這個類型是用來幹嗎的！

改造 MyValueTaskSource 以下：

// 一個能夠將同步任務、不一樣線程同步操做，經過狀態機構建異步方法
    public class MyValueTaskSource<TRusult> : IValueTaskSource<TRusult>
    {
        private ManualResetValueTaskSourceCore<TRusult> _source = new ManualResetValueTaskSourceCore<TRusult>();

        #region 實現接口，告訴調用者，任務是否已經完成，以及是否有結果，是否有異常等
        // 獲取結果
        public TRusult GetResult(short token)
        {
            return _source.GetResult(token);
        }

        // 獲取狀態，這個示例中，用不到令牌 token
        public ValueTaskSourceStatus GetStatus(short token)
        {
            return _source.GetStatus(token); ;
        }

        // 實現延續
        public void OnCompleted(Action<object> continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags)
        {
            _source.OnCompleted(continuation, state, token, flags);
        }

        #endregion

        #region 實現狀態機，可以控制此任務是否已經完成，以及是否有異常

        // 以及完成任務，並給出結果
        public void SetResult(TRusult result)
        {
            _source.SetResult(result);
        }

        // 要執行的任務出現異常
        public void SetException(Exception exception)
        {
            _source.SetException(exception);
        }

        #endregion
    }

以後，咱們能夠直接在 GetStringAsync 使用 await 了！

public async ValueTask<string> GetStringAsync(string key)
        {
            // 自定義狀態機
            MyValueTaskSource<string> source = new MyValueTaskSource<string>();
            // 建立異步任務
            ValueTask<string> task = new ValueTask<string>(source, 0);

            // 加入隊列中
            queue.Enqueue(source);

            // 發送獲取值的命令
            SendCommand($"GET {key}");

            return await task;
        }

到此爲止，ValueTask、IValueTaskSource、ManualResetValueTaskSourceCore，你搞明白了沒有！

有人給 ValueTask 實現了大量拓展，使得 ValueTask 擁有跟 Task 同樣多任務併發能力，例如 WhenAll、WhenAny、Factory等，拓展庫地址：https://github.com/Cysharp/ValueTaskSupplement

時間緣由（筆者通常11點就睡），本文筆者就不給出併發以及其它狀況下的 GC 和性能比較了，你們學會使用後，能夠自行測試。 可關注 NCC 公衆號，瞭解更多性能知識！

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程