串口使用Pipeline时诡异的ReadOnlySequence问题

借鉴之前的Pipeline的操作方式,现在目标是给串口读取操作也使用上Pipeline。稍微改造一下,以下代码可以直接运行。

协议为使用连续的4个0XFF作为结尾,没有头标志。数据总长为68位定长。

我需要判断从开始到选定位置是否长度足够,这里面用来判断segment长度我用了这个方式。

buffer.Slice(0, start.Value).Length >= 64

其实最早不是使用这个东西的,而是使用的SequencePosition的GetInteger()方法,获取到了位置的index,自然就知道了长度等信息,而且非常方便进行截取操作。可是在使用的时候,发现一个非常诡异的问题:通过这个方法获取到的index值要大于Buffer的总长度。Slice直接弹出ArgumentOutOfRangeException,但是不弹出错误,调试的时候非常麻烦。

查看这个方法定义的时候,发现签名是这样的:

[EditorBrowsable(EditorBrowsableState.Never)]
public int GetInteger();

这个东西第一次见到,VS并不会提示,但是你强行写的话,能够正常编译。看来微软并不是很像让我们看到这个玩意。仔细挖掘一下,发现通过PositionOf方法获得的SequencePosition内部引用了一段长度为4096的内存。这个GetInteger()有时候返回的是在这段Memory上面的Index值。
串口使用Pipeline时诡异的ReadOnlySequence问题
看来这个东西是内部使用的,不太推荐我们使用。

有几个问答也说到了这个事情:问答1问答2

我们只能使用GetPosition方法来获得相对的位置。不过我用的这个设备,协议是尾部标志,如果使用PostionOf的话,偏移量得是负数。在尝试了很多次不通之后,发现微软文档中有说到这个潜在问题:

  • SequencePosition 是特定 ReadOnlySequence 的位置标记,而不是绝对位置。 由于它是相对于特定 ReadOnlySequence 的,因此如果在其起源的 ReadOnlySequence 之外使用,则没有意义。
  • 不能对没有 ReadOnlySequence 的 SequencePosition 执行算术运算。 这意味着,执行 position++ 等基本操作将以 ReadOnlySequence .GetPosition(position, 1) 的形式写入。
  • GetPosition(long) 不支持负索引。 这意味着,如果没有遍历所有段,就无法获取倒数第二个字符。
    后面还有一些我就不贴了,总之。那我这种情况只能使用buffer.Slice(0, pos)获得最长的片段,并将内容传输给处理程序进行。

总之,不要使用操作ReadOnlySpan /ReadOnlyMemory 的思路来操作ReadOnlySequence /SequencePosition!

为了解决操作的复杂性,.NET Core 3.0引入了一个SequenceReader<T>简化了操作,以后有机会使用的时候在写吧。

最后程序如下:


private async void Sp_DataReceived(object sender, SerialDataReceivedEventArgs e)
        {
            SerialPort p = sender as SerialPort;
            byte[] bytes = new byte[1024 * 4];
            var dataCount = p.Read(bytes, 0, p.BytesToRead);
            var span = bytes.AsMemory().Slice(0, dataCount);
            await FillPipeAsync(span);
        }

private void InitPipe()
        {
            Pipe pipe = new Pipe();
            writer = pipe.Writer;
            //Task writing = FillPipeAsync(port, pipe.Writer);
            Task.Run(() => ReadPipeAsync(pipe.Reader));
            //await Task.WhenAll(reading, writing);
        }

        private PipeWriter writer;

        private async Task FillPipeAsync(ReadOnlyMemory<byte> memory)
        {
            await writer.WriteAsync(memory);
            //writer.Advance(memory.Length);

            // Make the data available to the PipeReader
            FlushResult result = await writer.FlushAsync();
            if (result.IsCompleted)
                writer.Complete();
        }

        private async Task ReadPipeAsync(PipeReader reader)
        {
            while (true)
            {
                try
                {
                    ReadResult result = await reader.ReadAsync();
                    ReadOnlySequence<byte> buffer = result.Buffer;

                    SequencePosition? start = null;
                    var headBytes = new byte[] { 0xff, 0xff, 0xff, 0xff };

                    do
                    {


                        // Find the EOL
                        start = buffer.PositionOf(headBytes[0]);

                        if (start != null)
                        {
                            if (buffer.Slice(start.Value).Length >= 4)
                            {
                                var headtoCheck = buffer.Slice(start.Value, 4).ToArray();
                                if (headtoCheck.SequenceEqual(headBytes))
                                {
                                    if(buffer.Slice(0, start.Value).Length >= 64)
                                    {
                                        var pos = buffer.GetPosition(4, start.Value);
                                        var mes = buffer.Slice(0, pos);
                                        DataProcess.Process(mes.ToArray());
                                        var next = buffer.GetPosition(4, start.Value);
                                        buffer = buffer.Slice(next);
                                    }
                                    else
                                    {
                                        var next = buffer.GetPosition(4, start.Value);
                                        buffer = buffer.Slice(next);
                                    }
                                }
                                else
                                {
                                    var next = buffer.GetPosition(1, start.Value);
                                    buffer = buffer.Slice(next);
                                }
                            }
                            else
                            {
                                var next = buffer.GetPosition(1, start.Value);
                                buffer = buffer.Slice(next);
                            }
                        }


                    }
                    while (start != null);
                    // We sliced the buffer until no more data could be processed
                    // Tell the PipeReader how much we consumed and how much we left to process

                    if (result.IsCompleted)
                    {
                        continue;
                    }
                    reader.AdvanceTo(buffer.Start, buffer.End);
                }
                catch (ArgumentOutOfRangeException e)
                {

                    //throw e;
                }
            }
            reader.Complete();
        }

发表评论

评论已关闭。

相关文章