Pipe——高性能IO(一)-FinClip官网

Pipe——高性能IO(一)

网友投稿 944 2022-09-11

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;

{

// 在缓冲数据中查找找一个行末尾

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避免重复分配缓冲区来改进这一点。

async Task ProcessLinesAsync(NetworkStream stream)

{

byte[] buffer = ArrayPool.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.Shared.Rent(buffer.Length * 2);

Buffer.BlockCopy(buffer, 0, newBuffer, 0, buffer.Length);

// 将旧的buffer丢回池中

ArrayPool.Shared.Return(buffer);

buffer = newBuffer;

bytesRemaining = buffer.Length - bytesBuffered;

}

var bytesRead = await stream.ReadAsync(buffer, bytesBuffered, bytesRemaining);

if (bytesRead == 0)

{

// EOF 末尾

break;

}

// 跟踪已缓冲的字节数

bytesBuffered += bytesRead;

{

// 在缓冲数据中查找找一个行末尾

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

var bytesConsumed = 0;

var bytesConsumedBufferIndex = 0;

var segment = new BufferSegment { Buffer = ArrayPool.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.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.Shared.Return(consumedSegment.Buffer);

segments.RemoveAt(i);

}

(int segmentIndex, int segmentOffest) IndexOf(List 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查找新行分隔符时表示缓冲数据。其结果是，ProcessLine和IndexOf现在接受List作为参数，而不是一个byte[]，offset和count。我们的解析逻辑现在需要处理一个或多个缓冲区序列。

我们的服务器现在处理部分消息，它使用池化内存来减少总体内存消耗，但我们还需要进行更多更改：

我们使用的byte[]和ArrayPool的只是普通的托管数组。这意味着无论何时我们执行ReadAsync或WriteAsync，这些缓冲区都会在异步操作的生命周期内被固定（以便与操作系统上的本机IO API互操作）。这对GC有性能影响，因为无法移动固定内存，这可能导致堆碎片。根据异步操作挂起的时间长短，池的实现可能需要更改。

可以通过解耦读取逻辑和处理逻辑来优化吞吐量。这会创建一个批处理效果，使解析逻辑可以使用更大的缓冲区块，而不是仅在解析单个行后才读取更多数据。这引入了一些额外的复杂性

我们需要两个彼此独立运行的循环。一个读取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 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 buffer = result.Buffer;

SequencePosition? position = null;

{

// 在缓冲数据中查找找一个行末尾

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形式读取的数据和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

该Pipe实现存储了在PipeWriter和PipeReader之间传递的缓冲区的链接列表。PipeReader.ReadAsync暴露一个ReadOnlySequence新的BCL类型，它表示一个或多个ReadOnlyMemory段的视图，类似于Span和Memory提供数组和字符串的视图。

该Pipe内部维护指向reader和writer可以分配或更新它们的数据集合，。SequencePosition表示缓冲区链表中的单个点，可用于有效地对ReadOnlySequence进行切片。

PHP 对接微信公众平台消息接口开发流程实例详解

944 2022-09-11

Pipe——高性能IO(一)

微信小程序本地存储与登录页面处理实例详细讲解

PHP 对接微信公众平台消息接口开发流程实例详解

微信公众平台开发入门教程图文详解全面呈现

最近发表

更多内容

小程序SDK

Finclip技术文档

小程序开发

小程序容器

小程序框架

Finclip小程序平台

Finclip用户投稿

车联网

推荐文章

小程序SDK是什么意思？小程序sdk和插件有什么区别？

小程序支付功能怎么实现？

企业app开发流程是什么？

app运营模式有哪些？

小程序多端引流怎么做？

小程序生态分析的机会和威胁

Flutter入门这一篇效率文章就够了

原生与跨平台解决方案分析,跨平台软件开发技术方案

热更新技术：让软件更新变得更加轻松快速

解决方案

银行解决方案

证券解决方案

互联网解决方案

政企OA解决方案

科技解决方案

loT解决方案

信任解决方案

热评文章

AppCan:基于混合模式的移动应用开发,移动混合模

Hybrid App混合模式开发的了解

小程序容器技术助力券商数字营销突围，小程序容器化的意

用mpvue开发微信小程序基础知识（vue.js开发

小程序多端框架全面测评对比，强烈推荐！

开放银行银行案例，迎接金融创新的未来