Pipelines - .NET中的新IO API指引(一)
原文：https://blog.marcgravell.com/2018/07/pipe-dreams-part-1.html
作者：marcgravell
大約兩年前，我發(fā)表了一篇關(guān)于.NET中即將到來的體驗(yàn)性新IO API的博文——在那時(shí)它被叫做"Channels"；在2018年的五月末，它終于在System.IO.Pipelines命名空間中落地，我對(duì)這系列API巨感興趣，而在幾個(gè)星期前，我被分配去用"Pipelines"改造StackExchange.Redis以作為我們2.0更新的一部分
我希望在這個(gè)系列可以討論：
"Pipelines"是什么
如何在代碼方面使用它們
什么時(shí)候你也許會(huì)想要使用它們
為了表達(dá)地更具體，在介紹完"Pipelines"后，我打算大篇幅地講解StackExchange.Redis中的相關(guān)轉(zhuǎn)換，并且作為討論在不同場(chǎng)景下它分別解決了哪些問題的一部分。簡(jiǎn)略地說：在幾乎所有的情況下，答案可以概括為：
它非常適合那些在IO代碼中復(fù)雜卻普遍的痛點(diǎn)；使我們可以替換掉那些丑陋的封裝(kludge)、變通(workaround)或妥協(xié)(compromise)——用一個(gè)在框架中設(shè)計(jì)優(yōu)雅的專門的解決方案。
我敢肯定，我下面所覆蓋的那些痛點(diǎn)，對(duì)于那些工作在"數(shù)據(jù)協(xié)議(data protocol)"層面的人來說，一定非常熟悉。
Pipelines替代/完善了什么？
首先：現(xiàn)有框架中最接近Pipelines的是什么？很簡(jiǎn)單，Stream ,Stream API對(duì)于那些做過序列化或是數(shù)據(jù)協(xié)議工作的人來說非常熟悉，但是，Stream其實(shí)是一個(gè)非常模糊的API——它在不同的場(chǎng)景表現(xiàn)地非常不同：
一些Stream是只讀的，一些是只寫的，一些是讀/寫的
一樣的實(shí)體類型有時(shí)候是只讀的，而有時(shí)是只寫的（比如DeflateStream)
當(dāng)一個(gè)Stream是讀/寫時(shí)，它像是一個(gè)磁帶，讀寫操作全作用于同樣的下層數(shù)據(jù)（FileStream,MemoryStream) ，而有時(shí)它像是兩個(gè)不同的Stream，讀寫作用于本質(zhì)上完全不同的兩個(gè)Stream(NetworkStream,?SslStream)——即duplex stream
在許多deplex(雙工)場(chǎng)景下，很難甚至根本不可能表達(dá)“之后沒有新數(shù)據(jù)會(huì)到來，但是你應(yīng)該繼續(xù)讀取數(shù)據(jù)直到結(jié)束“——只有Close()，而它會(huì)將deplex的兩部分同時(shí)關(guān)閉
有時(shí)Stream會(huì)是可探查的(Seekable)并且支持Position和Length的概念，不過大多數(shù)不會(huì)
由于API隨著時(shí)間的推移，通常會(huì)有多種方法來表達(dá)同一種操作——比如，我們可以用Read(同步)，BeginRead/EndRead(IAsyncResult模式的異步)，或者ReadAsync(async/await模式的異步)；在多數(shù)情況下，調(diào)用代碼無從得知到底哪種方法才是推薦的/最佳的API
如果你使用任何一種異步API，通常很難清楚分辨它的線程模型是什么；它實(shí)質(zhì)上是同步的嗎？如果不是，是哪個(gè)線程會(huì)回調(diào)？它用了同步上下文嗎？線程池？IO complection-port線程？
并且在最近，有了允許使用Span<byte>/Memory<byte>替換byte[]的API——再一次的，調(diào)用者無法知道哪一種才是”更好的“API
這種API本質(zhì)上鼓勵(lì)復(fù)制數(shù)據(jù)；需要緩沖區(qū)？那是將數(shù)據(jù)復(fù)制到了另一塊內(nèi)存中，需要一個(gè)尚未處理的數(shù)據(jù)倉(cāng)庫(kù)？同樣是復(fù)制了數(shù)據(jù)到另一塊內(nèi)存中
所以即使在我們開始討論現(xiàn)實(shí)世界中的Stream例子和使用它們所導(dǎo)致的問題之前，很明顯Stream API本身已經(jīng)有了很多問題，所以首先顯而易見的是，Pipelines解決了這些混亂
什么是Pipelines
說起"Pipelines"，我指的是一組4個(gè)關(guān)鍵API，它們實(shí)現(xiàn)對(duì)一個(gè)二進(jìn)制流解耦、重疊(overlapped)的讀寫訪問，包括緩沖區(qū)管理(池化，回收)，線程感知，豐富的積壓控制，和通過背壓達(dá)到的溢出保護(hù)——所有這些都基于一個(gè)圍繞非連續(xù)內(nèi)存設(shè)計(jì)的 API，That's a?heck?of a word salad——但是不要擔(dān)心，我會(huì)討論每一個(gè)元素來解釋我的意思。
從簡(jiǎn)單的開始：對(duì)一個(gè)單獨(dú)的管道進(jìn)行寫入和讀取
讓我們先準(zhǔn)備一個(gè)對(duì)等的Stream，然后寫入一些簡(jiǎn)單的東西，然后再讀取回來——堅(jiān)持只使用Stream API。我們將只使用ASCII文本以便不用擔(dān)心有任何復(fù)雜編碼的狀況，并且我們的讀寫代碼不對(duì)下層數(shù)據(jù)流做任何假設(shè)。我們只是寫入數(shù)據(jù)，并且讀取到流的末尾從而消費(fèi)它。
我們將先用Stream來做這些——熟悉的領(lǐng)域，然后我們用Pipelines重新實(shí)現(xiàn)它，來看其中的相似和不同之處，在之后，我們將研究在其內(nèi)部究竟發(fā)生了什么，然后我們就能明白為什么它會(huì)吸引我們
也許你會(huì)說"啊，我想起來了TextReader/TextWriter"，我故意不去使用它們——因?yàn)槲以谶@里是在嘗試談?wù)揝tream API，這樣我們的例子可以擴(kuò)展到廣泛的數(shù)據(jù)協(xié)議和場(chǎng)景
using (MemoryStream ms = new MemoryStream()){// write somethingWriteSomeData(ms);// rewind - MemoryStream works like a tapems.Position = 0;// consume itReadSomeData(ms);}現(xiàn)在，要寫入Stream，調(diào)用方需要獲取并填充一個(gè)緩沖區(qū)然后將其傳遞給Stream，此時(shí)我們?yōu)榱撕?jiǎn)化它，使用同步的API，并且簡(jiǎn)單地分配一個(gè)byte數(shù)組
void WriteSomeData(Stream stream){byte[] bytes = Encoding.ASCII.GetBytes("hello, world!");stream.Write(bytes, 0, bytes.Length);stream.Flush();}注意：如果要提高效率地話，在上面的代碼中有很多可以做的，但是這不是重點(diǎn)。所以如果你熟悉這類代碼并且看著膈應(yīng)，別慌，之后我們會(huì)讓它變得更丑陋——呃，我是說更有效率
讀邏輯的代碼會(huì)比寫邏輯更復(fù)雜，因?yàn)樽x代碼無法假定一次單獨(dú)的調(diào)用就可以獲得所有的數(shù)據(jù)，一個(gè)對(duì)Stream的讀操作可能會(huì)什么也不返回(表明已經(jīng)讀到數(shù)據(jù)末尾)，也可能填滿我們的緩沖區(qū)，或者只是返回了一個(gè)字節(jié)即使我們準(zhǔn)備了一個(gè)巨大的緩沖區(qū)。所以Stream的讀代碼大多數(shù)會(huì)是一個(gè)循環(huán)：
void ReadSomeData(Stream stream){int bytesRead;// note that the caller usually can't know much about// the size; .Length is not usually usablebyte[] buffer = new byte[256];do{bytesRead = stream.Read(buffer, 0, buffer.Length);if (bytesRead > 0){ ? // note this only works for single-byte encodingsstring s = Encoding.ASCII.GetString(buffer, 0, bytesRead);Console.Write(s);}} while (bytesRead > 0);}現(xiàn)在我們將它翻譯成pipelines，一個(gè)Pipe可以大略地比作一個(gè)MemoryStream，除了不能多次倒帶(rewind)，數(shù)據(jù)是一個(gè)簡(jiǎn)單的先進(jìn)先出隊(duì)列，我們有一個(gè)writerAPI可以在一端推入數(shù)據(jù)，而一個(gè)readerAPI可以在另一端將數(shù)據(jù)取出，Pipe就是坐在二這之中的一個(gè)緩沖區(qū)。讓我們重現(xiàn)之前的場(chǎng)景，但是用一個(gè)Pipe替換掉MemoryStream（同樣，實(shí)踐中我們通常不會(huì)這么做，但是易于舉例）：
Pipe pipe = new Pipe();// write somethingawait WriteSomeDataAsync(pipe.Writer);// signal that there won't be anything else writtenpipe.Writer.Complete();// consume itawait ReadSomeDataAsync(pipe.Reader);首先我們用默認(rèn)選項(xiàng)創(chuàng)造一個(gè)pipe，然后我們寫入它。注意在Pipe中的IO操作通常都是異步的，所以我們需要await我們的兩個(gè)幫助方法，同樣注意，我們并沒有將這個(gè)Pipe傳入它們——和Stream不同，pipelines 對(duì)于讀和寫有著不同的API層面，所以我們將一個(gè)PipeWriter?傳入幫助方法用來寫入數(shù)據(jù)，然后傳入一個(gè)PipeReader來讀取數(shù)據(jù)，寫入數(shù)據(jù)后，我們?cè)赑ipeWriter上調(diào)用Complete()。我們不需要在MemoryStream中做這個(gè)因?yàn)楫?dāng)它到達(dá)緩沖數(shù)據(jù)的末尾時(shí)會(huì)自動(dòng)EOFs——但是在一些其它的Stream實(shí)現(xiàn)中——尤其是單向流——我們也許需要在寫入數(shù)據(jù)后調(diào)用Close
好了，那么我們的WriteSomeDataAsync?是什么呢？注意，我在下面的代碼中故意多寫了注釋：
async ValueTask WriteSomeDataAsync(PipeWriter writer){// use an oversized size guessMemory<byte> workspace = writer.GetMemory(20);// write the data to the workspaceint bytes = Encoding.ASCII.GetBytes("hello, world!", workspace.Span);// tell the pipe how much of the workspace// we actually want to commitwriter.Advance(bytes);// this is **not** the same as Stream.Flush!await writer.FlushAsync();}首先要注意的是，在處理pipelines時(shí)：不是你控制緩沖區(qū)，而是Pipe，回想我們的Stream代碼，讀和寫代碼都創(chuàng)建了本地byte[]，但是在這里我們沒有，相反，我們通過GetMemory?(或者它的孿生方法GetSpan)向Pipe請(qǐng)求了一個(gè)緩沖區(qū)(workspace)，就先你從名字中想到的那樣，這給了我們一個(gè)Memory<byte>或是一個(gè)Span<byte>?——其容量為最少20字節(jié)
獲取這個(gè)緩沖區(qū)后，將我們的字符串編碼進(jìn)去，這意味著我們是直接寫入Pipe的內(nèi)存，并且記錄下實(shí)際上我們使用了多少字節(jié)，然后我們通過Advance告訴Pipe，我們不受之前請(qǐng)求的20字節(jié)的限制——我們可以寫入0，20，甚至50字節(jié)，最后一個(gè)看起來也許會(huì)令人意外，但是這實(shí)際上是被鼓勵(lì)的！之前的重點(diǎn)是“至少”——實(shí)際上writer可以給我們一個(gè)比我們請(qǐng)求的大很多的緩沖區(qū)。當(dāng)處理較大的數(shù)據(jù)時(shí)，得隴望蜀是很常見的：請(qǐng)求一個(gè)我們能有效利用的最小空間，但是之后在檢查提供給我們的memory/span的體積后，再?zèng)Q定最終實(shí)際寫入多少。
對(duì)Advance的調(diào)用很重要，它意味著一次寫操作的終結(jié)，使得Pipe中的數(shù)據(jù)可用從而被reader消費(fèi)。對(duì)FlushAsync?的調(diào)用同樣重要，但是有微妙的區(qū)別，但是在我們可以充分地闡明這區(qū)別是什么前，我們需要先看一看reader。這是我們的ReadSomeDataAsync?方法：
async ValueTask ReadSomeDataAsync(PipeReader reader){while (true){// await some data being availableReadResult read = await reader.ReadAsync();ReadOnlySequence<byte> buffer = read.Buffer;// check whether we've reached the end// and processed everythingif (buffer.IsEmpty && read.IsCompleted)break; // exit loop// process what we receivedforeach (Memory<byte> segment in buffer){string s = Encoding.ASCII.GetString(segment.Span);Console.Write(s);}// tell the pipe that we used everythingreader.AdvanceTo(buffer.End);}}就像Stream例子一樣，我們有一個(gè)循環(huán)持續(xù)到我們讀取到數(shù)據(jù)的末尾，在Stream中，這種情況通過Read方法返回一個(gè)非正結(jié)果時(shí)判定，但是在pipeline中有兩種檢查方式：
read.IsCompleted告訴我們那個(gè)寫pipe是否被通知完成，并且不會(huì)再有數(shù)據(jù)被寫入(pipe.Writer.Complete();之前代碼中的這句)
buffer.IsEmpty告訴我們?cè)谶@次操作中沒有剩余的數(shù)據(jù)需要處理
如果pipe中不再有數(shù)據(jù)并且writer被通知complete，那么將永遠(yuǎn)不會(huì)有東西存在于這個(gè)pipe中，那我們就可以退出了
如果我們有數(shù)據(jù)存在，我們可以查看緩沖區(qū)，所以首先——我們要談?wù)劸彌_；在代碼中那是個(gè)新類型ReadOnlySequence<byte>——這個(gè)概念結(jié)合了幾個(gè)角色：
描述不連續(xù)內(nèi)存，特別是一個(gè)由0個(gè)，1個(gè)或多個(gè)ReadOnlyMemory<byte>塊組成的序列
描述在這個(gè)數(shù)據(jù)流中的一個(gè)邏輯位置(SequencePosition)—— in particular via?buffer.Start?and?buffer.End
非連續(xù)在此非常重要，我們很快將看到這些數(shù)據(jù)實(shí)際上的去向，但在讀方面：我們需要準(zhǔn)備好處理可以跨多個(gè)部分傳播的數(shù)據(jù)。在這里，我們通過簡(jiǎn)單的遍歷緩沖區(qū)，輪流解碼每一段數(shù)據(jù)來達(dá)到目的。請(qǐng)注意, 即使 API 被設(shè)計(jì)為可以描述多個(gè)非連續(xù)緩沖區(qū), 但通常情況下, 接收到的數(shù)據(jù)在單個(gè)緩沖區(qū)中是連續(xù)的。在這種情況下, 通常可以為單個(gè)緩沖區(qū)編寫優(yōu)化的實(shí)現(xiàn)。你可以通過檢查buffer.IsSingleSegment和訪問buffer.First來做到。
最終，我們調(diào)用AdvanceTo，告訴Pipe我們實(shí)際上使用了多少數(shù)據(jù)。
關(guān)鍵點(diǎn)：你無需取出你提供的所有數(shù)據(jù)
對(duì)比流：當(dāng)你在Stream上調(diào)用Read時(shí)，它會(huì)將所有數(shù)據(jù)放到你給它的緩沖區(qū)中，在大多數(shù)現(xiàn)實(shí)場(chǎng)景中，并不是總是能及時(shí)消費(fèi)掉所有的數(shù)據(jù)——maybe it only makes sense to consider "commands" as "entire text lines",, and you haven't yet seen a?cr/lf?in the data. 對(duì)于Stream來說，這點(diǎn)很坑——一旦數(shù)據(jù)給了你，就是你的問題了，如果你現(xiàn)在用不上它，那你就要在某處儲(chǔ)備這段數(shù)據(jù)，但是對(duì)于Pipelines，你可以告訴它你消費(fèi)過了。在我們的例子中，我們通過傳遞buffer.End到AdvanceTo來告訴它我們消費(fèi)掉了之前提供的所有數(shù)據(jù)。這意味著我們將永遠(yuǎn)不會(huì)再見到這段數(shù)據(jù)，就像用Stream一樣，但是，我們也可以傳遞buffer.Start，意味著“我們什么都還沒使用”——及時(shí)我們能夠檢查這段數(shù)據(jù)，它也依然會(huì)留存在pipe中以供后續(xù)讀取。我們也可以獲取緩沖區(qū)中任意的SequencePosition?值——例如如果我們讀取20字節(jié)——所以我們可以完全控制有多少數(shù)據(jù)被從pipe中丟棄。這里有兩種方法取得SequencePosition?：
你可以就像Slice(...)一個(gè)?Span<T>?o或者M(jìn)emory<T>一樣Slice(...)一個(gè)ReadOnlySequence<byte>?——然后訪問子集中的.Start或.End
你可以使用ReadOnlySequence<byte>中的.GetPosition(...)?方法，它返回一個(gè)相關(guān)位置而無需真正分割
更微妙的是：我們可以分別告訴它我們消費(fèi)了一些數(shù)量，但是我們已檢查了另一個(gè)不同的數(shù)量，這里最常見的例子是表達(dá)“你可以丟棄這么多——這些我做完了；但是我看完了所有的數(shù)據(jù)，我此時(shí)無法處理——我需要更多數(shù)據(jù)（you can drop?this much?- I'm done with that; but I looked at everything, I can't make any more progress at the moment - I need more data）”，具體來說：
reader.AdvanceTo(consumedToPosition, buffer.End);這里正是PipeWriter.FlushAsync()和PipeReader.ReadAsync()微妙的相互作用出場(chǎng)的地方了，我之前跳過了PipeWriter.FlushAsync()，它實(shí)際上在一次調(diào)用里提供了兩個(gè)功能：
如果存在一個(gè)ReadAsync?調(diào)用，它會(huì)被注意到，因?yàn)樗枰獢?shù)據(jù)，然后它喚醒reader，使讀取循環(huán)繼續(xù)
如果writer快過reader，比如pipe中充滿了沒有被reader清楚的數(shù)據(jù)，它會(huì)掛起writer(通過同步的not completing)——當(dāng)pipe有了更多空間后，才會(huì)被重新激活(writer掛起/恢復(fù)的閾值可以在創(chuàng)建Pipe實(shí)例時(shí)被指定)
顯然, 這些概念在我們的示例中沒有發(fā)揮作用, 但它們是Pipelines工作原理的核心思想。將數(shù)據(jù)推送回pipe的能力極大地簡(jiǎn)化了大量 IO 場(chǎng)景。實(shí)際上, 我在有pipelines之前看到的每一個(gè)協(xié)議處理代碼都有大量的代碼與處理不完整數(shù)據(jù)的積壓有關(guān)——它是這樣一個(gè)重復(fù)的邏輯, 我非常高興地看到它能在框架中被處理得很好。
“喚醒”或者說“響應(yīng)式”指的是什么
你可能會(huì)注意到，我并沒有真正定義我之前表達(dá)的意思，在表層上，我的意思是：對(duì)于ReadAsync?或FlushAsync?的一個(gè)await操作在其返回之前是未完成的，然后現(xiàn)在異步延續(xù)被產(chǎn)生，允許我們的async方法恢復(fù)執(zhí)行，是，沒錯(cuò)，不過這只是重新說明了?async/await?是什么意思。但是我debug的重點(diǎn)關(guān)注在于代碼運(yùn)行于哪個(gè)線程上——原因我會(huì)在之后的系列中討論。所以說 "異步延續(xù)被產(chǎn)生 " 對(duì)我來說還不夠。我想了解是誰在調(diào)用它, 就線程而言。最常見的答案是：
它通過SynchronizationContext?委托（注意：在許多系統(tǒng)中沒有SynchronizationContext?）
觸發(fā)狀態(tài)更改的線程會(huì)在狀態(tài)更改時(shí)使用, 以產(chǎn)生延續(xù)
全局線程池會(huì)被用來產(chǎn)生延續(xù)
在某些情況下，所有這些都可以是沒問題的，而在某些情況下，所有這些都可能是糟糕的！同步上下文是一種完善的機(jī)制，可以從工作線程返回到主應(yīng)用程序線程 (例外：桌面應(yīng)用程序中的 UI 線程)。然而，它是沒有必要的如果只是說我們完成了一個(gè)IO操作然后準(zhǔn)備跳回一個(gè)應(yīng)用線程；并且這么做會(huì)實(shí)際上將大量IO代碼和數(shù)據(jù)處理代碼轉(zhuǎn)移到應(yīng)用線程——這通常是我們想要避免的。并且，如果應(yīng)用代碼在異步調(diào)用時(shí)使用了Wait()或.Result會(huì)導(dǎo)致死鎖（假設(shè)你不是故意的）。第二種選項(xiàng)（“內(nèi)聯(lián)”地在一個(gè)觸發(fā)它的線程上執(zhí)行回調(diào)）可能會(huì)有問題，因?yàn)樗梢酝等∧阆胍脕碜鰟e的事的線程（并且有可能導(dǎo)致死鎖）；并且在某些極端情況下，當(dāng)兩個(gè)異步方法本質(zhì)上作為協(xié)程運(yùn)行時(shí)，可能會(huì)導(dǎo)致stack-dive（最終棧溢出）。最后一個(gè)選項(xiàng) (全局線程池) 沒有前兩個(gè)的問題, 但在某些負(fù)載條件下可能會(huì)遇到嚴(yán)重問題——我將在本系列后面的部分討論這一點(diǎn)。
但是好消息是，pipelines在這里給了你控制權(quán)。當(dāng)創(chuàng)建Pipe實(shí)例時(shí)，我們可以提供PipeScheduler?實(shí)例給reader和writer（分別地）使用。PipeScheduler?用來執(zhí)行這些激活。如果沒有制定，那么它默認(rèn)受i按檢查SynchronizationContext，然后使用全局線程池使用“內(nèi)聯(lián)”延續(xù)（使用那個(gè)導(dǎo)致狀態(tài)改變的線程）作為另一個(gè)可用選項(xiàng)。但是：你可以提供你對(duì)于PipeScheduler自己的實(shí)現(xiàn)，給予你對(duì)線程模型的完全控制。
總結(jié)
所以：我們已經(jīng)研究了什么是Pipe?，和我們?cè)鯓硬拍苡肞ipeWriter寫入一個(gè)pipe，和用PipeReader?從pipe中讀取——和怎樣"advance"二者。我們已經(jīng)研究了其于Stream的相似和差異，我們討論了ReadAsync()和?FlushAsync()?怎樣交互控制writer和reader的分片執(zhí)行。我們研究了通過pipe提供所有緩沖區(qū)后，對(duì)緩沖區(qū)的責(zé)任怎樣被反轉(zhuǎn)——和pipe怎樣簡(jiǎn)化了積壓數(shù)據(jù)的管理。最終，我們討論了激活對(duì)await操作的延續(xù)進(jìn)行激活的線程模型。
這對(duì)于第一步來說可能已經(jīng)足夠了。在之后，我們將研究pipelines工作時(shí)的內(nèi)存模型——比如數(shù)據(jù)存活在哪里。我們也將研究如何在現(xiàn)實(shí)場(chǎng)景中利用pipelines來開始做些有趣的東西。
相關(guān)文章：
System.IO.Pipelines: .NET高性能IO
原文地址:?https://zhuanlan.zhihu.com/p/39223648
.NET社區(qū)新聞，深度好文，歡迎訪問公眾號(hào)文章匯總 http://www.csharpkit.com
總結(jié)
以上是生活随笔為你收集整理的Pipelines - .NET中的新IO API指引(一)的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Microsoft宣布正式发布Linux
- 下一篇： Pipelines - .NET中的新I