串口通信应用的陷阱 - 涵盖重连与日志设计

using System;
using System.Buffers.Binary;
using System.Collections.Generic;
using System.Diagnostics;

public static class Crc16Modbus
{
    // CRC-16/MODBUS: 初始值 0xFFFF、多项式 0xA001 的右移
    public static ushort Compute(ReadOnlySpan<byte> data)
    {
        ushort crc = 0xFFFF;
        foreach (var b in data)
        {
            crc ^= b;
            for (var i = 0; i < 8; i++)
            {
                crc = (crc & 1) != 0 ? (ushort)((crc >> 1) ^ 0xA001) : (ushort)(crc >> 1);
            }
        }

        return crc;
    }
}

public static class Frame
{
    public const byte Stx = 0x02;
    public const int HeaderLength = 2;   // STX + LEN
    public const int CrcLength = 2;

    public static byte[] Build(ReadOnlySpan<byte> payload)
    {
        // LEN 只有 1 byte，所以 256 字节以上会在强制转换时回绕。
        // 而 payload 仍然被整块复制，于是接收方按被截断的长度切帧，
        // 把 payload 中间的某处读成 CRC。之后的帧边界也会全线崩溃。
        // 是拆分发送还是把 LEN 改成 2 byte，属于协议的约定，
        // 这里只负责拦下来
        if (payload.Length > byte.MaxValue)
        {
            throw new ArgumentOutOfRangeException(
                nameof(payload),
                $"一帧的 payload 最多 {byte.MaxValue} 字节（因为 LEN 只有 1 byte）。");
        }

        var frame = new byte[HeaderLength + payload.Length + CrcLength];
        frame[0] = Stx;
        frame[1] = (byte)payload.Length;
        payload.CopyTo(frame.AsSpan(HeaderLength));

        var body = frame.AsSpan(0, frame.Length - CrcLength);
        BinaryPrimitives.WriteUInt16LittleEndian(frame.AsSpan(frame.Length - CrcLength), Crc16Modbus.Compute(body));
        return frame;
    }
}

public sealed class FrameParser
{
    private readonly List<byte> _buffer = new();

    /// <summary>3.3 说的 inter-byte timeout。放弃正在组装的帧之前等待的时间。</summary>
    private static readonly TimeSpan AssemblyTimeout = TimeSpan.FromMilliseconds(200);

    /// <summary>当前正在组装的候选，是从什么时候开始进入等待的（单调递增的值）。</summary>
    private long _pendingSince;

    /// <summary>通知 CRC 不匹配而被丢弃的帧。为了写进日志，务必订阅。</summary>
    public event Action<byte[]>? FrameDiscarded;

    /// <summary>通知放弃组装并重新同步。如果这里持续增加，就要怀疑接线或设置。</summary>
    public event Action<int>? Resynchronized;

    /// <summary>把收到的字节序列存起来，只返回已经能切出来的帧。</summary>
    public IReadOnlyList<byte[]> Append(ReadOnlySpan<byte> received)
    {
        foreach (var b in received)
        {
            _buffer.Add(b);
        }

        var frames = new List<byte[]>();

        while (true)
        {
            // 1. 一直丢到开头是 STX 为止。噪声和上一帧的残留在这里被吸收掉
            var stxIndex = _buffer.IndexOf(Frame.Stx);
            if (stxIndex < 0)
            {
                _buffer.Clear();
                _pendingSince = 0;   // 候选没有了，等待计时也停下来
                break;
            }

            if (stxIndex > 0)
            {
                // 候选的开头变了 = 开始组装另一帧
                _buffer.RemoveRange(0, stxIndex);
                _pendingSince = 0;
            }

            // 2. 是否已经收到能读出长度的位置
            if (_buffer.Count < Frame.HeaderLength)
            {
                if (GiveUpOnStaleCandidate()) { continue; }
                break;   // 不是“已损坏”，而是“还没收全”
            }

            int payloadLength = _buffer[1];
            int frameLength = Frame.HeaderLength + payloadLength + Frame.CrcLength;

            // 3. 是否已经凑齐一整帧
            if (_buffer.Count < frameLength)
            {
                // “还没收全”和“LEN 被噪声打乱了”在这个时刻还
                // 分不出来。噪声或者假的 STX 让 LEN 变成 255 时，
                // 之后到达的正确帧也会被当作 payload 一直吸进来，
                // 直到凑满 259 字节被 CRC 判掉之前什么都不会上报。
                // 在通信量小的设备上，这看起来就是几分钟没有反应。
                // 给等待设一个上限，超过就丢掉候选，重新去找 STX
                if (GiveUpOnStaleCandidate()) { continue; }
                break;   // 在这里退出，等待下一次接收
            }

            var frame = _buffer.GetRange(0, frameLength).ToArray();
            _buffer.RemoveRange(0, frameLength);
            _pendingSince = 0;

            // 4. CRC 不匹配的丢掉。丢掉这件事一定要报到外面
            var expected = BinaryPrimitives.ReadUInt16LittleEndian(frame.AsSpan(frame.Length - Frame.CrcLength));
            if (expected == Crc16Modbus.Compute(frame.AsSpan(0, frame.Length - Frame.CrcLength)))
            {
                frames.Add(frame);
            }
            else
            {
                // 在这里是整帧一起丢掉，还是只丢 1 byte 的 STX 重新读，属于设计判断。
                // 前者简单，后者在 LEN 本身就是噪声时更稳。选哪个要定下来并写进文档。
                FrameDiscarded?.Invoke(frame);
            }
        }

        return frames;
    }

    /// <summary>
    /// 如果正在组装的候选超过了 AssemblyTimeout，就只丢掉开头 1 byte 的 STX。
    /// 丢掉时返回 true，调用方从下一个 STX 重新读。
    /// 不整帧丢掉，是因为真正的 STX 有可能就埋在这个候选里面。
    /// </summary>
    private bool GiveUpOnStaleCandidate()
    {
        if (_pendingSince == 0)
        {
            // 刚开始等待的这一刻。墙上时钟会因为 NTP 同步而跳变，所以用单调递增的值来计时
            _pendingSince = Stopwatch.GetTimestamp();
            return false;
        }

        if (Stopwatch.GetElapsedTime(_pendingSince) < AssemblyTimeout)
        {
            return false;
        }

        _buffer.RemoveAt(0);
        _pendingSince = 0;
        Resynchronized?.Invoke(_buffer.Count);
        return true;
    }
}

using System;
using System.IO.Ports;
using System.Threading;
using System.Threading.Tasks;

public sealed class SerialReader
{
    private readonly SerialPort _port;
    private readonly FrameParser _parser;
    private readonly byte[] _readBuffer = new byte[4096];

    public SerialReader(SerialPort port, FrameParser parser)
    {
        _port = port;
        _parser = parser;
    }

    public event Action<byte[]>? FrameReceived;

    public async Task RunAsync(CancellationToken token)
    {
        while (!token.IsCancellationRequested)
        {
            int count;
            try
            {
                count = await _port.BaseStream.ReadAsync(_readBuffer.AsMemory(), token);
            }
            catch (OperationCanceledException)
            {
                break;
            }

            if (count <= 0)
            {
                continue;
            }

            foreach (var frame in _parser.Append(_readBuffer.AsSpan(0, count)))
            {
                FrameReceived?.Invoke(frame);
            }
        }
    }
}

using System;
using System.IO.Ports;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

public sealed class SingleWriter
{
    private sealed record Outbound(byte[] FrameBytes, TaskCompletionSource<byte[]> Completion);

    /// <summary>发送队列的上限。按设备一次往返的时间 × 可以容忍的排队长度来定。</summary>
    private const int QueueCapacity = 64;

    private readonly SerialPort _port;
    private readonly TimeSpan _responseTimeout;

    // 不要做成无上限。UI、定时器、worker 放入的速度快过设备的往返时，
    // 帧和 TaskCompletionSource 会无止境地堆积，设备明明在应答，
    // 内存却一路往上涨。定好上限，溢出时就返回给发送方
    private readonly Channel<Outbound> _queue = Channel.CreateBounded<Outbound>(
        new BoundedChannelOptions(QueueCapacity)
        {
            // 满了 TryWrite 就返回 false。调用方能立刻知道“现在堵住了”。
            // 不用 DropOldest —— 放入的一方正在等待 Task，
            // 默默丢掉就会永远等不回来
            FullMode = BoundedChannelFullMode.Wait,
            SingleReader = true,
        });

    private Outbound? _inFlight;

    public SingleWriter(SerialPort port, TimeSpan responseTimeout)
    {
        _port = port;
        _responseTimeout = responseTimeout;
    }

    /// <summary>从 UI 调用还是从定时器调用都可以。真正的 Write 只有一个 worker 执行。</summary>
    public Task<byte[]> SendAsync(ReadOnlySpan<byte> payload)
    {
        var item = new Outbound(
            Frame.Build(payload),
            new TaskCompletionSource<byte[]>(TaskCreationOptions.RunContinuationsAsynchronously));

        if (!_queue.Writer.TryWrite(item))
        {
            // 队列满了，或者 worker 已经停止。两种情况都要把“没能放进去”
            // 返回给调用方。默默丢掉的话，等待中的 Task 就永远回不来
            item.Completion.TrySetException(new InvalidOperationException(
                $"未能放入发送队列（上限 {QueueCapacity} 条，或者 worker 已停止）。"));
        }

        return item.Completion.Task;
    }

    /// <summary>parser 切出帧之后调用。把它和正在等待应答的那一条关联起来。</summary>
    public void OnFrameReceived(byte[] frame)
    {
        var pending = Interlocked.Exchange(ref _inFlight, null);
        pending?.Completion.TrySetResult(frame);
    }

    public async Task RunAsync(CancellationToken token)
    {
        try
        {
            await foreach (var item in _queue.Reader.ReadAllAsync(token))
            {
                Interlocked.Exchange(ref _inFlight, item);
                try
                {
                    await _port.BaseStream.WriteAsync(item.FrameBytes.AsMemory(), token);
                }
                catch (Exception ex)
                {
                    // 断线、端口被关闭、取消都会从这里抛出来。
                    // 不把这一条完成掉就退出，await SendAsync 的
                    // 调用方就会永远等下去
                    Interlocked.Exchange(ref _inFlight, null);
                    item.Completion.TrySetException(ex);
                    throw;
                }

                // 因为在这里一直等到应答，所以下一条命令不会插队
                var timeout = Task.Delay(_responseTimeout, token);
                var finished = await Task.WhenAny(item.Completion.Task, timeout);

                if (finished != item.Completion.Task)
                {
                    // 被要求停止时，Task.Delay 也会被取消而先结束。
                    // 不看这一点，正常的停止流程就会被当成超时，
                    // 调用方会收到 TimeoutException
                    token.ThrowIfCancellationRequested();

                    // WhenAny 选中 timeout 之后，应答仍有可能到达。
                    // 那时 OnFrameReceived 已经取走了 _inFlight，把这一条
                    // 以成功完成掉了。取不到就是应答赢了，这里即使发出
                    // TrySetException 也不会生效。没注意到这一点而
                    // 一路走到 throw，调用方明明已经拿到了结果，
                    // 却只有 worker 挂掉。这场争抢用 CompareExchange 来定胜负
                    if (Interlocked.CompareExchange(ref _inFlight, null, item) != item)
                    {
                        // 应答赢了。完成由 OnFrameReceived 写入（就在刚才）
                        await item.Completion.Task;
                        continue;
                    }

                    item.Completion.TrySetException(new TimeoutException("没有收到应答。"));

                    // 一旦超时，这条连接就已经不可信了。
                    // 不能用“放弃这条，接着发下一条”来收场，理由写在下面：
                    // 这个协议里没有请求 ID，所以迟到的 A 的应答
                    // 会被关联成下一条 B 的应答。按 3.6 的状态迁移图
                    // 落到 Fault，重新生成 session
                    throw new TimeoutException("因为没有应答，将重新生成 session。");
                }
            }
        }
        finally
        {
            // 不管 worker 是因为什么停下来的，都必须把正在等待的条目结束掉。
            // 包括正在等待应答的那一条，以及还堆在队列里没发出去的部分
            var stopped = new OperationCanceledException("发送 worker 已停止。");
            Interlocked.Exchange(ref _inFlight, null)?.Completion.TrySetException(stopped);
            _queue.Writer.TryComplete();
            while (_queue.Reader.TryRead(out var pending))
            {
                pending.Completion.TrySetException(stopped);
            }
        }
    }
}

using System;
using System.IO.Ports;
using System.Threading;
using System.Threading.Tasks;

// 把 3.4 里定下来的设置集中放在一起，而不是散落成就地的字面值
using var port = new SerialPort("COM3", 115200, Parity.None, 8, StopBits.One)
{
    Handshake = Handshake.None,
    DtrEnable = true,
    RtsEnable = true,
    ReadTimeout = 500,
    WriteTimeout = 500,
};

Console.WriteLine($"open {port.PortName} baud={port.BaudRate} data={port.DataBits} parity={port.Parity} " +
                  $"stop={port.StopBits} handshake={port.Handshake} dtr={port.DtrEnable} rts={port.RtsEnable}");
port.Open();

var parser = new FrameParser();
var writer = new SingleWriter(port, TimeSpan.FromSeconds(2));
var reader = new SerialReader(port, parser);

// 定义好的事件一定要订阅。忘了这一步，被丢掉的帧和应答都不会露面
parser.FrameDiscarded += frame => Console.Error.WriteLine($"crc error: {Convert.ToHexString(frame)}");
reader.FrameReceived += writer.OnFrameReceived;

using var cts = new CancellationTokenSource();
var readerTask = reader.RunAsync(cts.Token);
var writerTask = writer.RunAsync(cts.Token);

var request = new byte[] { 0x10, 0x00 };
try
{
    var response = await writer.SendAsync(request);
    Console.WriteLine($"response: {Convert.ToHexString(response)}");
}
finally
{
    // 即使发送失败，也一定要先停掉 worker 再退出。跳过这一步，
    // 在 using 释放掉 SerialPort 之后，reader/writer 仍会继续去碰
    // 那个流，而且抛出的异常没有任何人看得到
    cts.Cancel();
    try
    {
        await Task.WhenAll(readerTask, writerTask);
    }
    catch (OperationCanceledException)
    {
        // 因停止请求而结束。这里按正常路径处理
    }
    catch (Exception ex)
    {
        // worker 一侧的失败。在这里重新抛出会掩盖真正的失败原因
        //（SendAsync 的异常），所以只记录下来
        Console.Error.WriteLine($"worker stopped with error: {ex.Message}");
    }
}