本文翻譯自dotnet團隊博客文章：https://blogs.msdn.microsoft.com/dotnet/2018/07/09/system-io-pipelines-high-performance-io-in-net/?

? 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調用不能處理多條消息。

這些是讀取流數據時常見的一些缺陷。為了解決這個問題，我們需要做一些改變：

• 我們需要緩沖傳入的數據，直到找到新的行。

• 我們需要解析緩沖區中返回的所有行

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

因此，我們可以在堆上分配緩沖區去處理更長的一行。我們從客戶端解析較長的一行時，可以通過使用ArrayPool<byte>避免重復分配緩沖區來改進這一點。

這段代碼有效，但現在我們正在重新調整緩沖區大小，從而產生更多緩沖區副本。它將使用更多內存，因為根據代碼在處理一行行后不會縮緩沖區的大小。為避免這種情況，我們可以存儲緩沖區序列，而不是每次超過1KiB大小時調整大小。

此外，我們不會增長1KiB的 緩沖區，直到它完全為空。這意味著我們最終傳遞給ReadAsync越來越小的緩沖區，這將導致對操作系統的更多調用。

為了緩解這種情況，我們將在現有緩沖區中剩余少于512個字節時分配一個新緩沖區：

譯者注：這段代碼太復雜了，懶得翻譯注釋了，大家將就看吧

此代碼只是得到很多更加復雜。當我們正在尋找分隔符時，我們同時跟蹤已填充的緩沖區序列。為此，我們此處使用List<BufferSegment>查找新行分隔符時表示緩沖數據。其結果是，ProcessLine和IndexOf現在接受List<BufferSegment>作為參數，而不是一個byte[]，offset和count。我們的解析邏輯現在需要處理一個或多個緩沖區序列。

我們的服務器現在處理部分消息，它使用池化內存來減少總體內存消耗，但我們還需要進行更多更改：

我們使用的byte[]和ArrayPool<byte>的只是普通的托管數組。這意味著無論何時我們執行ReadAsync或WriteAsync，這些緩沖區都會在異步操作的生命周期內被固定（以便與操作系統上的本機IO API互操作）。這對GC有性能影響，因為無法移動固定內存，這可能導致堆碎片。根據異步操作掛起的時間長短，池的實現可能需要更改。

可以通過解耦讀取邏輯和處理邏輯來優化吞吐量。這會創建一個批處理效果，使解析邏輯可以使用更大的緩沖區塊，而不是僅在解析單個行后才讀取更多數據。這引入了一些額外的復雜性

• 我們需要兩個彼此獨立運行的循環。一個讀取Socket和一個解析緩沖區。

• 當數據可用時，我們需要一種方法來向解析邏輯發出信號。

• 我們需要決定如果循環讀取Socket“太快”會發生什么。如果解析邏輯無法跟上，我們需要一種方法來限制讀取循環（邏輯）。這通常被稱為“流量控制”或“背壓”。

• 我們需要確保事情是線程安全的。我們現在在讀取循環和解析循環之間共享多個緩沖區，并且這些緩沖區在不同的線程上獨立運行。

• 內存管理邏輯現在分布在兩個不同的代碼段中，從填充緩沖區池的代碼是從套接字讀取的，而從緩沖區池取數據的代碼是解析邏輯。

• 我們需要非常小心在解析邏輯完成之后我們如何處理緩沖區序列。如果我們不小心，我們可能會返回一個仍由Socket讀取邏輯寫入的緩沖區序列。

復雜性已經到了極端（我們甚至沒有涵蓋所有案例）。高性能網絡應用通常意味著編寫非常復雜的代碼，以便從系統中獲得更高的性能。

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

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

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

我們的行讀取器的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以下內容的例程：

背壓和流量控制

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

為了解決這個問題，管道有兩個設置來控制數據的流量，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包中。

以下是使用管道處理基于行的消息的.NET Core 2.1服務器應用程序的示例（上面的示例）https://github.com/davidfowl/TcpEcho。它應該運行`dotnet?run`（或通過在Visual Studio中運行）。它偵聽端口8087上的套接字并將收到的消息寫入控制臺。您可以使用netcat或putty等客戶端建立與8087的連接，并發送基于行的消息以使其正常工作。

今天Pipelines為Kestrel和SignalR提供支持，我們希望看見它作為.NET社區中許多網絡庫和組件的核心。

資料：

轉載自System.IO.Pipelines: High performance IO in .NET

Pipelines - a guided tour of the new IO API in .NET, part 1

Pipelines - a guided tour of the new IO API in .NET, part 2

2號資料的中文翻譯?Pipelines - .NET中的新IO API指引(一)

System.IO.Pipelines-Nuget包

PS: 首次翻譯英文文章，不足錯漏請指出，多謝支持

原文地址：https://www.cnblogs.com/xxfy1/p/9290235.html

.NET社區新聞，深度好文，歡迎訪問公眾號文章匯總 http://www.csharpkit.com

總結

以上是生活随笔為你收集整理的System.IO.Pipelines: .NET高性能IO的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。