Pipelines - .NET中的新IO API指引 作者 marcgravell  原文

此係列前兩篇網上已有的譯文

Pipelines - .NET中的新IO API指引（一）

Pipelines - .NET中的新IO API指引（二）

關於System.IO.Pipelines的一篇說明

System.IO.Pipelines: .NET高效能IO

本篇不是翻譯，邊看邊譯邊記而已。

System.IO.Pipelines 是對IO的統一抽象，檔案、com口、網路等等，重點在於讓呼叫者注意力集中在讀、寫緩衝區上，典型的就是 IDuplexPipe中的Input Output。

可以理解為將IO類抽象為讀、寫兩個緩衝區。

目前官方實現還處於preview狀態，作者使用Socket和NetworkStream 實現了一個 Pipelines.Sockets.Unofficial

作者在前兩篇中提到使用System.IO.Pipelines 改造StackExchange.Redis,在本篇中作者採用了改造現有的SimplSockets庫來說明System.IO.Pipelines的使用。

 

文章中的程式碼（SimplPipelines，KestrelServer ）

## SimplSockets說明

+ 可以單純的傳送（Send），也可以完成請求/響應處理（SendRecieve）
+ 同步Api
+ 提供簡單的幀協議封裝訊息資料
+ 使用byte[]
+ 服務端可以向所有客戶端廣播訊息
+ 有心跳檢測等等     屬於非常典型的傳統Socket庫。

## 作者的改造說明

### 對緩衝區資料進行處理的一些方案及選型

   1. 使用byte[]拷貝出來，作為獨立的資料副本使用，簡單易用但成本高（分配和複製）
   2. 使用 ReadOnlySequence<byte> ，零拷貝，快速但有限制。一旦在管道上執行Advance操作，資料將被回收。在有嚴格控制的服務端處理場景（資料不會逃離請求上下文）下可以使用，言下之意使用要求比較高。
   3. 作為2的擴充套件方案，將資料載荷的解析處理程式碼移至類庫中（處理ReadOnlySequence<byte>），只需將解構完成的資料釋出出來，也許需要一些自定義的structs 對映（map）一下。這裡說的應該是直接將記憶體對映為Struct？
   4. 通過返回Memory<byte> 獲取一份資料拷貝，也許需要從ArrayPool<byte>.Share 池中返回一個大陣列；但是這樣對呼叫者要求較高，需要返回池。並且從Memory<T> 獲取一個T[]屬於高階和不安全的操作。不安全，有風險。（ not all Memory<T> is based on T[]）
   5. 一個妥協方案，返回一個提供Memory<T>(Span<T>)的東西，並且使用一些明確的顯而易見的Api給使用者,這樣使用者就知道應該如何處理返回結果。比如IDisposable/using這種，在Dispose()被呼叫時將資源歸還給池。
  
   作者認為，設計一個通用的訊息傳遞Api時，方案5更為合理，呼叫方可以儲存一段時間的資料並且不會干擾到管道的正常工作，也可以很好的利用ArrayPool。如果呼叫者沒有使用using也不會有什麼大麻煩，只是效率會降低一些，就像使用方案1一樣。
     但是方案的選擇需要充分考慮你的實際場景，比如在StackExchange.Redis 客戶端中使用的是方案3；在不允許資料離開請求上下文時使用方案2.。
  一旦選定方案，以後基本不可能再更改。    針對效率最高的方案2，作者提出的專業建議是 **使用ref struct** 。    此處選擇的是方案5，與方案4的區別就是對Memory<T> 的處理，作者使用 System.Buffers.IMemoryOwner<T>介面
 

 public
 
 interface IMemoryOwner<T> : IDisposable
 {
  Memory<T> Memory { get; }
 }

 

   以下為實現程式碼，Dispose時歸還借出的陣列，並考慮執行緒安全，避免多次歸還(very bad)。
  

private sealed class ArrayPoolOwner<T>:IMemoryOwner<T>{
 private readonly int _length;
 private T[] _oversized;
 internal ArrayPoolOwner(T[] oversized,int length){
  _length=length;
  _oversized=oversized;
 }
 public Memory<T> Memory=>new Memory<T>(GetArray(),0,_length);
 private T[] GetArray()=>Interlocked.CompareExchange(ref _oversized,null,null)
  ?? throw new ObjectDisposedException(ToString());
 public void Dispose(){
  var arr=Interlocked.Exchange(ref _oversized,null);
  if(arr!=null) ArrayPool<T>.Shared.Return(arr);
 }
}

 

  Dispose後如果再次呼叫Memory將會失敗，即 使用時 using，不要再次使用。 **對ArrayPool的一些說明**
+ 從ArrayPool借出的陣列比你需要的要大，你給定的大小在ArrayPool看來屬於下限（不可小於你給定的大小）,見：ArrayPool<T>.Shared.Rent(int minimumLength)；
+ 歸還時陣列預設不清空，因此你借出的陣列內可能會有垃圾資料；如果需要清空，在歸還時使用 ArrayPool<T>.Shared.Return(arr，true) ；   作者對ArrayPool的一些建議： 
增加 IMemoryOwner<T> RentOwned(int length)，T[] Rent(int minimumLength) 及借出時清空陣列，歸還時清空陣列的選項。
 這裡的想法是通過IMemoryOwner<T>實現一種所有權的轉移，典型呼叫方法如下
 

 void DoSomething(IMemoryOwner<byte> data){
  using(data){
       // ... other things here ...
                DoTheThing(data.Memory);
  }
  // ... more things here ...
 }

 

 通過ArrayPool的借、還機制避免頻繁分配。    **作者的警告：**
 + 不要把data.Memory 單獨取出亂用，using完了就不要再操作它了（這種錯誤比較基礎）
 + 有人會用MemoryMarshal搞出陣列使用，作者認為可以實現一個 MemoryManager<T>（ArrayPoolOwner<T> : MemoryManager<T>, since MemoryManager<T> : IMemoryOwner<T>）讓.Span如同.Memory一樣失敗。
 ---- 作者也挺糾結（周道）的 ：）。 使用  ReadOnlySequence<T> 填充ArrayPoolOwner（構造,例項化）
 

public static IMemoryOwner<T> Lease<T>(this ReadOnlySequence<T> source)
    {
        if (source.IsEmpty) return Empty<T>();
        int len = checked((int)source.Length);
        var arr = ArrayPool<T>.Shared.Rent(len);//借出
        source.CopyTo(arr);
        return new ArrayPoolOwner<T>(arr, len);//dispose時歸還
    }

 

### 基本API

  服務端和客戶端雖然不同但程式碼有許多重疊的地方，比如都需要某種執行緒安全機制的寫入，需要某種讀迴圈來處理接收的資料，因此可以共用一個基類。
基類中使用IDuplexPipe（包括input，output兩個管道）作為管道。

public abstract class SimplPipeline : IDisposable
    {
        private IDuplexPipe _pipe;
        protected SimplPipeline(IDuplexPipe pipe)
            => _pipe = pipe;
        public void Dispose() => Close();
        public void Close() {/* burn the pipe*/}
    }

 

首先，需要一種執行緒安全的寫入機制並且不會過度阻塞呼叫方。在原SimplSockets（包括StackExchange.Redis v1）中使用訊息佇列來處理。呼叫方Send時同步的將訊息入隊，在將來的某刻，訊息出隊並寫入到Socket中。此方式存在的問題
+ 有許多移動的部分
+ 與“pipelines”有些重複 管道本身即是佇列，本身具備輸出（寫、傳送）緩衝區，沒必要再增加一個佇列，直接把資料寫入管道即可。取消原有佇列只有一些小的影響，在StackExchange.Redis v1 中使用佇列完成優先順序排序處理（佇列跳轉），作者表示不擔心這一點。 **寫入Api設計**
 + 不一定時同步的
 + 呼叫方可以單純的傳入一段記憶體資料（ReadOnlyMember<byte>），或者是一個（IMemoryOwner<byte>）由Api寫入後進行清理。
  + 先假設讀、寫分開（暫不考慮響應）

protected async ValueTask WriteAsync(IMemoryOwner<byte> payload, int messageId)//呼叫方不再使用payload，需要我們清理
    {
        using (payload)
  {
   await WriteAsync(payload.Memory, messageId);
  }
 }
protected ValueTask WriteAsync(ReadOnlyMemory<byte> payload, int messageId);//呼叫方自己清理

 

 messageId標識一條訊息，寫入訊息頭部， 用於之後處理響應回覆資訊。
   返回值使用ValueTask因為寫入管道通常是同步的，只有管道執行Flush時才可能是非同步的（大多數情況下也是同步的，除非在管道被備份時）。

### 寫入與錯誤

首先需要保證單次寫操作，lock在此不合適，因為它不能與非同步操作很好的協同。考慮flush有可能是非同步的，導致後續（continuation ）部分可能會在另外的執行緒上。這裡使用與非同步相容的SemaphoreSlim。
下面是一條指南：**一般來說, 應用程式程式碼應針對可讀性進行優化;庫程式碼應針對性能進行優化。**
以下為機翻原文
> 您可能同意也可能不同意這一點, 但這是我編寫程式碼的一般指南。我的意思是,類庫程式碼往往有一個單一的重點目的, 往往由一個人的經驗可能是 "深刻的, 但不一定是    廣泛的" 維護;你的大腦專注於那個領域, 用奇怪的長度來優化程式碼是可以的。相反,應用程式程式碼往往涉及更多不同概念的管道-"寬但不一定深" (深度隱藏在各種庫      中)。應用程式程式碼通常具有更復雜和不可預知的互動, 因此重點應放在可維護和 "明顯正確" 上。
  基本上, 我在這裡的觀點是, 我傾向於把很多注意力集中在通常不會放入應用程式程式碼中的優化上, 因為我從經驗和廣泛的基準測試中知道它們真的很重要。所以。。。我要做一些看起來很奇怪的事情, 我希望你和我一起踏上這段旅程。   “明顯正確”的程式碼

private readonly SemaphoreSlim _singleWriter= new SemaphoreSlim(1);
protected async ValueTask WriteAsync(ReadOnlyMemory<byte> payload, int messageId)
{
    await _singleWriter.WaitAsync();
    try
    {
        WriteFrameHeader(writer, payload.Length, messageId);
        await writer.WriteAsync(payload);
    }
    finally
    {
        _singleWriter.Release();
    }
}

 

這段程式碼沒有任何問題，但是即便所有部分都是同步完成的，就會產生多餘的狀態機-------大概是 不是所有地方都需要非同步處理 的意思。
通過兩個問題進行重構
- 單次寫入是否沒有競爭？（無人爭用）
- Flush是否為同步 重構，將原WriteAsync 更名為 WriteAsyncSlowPath，增加新的WriteAsync 作者的“一些看起來很奇怪的” 實現  

protected ValueTask WriteAsync(ReadOnlyMemory<byte> payload, int messageId)
{
    // try to get the conch; if not, switch to async
//writer已經被佔用，非同步
    if (!_singleWriter.Wait(0))
        return WriteAsyncSlowPath(payload, messageId);
    bool release = true;
    try
    {
        WriteFrameHeader(writer, payload.Length, messageId);
        var write = writer.WriteAsync(payload);
        if (write.IsCompletedSuccessfully) return default;
        release = false;
        return AwaitFlushAndRelease(write);
    }
    finally
    {
        if (release) _singleWriter.Release();
    }
}
async ValueTask AwaitFlushAndRelease(ValueTask<FlushResult> flush)
{
    try { await flush; }
    finally { _singleWriter.Release(); }
}

 

三個地方
1. _singleWriter.Wait(0) 意味著writer處於空閒狀態，沒有其他人在呼叫
2. write.IsCompletedSuccessfully 意味著writer同步flush
3. 輔助方法 AwaitFlushAndRelease 處理非同步flush情況
-------------------------------------------------------------------------------------

### 協議頭處理

協議頭由兩個int組成，小端，第一個是長度，第二個是messageId，共8位元組。

void WriteFrameHeader(PipeWriter writer, int length, int messageId)
{
    var span = writer.GetSpan(8);
    BinaryPrimitives.WriteInt32LittleEndian(
        span, length);
    BinaryPrimitives.WriteInt32LittleEndian(
        span.Slice(4), messageId);
    writer.Advance(8);
}

 

### 管道客戶端實現 傳送
 

public class SimplPipelineClient : SimplPipeline
{
    public async Task<IMemoryOwner<byte>> SendReceiveAsync(ReadOnlyMemory<byte> message)
    {
        var tcs = new TaskCompletionSource<IMemoryOwner<byte>>();
        int messageId;
        lock (_awaitingResponses)
        {
            messageId = ++_nextMessageId;
            if (messageId == 0) messageId = 1;
            _awaitingResponses.Add(messageId, tcs);
        }
        await WriteAsync(message, messageId);
        return await tcs.Task;
    }
    public async Task<IMemoryOwner<byte>> SendReceiveAsync(IMemoryOwner<byte> message)
    {
        using (message)
        {
            return await SendReceiveAsync(message.Memory);
        }
    }
}

 

- _awaitingResponses 是個字典，儲存已經發送的訊息，用於將來處理對某條（messageId）訊息的回覆。 ### 接收迴圈  

protected async Task StartReceiveLoopAsync(CancellationToken cancellationToken = default)
{
   try
   {
       while (!cancellationToken.IsCancellationRequested)
       {
           var readResult = await reader.ReadAsync(cancellationToken);
           if (readResult.IsCanceled) break;
           var buffer = readResult.Buffer;
           var makingProgress = false;
           while (TryParseFrame(ref buffer, out var payload, out var messageId))
           {
               makingProgress = true;
               await OnReceiveAsync(payload, messageId);
           }
           reader.AdvanceTo(buffer.Start, buffer.End);
           if (!makingProgress && readResult.IsCompleted) break;
       }
       try { reader.Complete(); } catch { }
   }
   catch (Exception ex)
   {
       try { reader.Complete(ex); } catch { }
   }
}
protected abstract ValueTask OnReceiveAsync(ReadOnlySequence<byte> payload, int messageId);

 

接收緩衝區裡什麼時間會有什麼東西由傳送方和系統環境決定，因此延遲是必然的，所以這裡全部按非同步處理就好。
- TryParseFrame 讀出緩衝區資料，根據幀格式解析出實際資料、id等
- OnRecieveAsync 處理資料,比如對於回覆/響應的處理等
- reader中的資料讀出後儘快Advance一下，通知管道你的讀取進度； 解析幀  

private bool TryParseFrame(
    ref ReadOnlySequence<byte> input,
    out ReadOnlySequence<byte> payload, out int messageId)
{
    if (input.Length < 8)
    {   // not enough data for the header
        payload = default;
        messageId = default;
        return false;
    }
    int length;
    if (input.First.Length >= 8)
    {   // already 8 bytes in the first segment
        length = ParseFrameHeader(
            input.First.Span, out messageId);
    }
    else
    {   // copy 8 bytes into a local span
        Span<byte> local = stackalloc byte[8];
        input.Slice(0, 8).CopyTo(local);
        length = ParseFrameHeader(
            local, out messageId);
    }
    // do we have the "length" bytes?
    if (input.Length < length + 8)
    {
        payload = default;
        return false;
    }
    // success!
    payload = input.Slice(8, length);
    input = input.Slice(payload.End);
    return true;
}

 

緩衝區是不連續的，一段一段的，像連結串列一樣，第一段就是input.First。
程式碼很簡單，主要演示一些用法；
輔助方法

static int ParseFrameHeader(
    ReadOnlySpan<byte> input, out int messageId)
{
    var length = BinaryPrimitives
            .ReadInt32LittleEndian(input);
    messageId = BinaryPrimitives
            .ReadInt32LittleEndian(input.Slice(4));
    return length;
}

 

OnReceiveAsync
 

protected override ValueTask OnReceiveAsync(
    ReadOnlySequence<byte> payload, int messageId)
{
    if (messageId != 0)
    {   // request/response
        TaskCompletionSource<IMemoryOwner<byte>> tcs;
        lock (_awaitingResponses)
        {
            if (_awaitingResponses.TryGetValue(messageId, out tcs))
            {
                _awaitingResponses.Remove(messageId);
            }
        }
        tcs?.TrySetResult(payload.Lease());
    }
    else
    {   // unsolicited
        MessageReceived?.Invoke(payload.Lease());
    }
    return default;
}

 

到此為止，其餘部分主要是一些服務端和其他功能實現及benchmark。。。