序列通訊應用程式的陷阱 - 從重連到日誌設計

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 byte 以上會在轉型時繞回去。
        // 即使如此 payload 仍然會整份複製過去，於是接收端會照被截短的
        // 長度切訊框，把 payload 的中段讀成 CRC。之後的訊框邊界
        // 也會全部亂掉。要拆開送還是把 LEN 改成 2 byte 是協定的
        // 約定，這裡只負責擋掉
        if (payload.Length > byte.MaxValue)
        {
            throw new ArgumentOutOfRangeException(
                nameof(payload),
                $"一個訊框的 payload 最多 {byte.MaxValue} byte（因為 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>把收到的 byte 序列累積起來，只回傳成功切出的訊框。</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 byte 被 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}");
    }
}