Pipe——高性能IO

System.IO.Pipelines是一個新的庫，旨在簡化在.NET中執行高性能IO的過程。它是一個依賴.NET Standard的庫，適用於全部.NET實現。

Pipelines誕生於.NET Core團隊，爲使Kestrel成爲業界最快的Web服務器之一。最初從做爲Kestrel內部的實現細節發展成爲可重用的API，它在.Net Core 2.1中做爲可用於全部.NET開發人員的最高級BCL API（System.IO.Pipelines）提供。

它解決了什麼問題？

爲了正確解析Stream或Socket中的數據，代碼有固定的樣板，而且有許多極端狀況，爲了處理他們，不得不編寫難以維護的複雜代碼。
實現高性能和正確性，同時也難以處理這種複雜性。Pipelines旨在解決這種複雜性。

有多複雜？

讓咱們從一個簡單的問題開始吧。咱們想編寫一個TCP服務器，它接收來自客戶端的用行分隔的消息（由\n分隔）。（譯者注：即一行爲一條消息）

使用NetworkStream的TCP服務器

在Pipelines以前用.NET編寫的典型代碼以下所示：

async Task ProcessLinesAsync(NetworkStream stream) { var buffer = new byte[1024]; await stream.ReadAsync(buffer, 0, buffer.Length); // 在buffer中處理一行消息 ProcessLine(buffer); }

此代碼可能在本地測試時正確工做，但它有幾個潛在錯誤：

• 一次ReadAsync調用可能沒有收到整個消息（行尾）。
• 它忽略了stream.ReadAsync()返回值中實際填充到buffer中的數據量。（譯者注：即不必定將buffer填充滿）
• 一次ReadAsync調用不能處理多條消息。

這些是讀取流數據時常見的一些缺陷。爲了解決這個問題，咱們須要作一些改變：

• 咱們須要緩衝傳入的數據，直到找到新的行。
• 咱們須要解析緩衝區中返回的全部行

async Task ProcessLinesAsync(NetworkStream stream) { var buffer = new byte[1024]; var bytesBuffered = 0; var bytesConsumed = 0; while (true) { var bytesRead = await stream.ReadAsync(buffer, bytesBuffered, buffer.Length - bytesBuffered); if (bytesRead == 0) { // EOF 已經到末尾 break; } // 跟蹤已緩衝的字節數 bytesBuffered += bytesRead; var linePosition = -1; do { // 在緩衝數據中查找找一個行末尾 linePosition = Array.IndexOf(buffer, (byte)‘\n‘, bytesConsumed, bytesBuffered - bytesConsumed); if (linePosition >= 0) { // 根據偏移量計算一行的長度 var lineLength = linePosition - bytesConsumed; // 處理這一行 ProcessLine(buffer, bytesConsumed, lineLength); // 移動bytesConsumed爲了跳過咱們已經處理掉的行 (包括\n) bytesConsumed += lineLength + 1; } } while (linePosition >= 0); } }

這一次，這可能適用於本地開發，但一行可能大於1KiB（1024字節）。咱們須要調整輸入緩衝區的大小，直到找到新行。

所以，咱們能夠在堆上分配緩衝區去處理更長的一行。咱們從客戶端解析較長的一行時，能夠經過使用ArrayPool<byte>避免重複分配緩衝區來改進這一點。

async Task ProcessLinesAsync(NetworkStream stream) { byte[] buffer = ArrayPool<byte>.Shared.Rent(1024); var bytesBuffered = 0; var bytesConsumed = 0; while (true) { // 在buffer中計算中剩餘的字節數 var bytesRemaining = buffer.Length - bytesBuffered; if (bytesRemaining == 0) { // 將buffer size翻倍 而且將以前緩衝的數據複製到新的緩衝區 var newBuffer = ArrayPool<byte>.Shared.Rent(buffer.Length * 2); Buffer.BlockCopy(buffer, 0, newBuffer, 0, buffer.Length); // 將舊的buffer丟回池中 ArrayPool<byte>.Shared.Return(buffer); buffer = newBuffer; bytesRemaining = buffer.Length - bytesBuffered; } var bytesRead = await stream.ReadAsync(buffer, bytesBuffered, bytesRemaining); if (bytesRead == 0) { // EOF 末尾 break; } // 跟蹤已緩衝的字節數 bytesBuffered += bytesRead; do { // 在緩衝數據中查找找一個行末尾 linePosition = Array.IndexOf(buffer, (byte)‘\n‘, bytesConsumed, bytesBuffered - bytesConsumed); if (linePosition >= 0) { // 根據偏移量計算一行的長度 var lineLength = linePosition - bytesConsumed; // 處理這一行 ProcessLine(buffer, bytesConsumed, lineLength); // 移動bytesConsumed爲了跳過咱們已經處理掉的行 (包括\n) bytesConsumed += lineLength + 1; } } while (linePosition >= 0); } }

這段代碼有效，但如今咱們正在從新調整緩衝區大小，從而產生更多緩衝區副本。它將使用更多內存，由於根據代碼在處理一行行後不會縮緩衝區的大小。爲避免這種狀況，咱們能夠存儲緩衝區序列，而不是每次超過1KiB大小時調整大小。

此外，咱們不會增加1KiB的 緩衝區，直到它徹底爲空。這意味着咱們最終傳遞給ReadAsync愈來愈小的緩衝區，這將致使對操做系統的更多調用。

爲了緩解這種狀況，咱們將在現有緩衝區中剩餘少於512個字節時分配一個新緩衝區：

public class BufferSegment { public byte[] Buffer { get; set; } public int Count { get; set; } public int Remaining => Buffer.Length - Count; } async Task ProcessLinesAsync(NetworkStream stream) { const int minimumBufferSize = 512; var segments = new List<BufferSegment>(); var bytesConsumed = 0; var bytesConsumedBufferIndex = 0; var segment = new BufferSegment { Buffer = ArrayPool<byte>.Shared.Rent(1024) }; segments.Add(segment); while (true) { // Calculate the amount of bytes remaining in the buffer if (segment.Remaining < minimumBufferSize) { // Allocate a new segment segment = new BufferSegment { Buffer = ArrayPool<byte>.Shared.Rent(1024) }; segments.Add(segment); } var bytesRead = await stream.ReadAsync(segment.Buffer, segment.Count, segment.Remaining); if (bytesRead == 0) { break; } // Keep track of the amount of buffered bytes segment.Count += bytesRead; while (true) { // Look for a EOL in the list of segments var (segmentIndex, segmentOffset) = IndexOf(segments, (byte)‘\n‘, bytesConsumedBufferIndex, bytesConsumed); if (segmentIndex >= 0) { // Process the line ProcessLine(segments, segmentIndex, segmentOffset); bytesConsumedBufferIndex = segmentOffset; bytesConsumed = segmentOffset + 1; } else { break; } } // Drop fully consumed segments from the list so we don‘t look at them again for (var i = bytesConsumedBufferIndex; i >= 0; --i) { var consumedSegment = segments[i]; // Return all segments unless this is the current segment if (consumedSegment != segment) { ArrayPool<byte>.Shared.Return(consumedSegment.Buffer); segments.RemoveAt(i); } } } } (int segmentIndex, int segmentOffest) IndexOf(List<BufferSegment> segments, byte value, int startBufferIndex, int startSegmentOffset) { var first = true; for (var i = startBufferIndex; i < segments.Count; ++i) { var segment = segments[i]; // Start from the correct offset var offset = first ? startSegmentOffset : 0; var index = Array.IndexOf(segment.Buffer, value, offset, segment.Count - offset); if (index >= 0) { // Return the buffer index and the index within that segment where EOL was found return (i, index); } first = false; } return (-1, -1); }

此代碼只是獲得不少更加複雜。當咱們正在尋找分隔符時，咱們同時跟蹤已填充的緩衝區序列。爲此，咱們此處使用List<BufferSegment>查找新行分隔符時表示緩衝數據。其結果是，ProcessLine和IndexOf如今接受List<BufferSegment>做爲參數，而不是一個byte[]，offset和count。咱們的解析邏輯如今須要處理一個或多個緩衝區序列。

咱們的服務器如今處理部分消息，它使用池化內存來減小整體內存消耗，但咱們還須要進行更多更改：

1. 咱們使用的byte[]和ArrayPool<byte>的只是普通的託管數組。這意味着不管什麼時候咱們執行ReadAsync或WriteAsync，這些緩衝區都會在異步操做的生命週期內被固定（以便與操做系統上的本機IO API互操做）。這對GC有性能影響，由於沒法移動固定內存，這可能致使堆碎片。根據異步操做掛起的時間長短，池的實現可能須要更改。
2. 能夠經過解耦讀取邏輯和處理邏輯來優化吞吐量。這會建立一個批處理效果，使解析邏輯可使用更大的緩衝區塊，而不是僅在解析單個行後纔讀取更多數據。這引入了一些額外的複雜性
   • 咱們須要兩個彼此獨立運行的循環。一個讀取Socket和一個解析緩衝區。
   • 當數據可用時，咱們須要一種方法來向解析邏輯發出信號。
   • 咱們須要決定若是循環讀取Socket「太快」會發生什麼。若是解析邏輯沒法跟上，咱們須要一種方法來限制讀取循環（邏輯）。這一般被稱爲「流量控制」或「背壓」。
   • 咱們須要確保事情是線程安全的。咱們如今在讀取循環和解析循環之間共享多個緩衝區，而且這些緩衝區在不一樣的線程上獨立運行。
   • 內存管理邏輯如今分佈在兩個不一樣的代碼段中，從填充緩衝區池的代碼是從套接字讀取的，而從緩衝區池取數據的代碼是解析邏輯。
   • 咱們須要很是當心在解析邏輯完成以後咱們如何處理緩衝區序列。若是咱們不當心，咱們可能會返回一個仍由Socket讀取邏輯寫入的緩衝區序列。

複雜性已經到了極端（咱們甚至沒有涵蓋全部案例）。高性能網絡應用一般意味着編寫很是複雜的代碼，以便從系統中得到更高的性能。

System.IO.Pipelines的目標是使這種類型的代碼更容易編寫。

使用System.IO.Pipelines的TCP服務器

讓咱們來看看這個例子的樣子System.IO.Pipelines：

async Task ProcessLinesAsync(Socket socket) { var pipe = new Pipe(); Task writing = FillPipeAsync(socket, pipe.Writer); Task reading = ReadPipeAsync(pipe.Reader); return Task.WhenAll(reading, writing); } async Task FillPipeAsync(Socket socket, PipeWriter writer) { const int minimumBufferSize = 512; while (true) { // 從PipeWriter至少分配512字節 Memory<byte> memory = writer.GetMemory(minimumBufferSize); try { int bytesRead = await socket.ReceiveAsync(memory, SocketFlags.None); if (bytesRead == 0) { break; } // 告訴PipeWriter從套接字讀取了多少 writer.Advance(bytesRead); } catch (Exception ex) { LogError(ex); break; } // 標記數據可用，讓PipeReader讀取 FlushResult result = await writer.FlushAsync(); if (result.IsCompleted) { break; } } // 告訴PipeReader沒有更多的數據 writer.Complete(); } async Task ReadPipeAsync(PipeReader reader) { while (true) { ReadResult result = await reader.ReadAsync(); ReadOnlySequence<byte> buffer = result.Buffer; SequencePosition? position = null; do { // 在緩衝數據中查找找一個行末尾 position = buffer.PositionOf((byte)‘\n‘); if (position != null) { // 處理這一行 ProcessLine(buffer.Slice(0, position.Value)); // 跳過 這一行+\n (basically position 主要位置？) buffer = buffer.Slice(buffer.GetPosition(1, position.Value)); } } while (position != null); // 告訴PipeReader咱們以及處理多少緩衝 reader.AdvanceTo(buffer.Start, buffer.End); // 若是沒有更多的數據，中止都去 if (result.IsCompleted) { break; } } // 將PipeReader標記爲完成 reader.Complete(); }

咱們的行讀取器的pipelines版本有2個循環：

• FillPipeAsync從Socket讀取並寫入PipeWriter。
• ReadPipeAsync從PipeReader中讀取並解析傳入的行。

與原始示例不一樣，在任何地方都沒有分配顯式緩衝區。這是管道的核心功能之一。全部緩衝區管理都委託給PipeReader/PipeWriter實現。

這使得使用代碼更容易專一於業務邏輯而不是複雜的緩衝區管理。

在第一個循環中，咱們首先調用PipeWriter.GetMemory(int)從底層編寫器獲取一些內存; 而後咱們調用PipeWriter.Advance(int)告訴PipeWriter咱們實際寫入緩衝區的數據量。而後咱們調用PipeWriter.FlushAsync()來提供數據給PipeReader。

在第二個循環中，咱們正在使用PipeWriter最終來自的緩衝區Socket。當調用PipeReader.ReadAsync()返回時，咱們獲得一個ReadResult包含2條重要信息，包括以ReadOnlySequence<byte>形式讀取的數據和bool IsCompleted，讓reader知道writer是否寫完（EOF）。在找到行尾（EOL）分隔符並解析該行以後，咱們將緩衝區切片以跳過咱們已經處理過的內容，而後咱們調用PipeReader.AdvanceTo告訴PipeReader咱們消耗了多少數據。

在每一個循環結束時，咱們完成了reader和writer。這容許底層Pipe釋放它分配的全部內存。

System.IO.Pipelines

除了處理內存管理以外，其餘核心管道功能還包括可以在Pipe不實際消耗數據的狀況下查看數據。

PipeReader有兩個核心API ReadAsync和AdvanceTo。ReadAsync獲取Pipe數據，AdvanceTo告訴PipeReader再也不須要這些緩衝區，以即可以丟棄它們（例如返回到底層緩衝池）。

---

這是一個http解析器的示例，它在接收Pipe到有效起始行以前讀取部分數據緩衝區數據。

ReadOnlySequence<T>

該Pipe實現存儲了在PipeWriter和PipeReader之間傳遞的緩衝區的連接列表。PipeReader.ReadAsync暴露一個ReadOnlySequence<T>新的BCL類型，它表示一個或多個ReadOnlyMemory<T>段的視圖，相似於Span<T>和Memory<T>提供數組和字符串的視圖。

該Pipe內部維護指向reader和writer能夠分配或更新它們的數據集合，。SequencePosition表示緩衝區鏈表中的單個點，可用於有效地對ReadOnlySequence<T>進行切片。

這段實在翻譯困難，給出原文
The Pipe internally maintains pointers to where the reader and writer are in the overall set of allocated data and updates them as data is written or read. The SequencePosition represents a single point in the linked list of buffers and can be used to efficiently slice the ReadOnlySequence

因爲ReadOnlySequence<T>能夠支持一個或多個段，所以高性能處理邏輯一般基於單個或多個段來分割快速和慢速路徑（fast and slow paths?）。

例如，這是一個將ASCII ReadOnlySequence<byte>轉換爲string如下內容的例程：

string GetAsciiString(ReadOnlySequence<byte> buffer) { if (buffer.IsSingleSegment) { return Encoding.ASCII.GetString(buffer.First.Span); } return string.Create((int)buffer.Length, buffer, (span, sequence) => { foreach (var segment in sequence) { Encoding.ASCII.GetChars(segment.Span, span); span = span.Slice(segment.Length); } }); }

背壓和流量控制

在一個完美的世界中，讀取和解析工做是一個團隊：讀取線程消耗來自網絡的數據並將其放入緩衝區，而解析線程負責構建適當的數據結構。一般，解析將比僅從網絡複製數據塊花費更多時間。結果，讀取線程能夠輕易地壓倒解析線程。結果是讀取線程必須減慢或分配更多內存來存儲解析線程的數據。爲得到最佳性能，在頻繁暫停和分配更多內存之間存在平衡。

爲了解決這個問題，管道有兩個設置來控制數據的流量，PauseWriterThreshold和ResumeWriterThreshold。PauseWriterThreshold決定有多少數據應該在調用PipeWriter.FlushAsync以前進行緩衝停頓。ResumeWriterThreshold控制reader消耗多少後寫入能夠恢復。

當Pipe的數據量超過PauseWriterThreshold，PipeWriter.FlushAsync會異步阻塞。數據量變得低於ResumeWriterThreshold，它會解鎖時。兩個值用於防止在極限附近發生反覆阻塞和解鎖。

IO調度

一般在使用async / await時，會在線程池線程或當前線程上調用continuation SynchronizationContext。

在執行IO時，對執行IO的位置進行細粒度控制很是重要，這樣能夠更有效地利用CPU緩存，這對於Web服務器等高性能應用程序相當重要。Pipelines公開了一個PipeScheduler肯定異步回調運行位置的方法。這使得調用者能夠精確控制用於IO的線程。

實踐中的一個示例是在Kestrel Libuv傳輸中，其中IO回調在專用事件循環線程上運行。

PipeReader模式的其餘好處：

• 一些底層系統支持「無緩衝等待」，即，在底層系統中實際可用數據以前，永遠不須要分配緩衝區。例如，在帶有epoll的Linux上，能夠等到數據準備好以後再實際提供緩衝區來進行讀取。這避免了具備大量線程等待數據的問題不會當即須要保留大量內存。
• 默認狀況下Pipe，能夠輕鬆地針對網絡代碼編寫單元測試，由於解析邏輯與網絡代碼分離，所以單元測試僅針對內存緩衝區運行解析邏輯，而不是直接從網絡中消耗。它還能夠輕鬆測試那些難以測試發送部分數據的模式。ASP.NET Core使用它來測試Kestrel的http解析器的各個方面。
• 容許將底層OS緩衝區（如Windows上的Registered IO API）暴露給用戶代碼的系統很是適合管道，由於緩衝區始終由PipeReader實現提供。

其餘相關類型

做爲製做System.IO.Pipelines的一部分，咱們還添加了許多新的原始BCL類型：

• MemoryPool<T>，IMemoryOwner<T>，MemoryManager<T> - .NET Core 1.0添加了ArrayPool<T>，在.NET Core 2.1中，咱們如今有一個更通用的抽象，適用於任何工做的池Memory<T>。這提供了一個可擴展點，容許您插入更高級的分配策略以及控制緩衝區的管理方式（例如，提供預先固定的緩衝區而不是純託管的陣列）。
• IBufferWriter<T> - 表示用於寫入同步緩衝數據的接收器。（PipeWriter實現這個）
• IValueTaskSource - ValueTask<T>自.NET Core 1.1以來就已存在，但在.NET Core 2.1中得到了一些超級權限，容許無分配的等待異步操做。有關詳細信息，請參閱https://github.com/dotnet/corefx/issues/27445。

如何使用管道？

API存在於System.IO.Pipelines nuget包中。

主要包含一個Pipe對象，它有一個Writer屬性和Reader屬性。

var pipe = new Pipe();
var writer = pipe.Writer;
var reader = pipe.Reader; 

Writer對象

Writer對象用於從數據源讀取數據，將數據寫入管道中；它對應業務中的"讀"操做。

var content = Encoding.Default.GetBytes("hello world");
var data = new Memory<byte>(content);
var result = await writer.WriteAsync(data);

另外，它也有一種使用Pipe申請Memory的方式

var buffer = writer.GetMemory(512);
content.CopyTo(buffer);
writer.Advance(content.Length);
var result = await writer.FlushAsync(); 

Reader對象

Reader對象用於從管道中獲取數據源，它對應業務中的"用"操做。

首先獲取管道的緩衝區：

var result = await reader.ReadAsync();
var buffer = result.Buffer;

這個Buffer是一個ReadOnlySequence<byte>對象，它是一個至關好的動態內存對象，而且至關高效。它自己由多段Memory<byte>組成，查看Memory段的方法有：

IsSingleSegment： 判斷是否只有一段Memory<byte>
First： 獲取第一段Memory<byte>
GetEnumerator: 獲取分段的Memory<byte>
它從邏輯上也能夠當作一段連續的Memory<byte>，也有相似的方法：

Length： 整個數據緩衝區長度
Slice： 分割緩衝區
CopyTo:　將內容複製到Span中
ToArray： 將內容複製到byte[]中
另外，它還有一個相似遊標的位置對象SequencePosition，能夠從其Position相關函數中使用，這裏就很少介紹了。

這個緩衝區解決了"數據讀不夠"的問題，一次讀取的不夠下次能夠接着讀，不用緩衝區的動態分配，高效的內存管理方式帶來了良好的性能，好用的接口是咱們能更關注業務。

獲取到緩衝區後，就是使用緩衝區的數據

var data = buffer.ToArray();

使用完後，告訴PIPE當前使用了多少數據，下次接着從結束位置後讀起

reader.AdvanceTo(buffer.GetPosition(4));

這是一個至關實用的設計，它解決了"讀了就得用"的問題，不只能夠將不用的數據下次再使用，還能夠實現Peek的操做，只讀但不改變遊標。 

交互

除了"讀"和"用"操做外，它們之間還須要一些交互，例如：

讀過程當中數據源不可用，須要中止使用
使用過程當中業務結束，須要停止數據源。
Reader和Writer都有一個Complete函數，用於通知結束：

reader.Complete();
writer.Complete();

在Writer寫入和Reader讀取時，會得到一個結果

FlushResult result = await writer.FlushAsync();
ReadResult result = await reader.ReadAsync();

它們都有一個IsComplete屬性，能夠根據它是否爲true判斷是否已經結束了讀和寫的操做。 

取消

在寫入和讀取的時候，也能夠傳入一個CancellationToken，用於取消相應的操做。

writer.FlushAsync(CancellationToken.None);
reader.ReadAsync(CancellationToken.None);

若是取消成功，對應的Result的IsCanceled則爲true

 

轉載請標明本文來源：http://www.javashuo.com/article/p-kbcsjzhs-gw.html
更多內容歡迎Star、Fork個人的github：https://github.com/yswenli/若是發現本文有什麼問題和任何建議，也隨時歡迎交流~