System.IO.Pipelines——高性能IO(二)

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，，，  

System.IO.Pipelines——高性能IO(一) 

System.IO.Pipelines——高性能IO(二)   

System.IO.Pipelines——高性能IO(三)  

轉載請註明出處，本文鏈接：https://www.uj5u.com/net/100042.html

標籤：C#

上一篇：c# 例外精準定位

下一篇：委托的自己理解