Pipe——高性能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。。。