System.IO.Pipelines: .NET में उच्च-प्रदर्शन IO

System.IO.Pipelines एक नई लाइब्रेरी है जो .NET में कोड के संगठन को सरल बनाती है। यदि आपको जटिल कोड से निपटना है तो उच्च प्रदर्शन और सटीकता सुनिश्चित करना मुश्किल है। System.IO.Pipelines का कार्य कोड को सरल बनाना है। कटौती के तहत अधिक जानकारी!



Kestrel को उद्योग में सबसे तेज़ वेब सर्वरों में से एक बनाने के लिए .NET कोर डेवलपमेंट टीम के प्रयासों के परिणामस्वरूप लाइब्रेरी का निर्माण हुआ। यह मूल रूप से Kestrel कार्यान्वयन के हिस्से के रूप में कल्पना की गई थी, लेकिन एक पुन: प्रयोज्य एपीआई में विकसित हुआ है, जो संस्करण 2.1 में प्रथम श्रेणी के BCL API (System.IO.Pipelines) के रूप में उपलब्ध है।

वह किन समस्याओं का समाधान करती है?

किसी स्ट्रीम या सॉकेट से डेटा का ठीक से विश्लेषण करने के लिए, आपको बड़ी मात्रा में मानक कोड लिखना होगा। इसी समय, कई नुकसान हैं जो कोड को स्वयं और इसके समर्थन को जटिल करते हैं।

आज कौन सी कठिनाइयाँ आती हैं?

चलो एक सरल कार्य के साथ शुरू करते हैं। हमें एक टीसीपी सर्वर लिखना होगा जो क्लाइंट से लाइन-सीमांकित संदेश (\ n) प्राप्त करता है।

NetworkStream के साथ TCP सर्वर

विचलन: किसी भी कार्य में उच्च प्रदर्शन की आवश्यकता के रूप में, प्रत्येक विशिष्ट मामले को आपके आवेदन की सुविधाओं के आधार पर माना जाना चाहिए। विभिन्न दृष्टिकोणों के उपयोग पर संसाधनों को खर्च करने के लिए यह समझ में नहीं आ सकता है, जो बाद में चर्चा की जाएगी, अगर नेटवर्क एप्लिकेशन का पैमाना बहुत बड़ा नहीं है।

पाइपलाइनों का उपयोग करने से पहले नियमित .NET कोड कुछ इस तरह दिखता है:

async Task ProcessLinesAsync(NetworkStream stream) { var buffer = new byte[1024]; await stream.ReadAsync(buffer, 0, buffer.Length); // Process a single line from the buffer ProcessLine(buffer); } 

नमूना देखें। github पर क्लिक करें

यह कोड संभवतः स्थानीय परीक्षण के साथ काम करेगा, लेकिन इसमें कई त्रुटियां हैं:

• ReadAsync के एक कॉल के बाद शायद, पूरा संदेश प्राप्त नहीं होगा (लाइन के अंत तक)।
• यह स्ट्रीम के परिणाम को अनदेखा करता है। 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; } // Keep track of the amount of buffered bytes bytesBuffered += bytesRead; var linePosition = -1; do { // Look for a EOL in the buffered data linePosition = Array.IndexOf(buffer, (byte)'\n', bytesConsumed, bytesBuffered - bytesConsumed); if (linePosition >= 0) { // Calculate the length of the line based on the offset var lineLength = linePosition - bytesConsumed; // Process the line ProcessLine(buffer, bytesConsumed, lineLength); // Move the bytesConsumed to skip past the line we consumed (including \n) bytesConsumed += lineLength + 1; } } while (linePosition >= 0); } } 

नमूना 2 देखें। जीथूब पर क्लिक करें

मैं दोहराता हूं: यह स्थानीय परीक्षण के साथ काम कर सकता है, लेकिन कभी-कभी 1 Kb (1024 बाइट्स) से अधिक लंबे तार होते हैं। नई लाइन मिलने तक इनपुट बफर का आकार बढ़ाना आवश्यक है।

इसके अलावा, हम लंबे स्ट्रिंग्स को संसाधित करते समय एक सरणी में बफ़र्स एकत्र करते हैं। हम ArrayPool के साथ इस प्रक्रिया में सुधार कर सकते हैं, जो क्लाइंट से लंबी लाइनों के विश्लेषण के दौरान बफ़र्स के पुन: आवंटन से बचा जाता है।

  async Task ProcessLinesAsync(NetworkStream stream) { byte[] buffer = ArrayPool<byte>.Shared.Rent(1024); var bytesBuffered = 0; var bytesConsumed = 0; while (true) { // Calculate the amount of bytes remaining in the buffer var bytesRemaining = buffer.Length - bytesBuffered; if (bytesRemaining == 0) { // Double the buffer size and copy the previously buffered data into the new buffer var newBuffer = ArrayPool<byte>.Shared.Rent(buffer.Length * 2); Buffer.BlockCopy(buffer, 0, newBuffer, 0, buffer.Length); // Return the old buffer to the pool ArrayPool<byte>.Shared.Return(buffer); buffer = newBuffer; bytesRemaining = buffer.Length - bytesBuffered; } var bytesRead = await stream.ReadAsync(buffer, bytesBuffered, bytesRemaining); if (bytesRead == 0) { // EOF break; } // Keep track of the amount of buffered bytes bytesBuffered += bytesRead; do { // Look for a EOL in the buffered data linePosition = Array.IndexOf(buffer, (byte)'\n', bytesConsumed, bytesBuffered - bytesConsumed); if (linePosition >= 0) { // Calculate the length of the line based on the offset var lineLength = linePosition - bytesConsumed; // Process the line ProcessLine(buffer, bytesConsumed, lineLength); // Move the bytesConsumed to skip past the line we consumed (including \n) bytesConsumed += lineLength + 1; } } while (linePosition >= 0); } } 

नमूना देखें। ३ub जीथूब पर

कोड काम करता है, लेकिन अब बफर आकार बदल गया है, परिणामस्वरूप, इसकी कई प्रतियां दिखाई देती हैं। अधिक मेमोरी का भी उपयोग किया जाता है, क्योंकि तर्क लाइनों को संसाधित करने के बाद बफर को कम नहीं करता है। इससे बचने के लिए, आप बफ़र्स की सूची को सहेज सकते हैं, बफ़र के आकार को बदलने के बजाय हर बार एक स्ट्रिंग 1 Kb से अधिक समय तक आती है।

इसके अलावा, हम 1 केबी के बफर आकार को नहीं बढ़ाते हैं, जब तक कि यह पूरी तरह से खाली न हो। इसका मतलब यह है कि हम छोटे और छोटे बफ़र्स को रीडासिंक में स्थानांतरित करेंगे, परिणामस्वरूप, ऑपरेटिंग सिस्टम पर कॉल की संख्या बढ़ जाएगी।

हम इसे खत्म करने की कोशिश करेंगे और जैसे ही मौजूदा बाइट का आकार 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); } 

नमूना देखें। 4 github पर क्लिक करें

नतीजतन, कोड काफी जटिल है। सीमांकक की खोज के दौरान, हम भरे हुए बफ़र्स को ट्रैक करते हैं। ऐसा करने के लिए, एक सूची का उपयोग करें, जो नई लाइन विभाजक की खोज करते समय बफर डेटा प्रदर्शित करता है। परिणामस्वरूप, ProcessLine और IndexOf बाइट [], ऑफ़सेट और काउंट के बजाय लिस्ट स्वीकार करेंगे। पार्सिंग तर्क बफर के एक खंड या कई को संसाधित करना शुरू कर देगा।

और अब सर्वर आंशिक संदेशों को संसाधित करेगा और समग्र मेमोरी खपत को कम करने के लिए साझा मेमोरी का उपयोग करेगा। हालाँकि, कई बदलाव करने की आवश्यकता है:

1. ArrayPoolbyte से हम केवल बाइट का उपयोग करते हैं [] - मानक रूप से प्रबंधित सरणियाँ। दूसरे शब्दों में, जब ReadAsync या WriteAsync फ़ंक्शन निष्पादित होते हैं, तो बफ़र्स की वैधता अवधि एसिंक्रोनस ऑपरेशन के समय से जुड़ी होती है (ऑपरेटिंग सिस्टम के अपने आई / ओ एपीआई के साथ बातचीत करने के लिए)। चूंकि पिन की गई मेमोरी को स्थानांतरित नहीं किया जा सकता है, यह कचरा संग्राहक के प्रदर्शन को प्रभावित करता है और सरणी के विखंडन का कारण बन सकता है। आपको पूल के कार्यान्वयन को बदलने की आवश्यकता हो सकती है, इस पर निर्भर करता है कि कितने अतुल्यकालिक संचालन निष्पादन के लिए इंतजार करेंगे।
2. पठन और प्रक्रिया तर्क के बीच की कड़ी को तोड़कर थ्रूपुट को बेहतर बनाया जा सकता है। हमें बैच प्रसंस्करण का प्रभाव मिलता है, और अब पार्सिंग लॉजिक बड़ी मात्रा में डेटा को पढ़ने में सक्षम होगा, बफ़र्स के बड़े ब्लॉक को संसाधित करने के बजाय, व्यक्तिगत लाइनों का विश्लेषण करने में सक्षम होगा। नतीजतन, कोड और भी जटिल हो जाता है:
   
   
   • दो चक्र बनाने के लिए आवश्यक है जो एक दूसरे से स्वतंत्र रूप से काम करते हैं। पहला सॉकेट से डेटा पढ़ेगा, और दूसरा बफ़र्स का विश्लेषण करेगा।
   • क्या जरूरत है, पार्सिंग लॉजिक बताने का एक तरीका है कि डेटा उपलब्ध हो रहा है।
   • यह निर्धारित करना भी आवश्यक है कि क्या होता है अगर लूप सॉकेट से डेटा को बहुत तेज़ी से पढ़ता है। हमें पढ़ने के चक्र को समायोजित करने के तरीके की आवश्यकता है यदि पार्सिंग तर्क इसके साथ नहीं है। इसे आमतौर पर "प्रवाह नियंत्रण" या "प्रवाह प्रतिरोध" के रूप में जाना जाता है।
   • हमें यह सुनिश्चित करना चाहिए कि डेटा सुरक्षित रूप से प्रसारित हो। अब बफ़र्स के सेट का उपयोग रीड साइकिल और पार्सिंग चक्र दोनों द्वारा किया जाता है, वे अलग-अलग थ्रेड्स पर एक-दूसरे से स्वतंत्र रूप से काम करते हैं।
   • मेमोरी प्रबंधन तर्क कोड के दो अलग-अलग टुकड़ों में भी शामिल है: बफर पूल से डेटा उधार लेना, जो सॉकेट से डेटा पढ़ता है, और बफर पूल से लौटता है, जो पार्सिंग लॉजिक है।
   • पार्सिंग लॉजिक को अंजाम देने के बाद बफ़र्स को लौटाने में बेहद सावधानी बरतनी चाहिए। अन्यथा, एक मौका है कि हम उस बफर को वापस कर देंगे जिसमें सॉकेट रीड लॉजिक अभी भी लिखा जा रहा है।

जटिलता छत के माध्यम से जाना शुरू होती है (और यह सभी मामलों से बहुत दूर है!)। उच्च-प्रदर्शन नेटवर्क बनाने के लिए, आपको बहुत ही जटिल कोड लिखना होगा।

System.IO.Pipelines का उद्देश्य इस प्रक्रिया को सरल बनाना है।

टीसीपी सर्वर और सिस्टम

आइए देखें कि 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) { // Allocate at least 512 bytes from the PipeWriter Memory<byte> memory = writer.GetMemory(minimumBufferSize); try { int bytesRead = await socket.ReceiveAsync(memory, SocketFlags.None); if (bytesRead == 0) { break; } // Tell the PipeWriter how much was read from the Socket writer.Advance(bytesRead); } catch (Exception ex) { LogError(ex); break; } // Make the data available to the PipeReader FlushResult result = await writer.FlushAsync(); if (result.IsCompleted) { break; } } // Tell the PipeReader that there's no more data coming writer.Complete(); } async Task ReadPipeAsync(PipeReader reader) { while (true) { ReadResult result = await reader.ReadAsync(); ReadOnlySequence<byte> buffer = result.Buffer; SequencePosition? position = null; do { // Look for a EOL in the buffer position = buffer.PositionOf((byte)'\n'); if (position != null) { // Process the line ProcessLine(buffer.Slice(0, position.Value)); // Skip the line + the \n character (basically position) buffer = buffer.Slice(buffer.GetPosition(1, position.Value)); } } while (position != null); // Tell the PipeReader how much of the buffer we have consumed reader.AdvanceTo(buffer.Start, buffer.End); // Stop reading if there's no more data coming if (result.IsCompleted) { break; } } // Mark the PipeReader as complete reader.Complete(); } 

देखें g5ub पर sample5.cs

हमारे लाइन रीडर के पाइपलाइन संस्करण में दो छोरें हैं:

• FillPipeAsync सॉकेट से पढ़ता है और पाइपविटर को लिखता है।
• ReadPipeAsync पाइपराइडर से पढ़ता है और आने वाली लाइनों को पार्स करता है।

पहले उदाहरणों के विपरीत, कोई विशेष रूप से निर्दिष्ट बफ़र्स नहीं हैं। यह System.IO.Pipelines के मुख्य कार्यों में से एक है। सभी बफ़र प्रबंधन कार्य पाइपराइडर / पाइपविटर कार्यान्वयन पर स्थानांतरित किए जाते हैं।

प्रक्रिया को सरल बनाया गया है: हम जटिल बफर प्रबंधन को लागू करने के बजाय केवल व्यावसायिक तर्क के लिए कोड का उपयोग करते हैं।

पहले लूप में, मुख्य लेखक से निश्चित मात्रा में मेमोरी प्राप्त करने के लिए सबसे पहले पाइपवेटर.गेटमैरी (इंट) को बुलाया जाता है। फिर पाइपवेटर। एडवांस (इंट) कहा जाता है, जो पाइपविटर को बताता है कि वास्तव में बफर को कितना डेटा लिखा गया है। इसके बाद pipWriter.FlushAsync () को कॉल किया जाता है ताकि पाइपआरडर डेटा तक पहुंच सके।

दूसरा लूप उन बफ़र्स का उपभोग करता है जो पाइपवर्टर द्वारा लिखे गए थे लेकिन मूल रूप से सॉकेट से प्राप्त हुए थे। जब pipReader.ReadAsync () को वापस करने का अनुरोध किया जाता है, तो हमें एक ReadResult मिलता है जिसमें दो महत्वपूर्ण संदेश होते हैं: डेटा ReadOnlySequence, साथ ही तार्किक डेटा प्रकार IsCompleted में पढ़ा जाता है, जो पाठक को बताता है कि क्या लेखक ने काम करना समाप्त कर दिया है (EOF)। जब लाइन टर्मिनेटर (ईओएल) पाया जाता है और स्ट्रिंग को पार्स किया जाता है, तो हम पहले से संसाधित किए गए टुकड़े को छोड़ने के लिए बफर को भागों में विभाजित करेंगे। उसके बाद, पाइपराइडर.एडवेंसटॉ को कहा जाता है और यह पाइपरीडर को बताता है कि कितना डेटा खपत किया गया है।

प्रत्येक चक्र के अंत में, पाठक और लेखक दोनों पूर्ण होते हैं। नतीजतन, मुख्य चैनल सभी आवंटित मेमोरी जारी करता है।

System.IO.Pipelines

आंशिक पढ़ना

मेमोरी को प्रबंधित करने के अलावा, System.IO.Pipelines एक और महत्वपूर्ण कार्य करता है: यह चैनल में डेटा को स्कैन करता है, लेकिन इसका उपभोग नहीं करता है।

पाइप-राइडर के दो मुख्य एपीआई हैं: ReadAsync और AdvanceTo। ReadAsync चैनल से डेटा प्राप्त करता है, AdvanceTo पाइपरीडर को बताता है कि इन बफ़र्स को अब रीडर की आवश्यकता नहीं है, इसलिए आप उनसे छुटकारा पा सकते हैं (उदाहरण के लिए, उन्हें मुख्य बफर पूल में लौटा दें)।

निम्नलिखित एक HTTP विश्लेषक का एक उदाहरण है जो आंशिक चैनल डेटा बफ़र्स से डेटा पढ़ता है जब तक कि यह एक उपयुक्त स्टार्ट लाइन प्राप्त नहीं करता है।

ReadOnlySequenceT

चैनल कार्यान्वयन पाइपवेटर और पाइपरीडर के बीच पारित संबंधित बफ़र्स की एक सूची संग्रहीत करता है। PipReader.ReadAsync ReadOnlySequence को उजागर करता है, जो BCL का एक नया प्रकार है और इसमें एक या अधिक ReadOnlyMemory <T> सेगमेंट होते हैं। यह स्पैन या मेमोरी के समान है, जो हमें सरणियों और तारों को देखने का अवसर देता है।



चैनल के अंदर ऐसे पॉइंटर्स होते हैं जो यह दर्शाते हैं कि पाठक और लेखक हाइलाइट किए गए डेटा के सामान्य सेट में स्थित हैं, और उन्हें डेटा लिखने और पढ़ने के रूप में अपडेट भी करें। SequencePosition बफ़र्स की लिंक की गई सूची में एक एकल बिंदु है और कुशलतापूर्वक ReadOnlySequence <T> को अलग करने के लिए उपयोग किया जाता है।

चूंकि ReadOnlySequence <T> एक या अधिक सेगमेंट का समर्थन करता है, इसलिए उच्च-प्रदर्शन तर्क का मानक संचालन सेगमेंट की संख्या के आधार पर तेज़ और धीमी पथ को अलग करना है।

एक उदाहरण के रूप में, यहाँ एक फ़ंक्शन है जो ASCII ReadOnlySequence को एक स्ट्रिंग में परिवर्तित करता है:

  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); } }); } 

नमूना देखें। github पर क्लिक करें

प्रवाह प्रतिरोध और प्रवाह नियंत्रण

आदर्श रूप से, पढ़ने और विश्लेषण एक साथ काम करते हैं: रीड स्ट्रीम नेटवर्क से डेटा की खपत करता है और इसे बफ़र्स में रखता है, जबकि विश्लेषण स्ट्रीम कुछ अन्य संरचनाएं बनाती है। विश्लेषण आमतौर पर नेटवर्क से डेटा ब्लॉक की नकल करने की तुलना में अधिक समय लेता है। नतीजतन, रीड स्ट्रीम आसानी से विश्लेषण स्ट्रीम को अधिभारित कर सकती है। इसलिए, विश्लेषण स्ट्रीम के लिए डेटा को बचाने के लिए रीड स्ट्रीम को या तो धीमा करना होगा या अधिक मेमोरी का उपभोग करना होगा। इष्टतम प्रदर्शन सुनिश्चित करने के लिए, ठहराव आवृत्ति और बड़ी मात्रा में स्मृति के आवंटन के बीच एक संतुलन की आवश्यकता होती है।

इस समस्या को हल करने के लिए, पाइपलाइन में दो डेटा प्रवाह नियंत्रण कार्य हैं: पॉज़वराइटरट्रेशोल्ड और रिज्यूमवेयर राइटरहॉल्ड। PauseWriterThreshold यह निर्धारित करता है कि पाइपविटर से पहले कितने डेटा को बफर करना है। ResumeWriterThreshold यह निर्धारित करता है कि रिकॉर्डर फिर से शुरू होने से पहले पाठक कितनी मेमोरी का उपभोग कर सकता है।



PipWriter.FlushAsync "लॉक" तब करता है जब पाइपलाइन्ड स्ट्रीम में डेटा की मात्रा पॉज़वर्टर थ्रेशोल्ड में सेट की गई सीमा से अधिक हो जाती है, और जब रेज्यूमेविटर थ्रेशोल्ड में निर्धारित सीमा से कम हो जाती है, तो "अनलॉक" हो जाता है। खपत की सीमा को पार करने से रोकने के लिए, केवल दो मूल्यों का उपयोग किया जाता है।

I / O शेड्यूलिंग

एसिंक्स / वेट का उपयोग करते समय, बाद के संचालन को आमतौर पर या तो पूल थ्रेड्स में या वर्तमान सिंक्रोनाइज़ेशन कॉनटेक्स्ट में कहा जाता है।

I / O का प्रदर्शन करते समय, यह सावधानीपूर्वक मॉनिटर करना बहुत महत्वपूर्ण है कि प्रोसेसर कैश का बेहतर उपयोग करने के लिए इसे कहाँ निष्पादित किया जाता है। यह वेब सर्वर जैसे उच्च-प्रदर्शन अनुप्रयोगों के लिए महत्वपूर्ण है। System.IO.Pipelines, एसिंक्रोनस कॉलबैक निष्पादित करने के लिए निर्धारित करने के लिए पाइपस्क्रिड्यूलर का उपयोग करता है। यह आपको बहुत ही सटीक नियंत्रण की अनुमति देता है जो I / O के लिए उपयोग करने के लिए स्ट्रीम करता है।

व्यावहारिक अनुप्रयोग का एक उदाहरण Kestrel Libuv परिवहन है, जिसमें I / O कॉलबैक इवेंट लूप के समर्पित चैनलों पर किए जाते हैं।

पाइपरेडर टेम्पलेट के अन्य लाभ हैं।

• कुछ बेस सिस्टम "बफरिंग के बिना प्रतीक्षा करें" का समर्थन करते हैं: आपको बेस सिस्टम में डेटा उपलब्ध होने तक बफर को आवंटित करने की आवश्यकता नहीं है। इसलिए, एपोल के साथ लिनक्स पर, आप डेटा तैयार होने तक रीड बफर प्रदान नहीं कर सकते। यह उस स्थिति से बचा जाता है जब डेटा के लिए कई थ्रेड इंतजार कर रहे होते हैं, और आपको तुरंत बड़ी मात्रा में मेमोरी आरक्षित करने की आवश्यकता होती है।
• डिफ़ॉल्ट पाइपलाइन नेटवर्क कोड के यूनिट टेस्ट लिखना आसान बनाता है: पार्सिंग लॉजिक नेटवर्क कोड से अलग है, और यूनिट टेस्ट केवल मेमोरी में बफ़र्स में इस लॉजिक को चलाते हैं, बजाय इसके नेटवर्क से सीधे उपभोग करने के। यह आंशिक डेटा भेजकर जटिल पैटर्न का परीक्षण करना भी आसान बनाता है। ASP.NET Core इसका उपयोग Kestrel के http पार्सिंग टूल्स के विभिन्न पहलुओं का परीक्षण करने के लिए करता है।
• सिस्टम जो उपयोगकर्ता कोड को मुख्य ओएस बफ़र्स का उपयोग करने की अनुमति देते हैं (उदाहरण के लिए, पंजीकृत विंडोज I / O एपीआई) पाइपलाइनों का उपयोग करने के लिए शुरू में उपयुक्त हैं, क्योंकि पाइपरेडर कार्यान्वयन हमेशा बफ़र्स प्रदान करता है।

अन्य संबंधित प्रकार

हमने System.IO.Pipelines पर कई नए सरल BCL प्रकार भी जोड़े:

• MemoryPoolT , IMemoryOwnerT , MemoryManagerT । ArrayPoolT को .NET Core 1.0 में जोड़ा गया था, और .NET Core 2.1 में अब पूल के लिए अधिक सामान्य सार प्रतिनिधित्व है जो किसी भी मेमोरीटी के साथ काम करता है। हमें एक एक्स्टेंसिबिलिटी बिंदु मिलता है जो हमें अधिक उन्नत वितरण रणनीतियों को लागू करने की अनुमति देता है, साथ ही साथ बफर प्रबंधन को नियंत्रित करता है (उदाहरण के लिए, विशेष रूप से प्रबंधित सरणियों के बजाय पूर्वनिर्धारित बफ़र्स का उपयोग करें)।
• IBufferWriterT सिंक्रनाइज़ बफ़र्ड डेटा (पाइपवर्टर द्वारा कार्यान्वित) को रिकॉर्ड करने के लिए एक रिसीवर है।
• IValueTaskSource - ValueTaskT .NET कोर 1.1 की रिलीज़ के बाद से अस्तित्व में है, लेकिन .NET Core 2.1 में इसने बेहद प्रभावी उपकरण प्राप्त किए हैं जो वितरण के बिना निर्बाध अतुल्यकालिक संचालन प्रदान करते हैं। अधिक जानकारी के लिए यहाँ देखें।

कन्वेयर का उपयोग कैसे करें?

API नगेट पैकेज System.IO.Pipelines में हैं ।

उदाहरण के लिए .NET सर्वर 2.1 सर्वर अनुप्रयोग जो निचली संदेशों (ऊपर के उदाहरण से) को संसाधित करने के लिए पाइपलाइनों का उपयोग करता है, यहां देखें। इसे डॉटनेट रन (या विजुअल स्टूडियो) का उपयोग करके शुरू किया जा सकता है। उदाहरण में, पोर्ट 8087 पर सॉकेट से डेटा प्रसारित होने की उम्मीद है, फिर प्राप्त संदेश कंसोल पर लिखे गए हैं। पोर्ट 8087 से कनेक्ट करने के लिए आप एक क्लाइंट, जैसे कि नेटकैट या पोटीन का उपयोग कर सकते हैं। एक लोअरकेस संदेश भेजें और देखें कि यह कैसे काम करता है।

वर्तमान में, पाइपलाइन Kestrel और SignalR पर चलती है, और हमें उम्मीद है कि यह भविष्य में कई नेटवर्क पुस्तकालयों और .NET समुदाय के घटकों में व्यापक आवेदन प्राप्त करेगा।