From 94b10179cec545247b4d6d00d66a0607674300d4 Mon Sep 17 00:00:00 2001 From: Marci Date: Sat, 29 Aug 2026 02:50:30 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20=E5=86=85=E7=BD=AE=E6=8C=87?= =?UTF-8?q?=E6=A0=87=20+=20StxEtx=20=E5=90=91=E9=87=8F=E5=8C=96=20+=20?= =?UTF-8?q?=E6=B5=B8=E6=B3=A1=E5=A5=97=E4=BB=B6=20+=20epoch=20=E6=94=B6?= =?UTF-8?q?=E5=8F=A3=20+=20demo=20=E5=9C=BA=E6=99=AF=205?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 内置指标（Meter StreamFrame，System.Diagnostics.Metrics，零外部依赖）： 帧/字节收发计数、重连次数、会话时长直方图、发送队列水位（入队采样）， 单一 endpoint 标签；ns2.0 经 System.Diagnostics.DiagnosticSource 包获得 同款 API（netfx 可用）。README 双语新增"内置指标"章节，MeterListener 可观测性测试覆盖全部仪器 - StxEtxFramer 切帧向量化：net8+ 用 SearchValues 跳扫候选（两套独立实现， 语义逐字节等价），实测 ≈1.1µs/帧 → ≈50ns/帧（约 22 倍）；ns2.0 回退原实现。 README/bench 性能数据以 2026-08-28 重跑刷新（含指标接入后的真实路径： 端到端 ≈16万条/秒、RTT ≈51µs，机器相关） - 会话纪元改为 Connected 发布点分配（与会话编号同点）：闭合"发布后、任务 未创建"窗口内旧纪元幽灵卷入新生会话的瞬态误杀（评审附带发现）， 发布窗口注入幽灵的回归测试锁定 - 浸泡/混沌套件（STREAMFRAME_SOAK_SECONDS 环境变量门控，默认不跑）： 长流完整性（顺序/无丢失/无重复）+ 随机故障混沌（FIN/RST/半帧/突发， 终局无悬挂、普通消息续送核对、绑定消息无重复投递），20s/60s 双档验证 - demo 场景 5：会话感知收发全流程演示 - 覆盖率：分支 79.8% → 81.2%（worker 编号校验分支/XML 超限与坏帧/守卫） --- CHANGELOG.md | 9 + README.en.md | 26 +- README.md | 26 +- bench/README.md | 36 +- samples/StreamFrame.Demo/Program.cs | 123 +++++- src/StreamFrame/Connection/FrameDecoder.cs | 4 + .../Connection/StreamConnection.cs | 45 ++- .../Diagnostics/ConnectionMetrics.cs | 63 +++ src/StreamFrame/Framing/StxEtxFramer.cs | 43 ++ src/StreamFrame/StreamFrame.csproj | 2 + .../ConnectionMetricsTests.cs | 164 ++++++++ test/StreamFrame.Tests/FrameDecoderTests.cs | 4 + .../SessionAwareConnectionTests.cs | 57 +++ test/StreamFrame.Tests/SoakTests.cs | 380 ++++++++++++++++++ .../XmlDocumentCodecTests.cs | 20 + 15 files changed, 969 insertions(+), 33 deletions(-) create mode 100644 src/StreamFrame/Diagnostics/ConnectionMetrics.cs create mode 100644 test/StreamFrame.Tests/ConnectionMetricsTests.cs create mode 100644 test/StreamFrame.Tests/SoakTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 06a9d4f..98bd91d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,16 @@ ## [Unreleased] +### 新增 +- **内置指标（Meter `StreamFrame`，`System.Diagnostics.Metrics`，零外部依赖）**：帧/字节收发计数（`streamframe.frames_sent/received`、`bytes_sent/received`）、重连次数（`reconnects`）、会话时长（`session_duration` 直方图）、发送队列水位（`send_queue_length` 直方图，入队采样）；单一 `endpoint` 标签，订阅方式与仪器清单见 README"内置指标"。netstandard2.0 目标经 `System.Diagnostics.DiagnosticSource` 包获得同款 API（netfx 可用）。 +- **浸泡/混沌测试套件（默认不运行）**：`STREAMFRAME_SOAK_SECONDS=<秒> dotnet test --filter FullyQualifiedName~Soak` 手动启用——长时间消息流完整性（顺序/无丢失/无重复）与随机故障混沌（对端 FIN/RST 死亡 + 半帧注入 + 绑定/普通发送突发，终局核对无悬挂、普通消息跨会话续发送达、绑定消息无重复投递）。已在本机通过 20s/60s 两档验证。 + ### 变更 +- **`StxEtxFramer` 切帧向量化**：net8+ 用 `SearchValues` 一次跳到最近的 STX/ETX 候选（不含候选的整段字节直接跳过），实测 ≈1.1 µs/帧 → ≈50 ns/帧（约 22 倍）；netstandard2.0 目标保留逐字节实现，语义逐字节等价（两套独立实现，现有 StxEtx/FrameDecoder 测试全绿）。README/bench 性能数据以 2026-08-28 重跑刷新（含内置指标的真实路径）。 +- 会话纪元（epoch）改为在 Connected 发布点分配（与会话编号同点），彻底闭合"发布后、任务未创建"窗口内旧纪元幽灵故障卷入新生会话的瞬态误杀（v2.3.0 评审附带发现）；发布窗口注入幽灵的回归测试锁定该行为。 +- demo 新增场景 5：会话感知收发（编号生命周期、整帧写完才完成、旧会话发送失效不重放、新会话照常收发）。 + +### 修复 - 暂无 ## [2.3.1] - 2026-08-28 diff --git a/README.en.md b/README.en.md index 85d375d..62df38a 100644 --- a/README.en.md +++ b/README.en.md @@ -98,6 +98,7 @@ public interface ICodec - **Robustness**: payload decode failures, incomplete-frame overflows/timeouts, send failures, and receive idle timeouts all invalidate the session and rebuild it automatically (no more "connection looks alive while messages silently vanish") - **Liveness detection (optional)**: TCP KeepAlive and receive idle timeout to catch half-open connections (peer power loss / unplugged cable) - **Events**: `ConnectionChanged` state changes, `FrameError` frame-level diagnostics, `RawBytesReceived/Sent` raw bytes (HEX debugging) +- **Built-in metrics**: `System.Diagnostics.Metrics` (meter `StreamFrame`) — frame/byte counters, reconnects, session duration, send-queue watermark; see the "Built-in metrics" section - **Backpressure**: bounded send queue — `SendAsync` waits when full; the receive side buffers without limit by default, or set `ReceiveQueueCapacity` — decoding pauses when the consumer lags, propagating TCP backpressure to the peer and preventing unbounded memory growth - **Session-aware messaging (optional, advanced)**: `CurrentSessionId` / `SendInSessionAsync` (completes only after the whole frame is written to the socket; fails on session loss and never replays across sessions) / `GetSessionMessages` (messages carry their session id) — for protocols with strict session boundaries such as HSMS; see the dedicated section below @@ -168,6 +169,27 @@ A peer that announces a huge frame but never completes it (or a stream with STX The incomplete-frame timeout only counts while a frame is actually in progress: an idle connection with an empty buffer never trips it (each received byte resets it; it clears once a whole frame is cut). It complements `ReceiveIdleTimeoutMs` (see below) — the latter also times total silence, which suits protocols with periodic traffic; for protocols that may stay idle for a long time but must not tolerate a stalled half-frame (e.g. HSMS T8), use the incomplete-frame timeout. Note its scope is "waiting for further bytes from the network": when `ReceiveQueueCapacity` is set and the consumer stalls completely, the decode loop blocks on the message-channel write and the timeout does not tick during that period (the memory guard `MaxIncompleteFrameBufferBytes` still applies). +### Built-in metrics + +Connections ship [System.Diagnostics.Metrics](https://learn.microsoft.com/dotnet/core/diagnostics/metrics-instrumentation) instruments (meter name `StreamFrame`, tag `endpoint`) with zero external dependencies — subscribe with a `MeterListener` or OpenTelemetry in production; unsubscribed, each record costs nanoseconds: + +| Instrument | Kind | Meaning | +|---|---|---| +| `streamframe.frames_sent` / `frames_received` | Counter | Business frames sent/received | +| `streamframe.bytes_sent` / `bytes_received` | Counter | Bytes sent/received (framing bytes/noise included) | +| `streamframe.reconnects` | Counter | Reconnect count | +| `streamframe.session_duration` | Histogram | Lifetime of one TCP session (seconds) | +| `streamframe.send_queue_length` | Histogram | Send-queue watermark (sampled per enqueue) | + +```csharp +// Minimal subscription (example): OTel's MeterProvider.AddMeter("StreamFrame") hooks the full ecosystem +using var listener = new MeterListener { InstrumentPublished = (i, l) => l.EnableMeasurementEvents(i) }; +listener.SetMeasurementEventCallback((instrument, value, _, _) => { /* report */ }); +listener.Start(); +``` + +> The netstandard2.0 target gets the same API via the `System.Diagnostics.DiagnosticSource` package (works on .NET Framework). + ### Logging (optional) Pass an `ILogger` when constructing the connection to route internal events (connect retries, session faults, user-callback exceptions) to your logs — no more silence in production: @@ -206,8 +228,8 @@ Choosing between the two receive timeouts: `ReceiveIdleTimeoutMs` demands **peri Measured with BenchmarkDotNet (details and how to reproduce in [bench/README.md](bench/README.md)): - **Streaming encode**: halves per-frame heap allocation (single vs. double buffer); 25–35% faster for small payloads, on par for large ones; -- **Frame decoding**: `LengthPrefixFramer` ≈10 ns/frame vs `StxEtxFramer` ≈1.1 µs/frame (byte-by-byte scanning) — prefer length-prefix for high-throughput scenarios; -- **End-to-end** (real TCP loopback): one-way throughput ≈246k msgs/s (1KB messages, LengthPrefix), round-trip latency ≈63 µs; the XML codec costs 2–16 µs per message (400B–4KB) — serialization dominates, not framing. +- **Frame decoding**: `LengthPrefixFramer` ≈8.3 ns/frame, `StxEtxFramer` ≈50 ns/frame (≈22× faster after SearchValues vectorization on net8+) — both handle tens of millions of frames per second; +- **End-to-end** (real TCP loopback, re-measured 2026-08-28 with built-in metrics on): one-way throughput ≈160k msgs/s (1KB messages, LengthPrefix ≈6.2 µs/msg), round-trip latency ≈51 µs; the XML codec costs 2–16 µs per message (400B–4KB) — serialization dominates, not framing. Absolute values vary by machine; see [bench/README.md](bench/README.md) to reproduce. ## Supported frameworks diff --git a/README.md b/README.md index 810b599..7702cfc 100644 --- a/README.md +++ b/README.md @@ -98,6 +98,7 @@ public interface ICodec - **健壮性**：帧内容解码失败、未完成帧超限/超时、发送失败、接收空闲超时都会判定会话失效并自动重建（不再产生"连接看似存活、消息静默消失"的假活） - **活性探测（可选）**：TCP KeepAlive 与接收空闲超时，兜底半开连接（对端断电/拔线） - **事件**：`ConnectionChanged` 状态变化、`FrameError` 帧层诊断、`RawBytesReceived/Sent` 原始字节（HEX 调试） +- **内置指标**：`System.Diagnostics.Metrics`（Meter `StreamFrame`）——帧/字节收发计数、重连次数、会话时长、发送队列水位，详见下文"内置指标" - **发送背压**：有界发送队列，队列满时 `SendAsync` 自动等待；接收侧默认无上限缓冲，可用 `ReceiveQueueCapacity` 设上限——消费慢时解码暂停、TCP 背压自然传导到对端，防内存无限增长 - **会话感知收发（可选高级）**：`CurrentSessionId` / `SendInSessionAsync`（整帧写入 socket 才完成、会话失效即失败、绝不跨会话重放）/ `GetSessionMessages`（消息带会话编号）——为有严格会话边界的协议（如 HSMS）准备，详见下文 @@ -166,6 +167,27 @@ socket 层全量输出（含被丢弃的噪声字节），发送侧按实际写 未完成帧超时只计"帧已开头、迟迟收不齐"的时间：缓冲为空的静默连接不计时（收到字节即重置，整帧切尽后归零）。它与 `ReceiveIdleTimeoutMs`（见下节）互补——后者在完全没流量时也计时，适合有周期报文的协议；对允许长时间空闲、但半帧卡死必须判死的协议（如 HSMS T8），用未完成帧超时。注意其作用域是"等待网络后续字节"期间：若设置了 `ReceiveQueueCapacity` 且消费端完全停滞，解码循环阻塞在消息通道写入上，此期间不计时（内存防线仍由 `MaxIncompleteFrameBufferBytes` 兜底）。 +### 内置指标（Metrics） + +连接自带 [System.Diagnostics.Metrics](https://learn.microsoft.com/dotnet/core/diagnostics/metrics-instrumentation) 指标（Meter 名 `StreamFrame`，标签 `endpoint`），零外部依赖——生产部署用 `MeterListener` 或 OpenTelemetry 订阅即可观测，不订阅则开销为每次记录纳秒级： + +| 指标 | 类型 | 含义 | +|---|---|---| +| `streamframe.frames_sent` / `frames_received` | Counter | 业务帧收发计数 | +| `streamframe.bytes_sent` / `bytes_received` | Counter | 字节收发计数（含帧定界字节/噪声） | +| `streamframe.reconnects` | Counter | 进入重连的次数 | +| `streamframe.session_duration` | Histogram | 单次 TCP 会话存活时长（秒） | +| `streamframe.send_queue_length` | Histogram | 发送队列水位（每次入队采样） | + +```csharp +// 最简订阅（示例）：OTel 的 MeterProvider.AddMeter("StreamFrame") 一行即可接入完整体系 +using var listener = new MeterListener { InstrumentPublished = (i, l) => l.EnableMeasurementEvents(i) }; +listener.SetMeasurementEventCallback((instrument, value, _, _) => { /* 上报 */ }); +listener.Start(); +``` + +> netstandard2.0 目标经 `System.Diagnostics.DiagnosticSource` 包提供同款 API（netfx 运行时可用）。 + ### 日志（可选） 构造连接时传入 `ILogger`，内部事件（连接重试、会话故障、用户回调异常等）输出到日志，生产环境不再静默： @@ -203,8 +225,8 @@ _ = Task.Run(async () => BenchmarkDotNet 实测（详见 [bench/README.md](bench/README.md)，可本地复现）： - **流式编码**：每帧堆分配减半（单缓冲 vs 双缓冲），小负载耗时快 25–35%，大负载持平； -- **切帧吞吐**：`LengthPrefixFramer` ≈10 ns/帧，`StxEtxFramer` ≈1.1 µs/帧（逐字节扫描）——高吞吐场景优先长度前缀； -- **端到端**（真实 TCP 回环）：单向吞吐 ≈24 万条/秒（1KB 消息，LengthPrefix），往返延迟 ≈63 µs；XML codec 每条报文 2–16 µs（400B–4KB）——序列化开销远大于定界层。 +- **切帧吞吐**：`LengthPrefixFramer` ≈8.3 ns/帧，`StxEtxFramer` ≈50 ns/帧（net8+ SearchValues 向量化后较逐字节版提速约 22 倍）——两者均可达每秒千万帧级； +- **端到端**（真实 TCP 回环，2026-08-28 重测，含内置指标）：单向吞吐 ≈16 万条/秒（1KB 消息，LengthPrefix ≈6.2 µs/条），往返延迟 ≈51 µs；XML codec 每条报文 2–16 µs（400B–4KB）——序列化开销远大于定界层。绝对值随机器浮动，复现方式见 [bench/README.md](bench/README.md)。 ## 支持框架 diff --git a/bench/README.md b/bench/README.md index 394be28..e6aaa7a 100644 --- a/bench/README.md +++ b/bench/README.md @@ -18,24 +18,24 @@ dotnet run -c Release --project bench/StreamFrame.Benchmarks -- --filter "*EndTo - **Codec 基准**（`CodecBenchmarks`）：官方 XML 驱动 `XmlDocumentCodec` 对典型设备报文（~400B 单值上报 / ~4KB 批量明细）的编解码开销。 - **端到端基准**（`EndToEndBenchmarks`）：真实 TCP 回环上的完整连接（收发队列 + Pipe + 定界 + codec + socket）——单向吞吐（连发 1 万条 1KB 消息，双 framer）与往返延迟（逐条乒乓 2000 次）。 -## 结果 +## 结果（2026-08-28 重跑：StxEtx 向量化 + 内置指标已接入；Codec 表为历史数据） | 方法 | 负载 | 耗时 | 分配 | |---|---|---:|---:| -| LengthPrefix_Plain | 64B | 36.3 ns | 64 B | -| LengthPrefix_Streaming | 64B | **27.3 ns（-25%）** | **32 B（-50%）** | -| StxEtx_Plain | 64B | 36.7 ns | 32 B | -| StxEtx_Streaming | 64B | **24.4 ns（-34%）** | 32 B | -| LengthPrefix_Plain | 1KB | 57.6 ns | 64 B | -| LengthPrefix_Streaming | 1KB | 67.3 ns（+17%）¹ | **32 B（-50%）** | -| StxEtx_Plain | 1KB | 77.0 ns | 32 B | -| StxEtx_Streaming | 1KB | 82.6 ns（+7%）¹ | 32 B | -| LengthPrefix_Plain | 64KB | 4.11 µs | 64 B | -| LengthPrefix_Streaming | 64KB | 4.11 µs（持平） | **32 B（-50%）** | -| StxEtx_Plain | 64KB | 6.38 µs | 32 B | -| StxEtx_Streaming | 64KB | 6.04 µs（-5%） | 32 B | -| LengthPrefix 切帧 100×1KB | — | **1.0 µs**（≈10 ns/帧） | 0 | -| StxEtx 切帧 100×1KB | — | **≈110 µs**（≈1.1 µs/帧） | 0 | +| LengthPrefix_Plain | 64B | 28.1 ns | 64 B | +| LengthPrefix_Streaming | 64B | **21.4 ns（-24%）** | **32 B（-50%）** | +| StxEtx_Plain | 64B | 28.6 ns | 32 B | +| StxEtx_Streaming | 64B | **19.1 ns（-33%）** | 32 B | +| LengthPrefix_Plain | 1KB | 45.5 ns | 64 B | +| LengthPrefix_Streaming | 1KB | 56.3 ns（+24%）¹ | **32 B（-50%）** | +| StxEtx_Plain | 1KB | 60.9 ns | 32 B | +| StxEtx_Streaming | 1KB | 68.3 ns（+12%）¹ | 32 B | +| LengthPrefix_Plain | 64KB | 3.44 µs | 64 B | +| LengthPrefix_Streaming | 64KB | 3.47 µs（持平） | **32 B（-50%）** | +| StxEtx_Plain | 64KB | 4.51 µs | 32 B | +| StxEtx_Streaming | 64KB | 4.64 µs（+3%） | 32 B | +| LengthPrefix 切帧 100×1KB | — | **0.82 µs**（≈8.3 ns/帧） | 0 | +| StxEtx 切帧 100×1KB | — | **≈5.0 µs**（≈50 ns/帧，向量化后提速约 22 倍） | 0 | ¹ 恰好压在 `EncodeBufferInitialSize`（默认 1024）的缓冲增长边界上：流式路径触发一次扩容租借， 抵消了省下的那次拷贝。负载远大于初始缓冲后两条路径的拷贝量趋同（几何扩容摊销）。 @@ -53,8 +53,8 @@ dotnet run -c Release --project bench/StreamFrame.Benchmarks -- --filter "*EndTo | 指标 | LengthPrefix | StxEtx | |---|---:|---:| -| 单向吞吐（每消息折算） | 4.06 µs ≈ **24.6 万条/秒** | 4.65 µs ≈ 21.5 万条/秒 | -| 往返延迟（乒乓串行） | 62.8 µs | 71.1 µs | +| 单向吞吐（每消息折算） | 6.21 µs ≈ **16.1 万条/秒** | 6.05 µs ≈ 16.5 万条/秒 | +| 往返延迟（乒乓串行） | 50.8 µs | 52.4 µs | | 每消息堆分配（吞吐模式） | 6.5 KB | 6.9 KB | 测试环境：Windows（本地开发机），.NET 8，BenchmarkDotNet 0.15.8，SimpleJob(warmup 3 / iteration 15；e2e 为 2/10)。 @@ -65,4 +65,4 @@ dotnet run -c Release --project bench/StreamFrame.Benchmarks -- --filter "*EndTo - **流式编码的实测收益主要是"每帧堆分配减半"（一个缓冲对象 vs 两个）**，小负载（≤ 百字节）耗时也快 25–35%； 大负载下省掉的那次 memcpy 被缓冲增长摊销，耗时基本持平。"零拷贝"指的是帧路径不再整体搬运负载，不是字面意义的零成本。 - **切帧吞吐相差约两个数量级**：`LengthPrefixFramer` 读长度头直接跳转（≈10 ns/帧）； - `StxEtxFramer` 逐字节扫描定位边界（≈1.1 µs/帧）。高吞吐/大帧场景优先长度前缀，STX/ETX 留给既有协议兼容。 + `StxEtxFramer` 在 net8+ 用 `SearchValues` 向量化定位边界（≈50 ns/帧；netstandard2.0 目标回退逐字节实现）。长度前缀在超大帧上仍占优，STX/ETX 适合既有协议兼容。 diff --git a/samples/StreamFrame.Demo/Program.cs b/samples/StreamFrame.Demo/Program.cs index 66b494a..3d6b1ab 100644 --- a/samples/StreamFrame.Demo/Program.cs +++ b/samples/StreamFrame.Demo/Program.cs @@ -1,15 +1,17 @@ using System.Buffers; using System.Net; +using System.Net.Sockets; using System.Text; using System.Xml.Linq; using StreamFrame; using StreamFrame.Protocols.Xml; -// 本 demo 演示 StreamFrame 的四种典型用法： +// 本 demo 演示 StreamFrame 的五种典型用法： // 1) XML 消息 + 4 字节长度前缀帧（对应 SamSung 风格） // 2) 纯文本消息 + STX/ETX 包裹帧 // 3) 断线自动重连 // 4) 心跳保活 + 接收空闲超时（半开连接探测） +// 5) 会话感知收发（整帧写完才完成、旧会话消息不重放） // 所有场景共用同一套连接核心，仅 framing / codec / 选项不同。 var cts = new CancellationTokenSource(); @@ -24,6 +26,7 @@ await RunStxEtxTextScenarioAsync(cts.Token); await RunReconnectScenarioAsync(cts.Token); await RunHeartbeatScenarioAsync(cts.Token); +await RunSessionAwareScenarioAsync(cts.Token); Console.WriteLine("演示结束。"); @@ -397,6 +400,124 @@ static void Assert(bool condition, string message) } } +// --------------------------------------------------------------------------- +// 场景 5：会话感知收发（ISessionAwareStreamConnection） +// --------------------------------------------------------------------------- +static async Task RunSessionAwareScenarioAsync(CancellationToken ct) +{ + const int port = 5500; + Console.WriteLine("\n=== 场景 5：会话感知收发（重连后旧会话消息不重放） ==="); + + var server = new StreamConnection( + new LengthPrefixFramer(), + new Utf8TextCodec(), + IPAddress.Loopback, + port, + isActive: false); + server.ConnectionChanged += (_, state) => + Console.WriteLine($"[server] 状态 -> {state}（会话编号 {server.CurrentSessionId}）"); + server.Start(ct); + + // 设备 1 接入：Connected 可见时即可读到有效的会话编号 + using (var device1 = new TcpClient()) + { + await device1.ConnectAsync(IPAddress.Loopback, port, ct); + await WaitForServerStateAsync(server, s => s == ConnectionState.Connected, ct); + var session1 = server.CurrentSessionId; + Console.WriteLine($"[server] 设备 1 接入，会话编号 = {session1}"); + Assert(session1 != 0, "Connected 后会话编号必须有效（非 0）"); + + // 会话绑定发送：任务在整帧写入 socket 后才完成（对端此刻才能读到完整帧） + var sendTask = server.SendInSessionAsync(session1, "hello-device1", ct); + var received = await ReadDemoFrameAsync(device1.GetStream(), ct); + await sendTask.WaitAsync(TimeSpan.FromSeconds(5)); + Console.WriteLine($"[device1] 收到完整帧: {received}"); + Assert(received == "hello-device1", "设备 1 应收到会话绑定消息"); + + // 设备 1 掉线（服务端将自动重连进入监听） + Console.WriteLine("[device1] 掉线（dispose）…"); + } + + await WaitForServerStateAsync(server, s => s != ConnectionState.Connected, ct); + + // 旧会话的绑定发送：立即以 SessionExpiredException 失败——不等待重连、绝不重放到新会话 + var expired = await AssertDemoThrowsAsync( + () => server.SendInSessionAsync(server.CurrentSessionId == 0 ? 1 : server.CurrentSessionId, "stale-message", ct)); + Console.WriteLine($"[server] 旧会话发送按预期失败：会话 {expired.SessionId} 已失效，消息不会转移到新会话"); + + // 设备 2 接入：会话编号单调递增，新会话照常收发 + using (var device2 = new TcpClient()) + { + await device2.ConnectAsync(IPAddress.Loopback, port, ct); + await WaitForServerStateAsync(server, s => s == ConnectionState.Connected, ct); + var session2 = server.CurrentSessionId; + Console.WriteLine($"[server] 设备 2 接入，会话编号 = {session2}"); + Assert(session2 > 1, "新会话编号必须大于旧会话"); + + var sendTask = server.SendInSessionAsync(session2, "hello-device2", ct); + var received = await ReadDemoFrameAsync(device2.GetStream(), ct); + await sendTask.WaitAsync(TimeSpan.FromSeconds(5)); + Console.WriteLine($"[device2] 收到完整帧: {received}"); + Assert(received == "hello-device2", "设备 2 应收到新会话消息"); + } + + await server.DisposeAsync(); + Console.WriteLine("场景 5 通过 ✓"); +} + +/// 轮询等待连接状态（demo 用，5 秒超时即断言失败）。 +static async Task WaitForServerStateAsync( + StreamConnection server, Func predicate, CancellationToken ct) +{ + var deadline = DateTime.UtcNow.AddSeconds(5); + while (DateTime.UtcNow < deadline) + { + if (predicate(server.State)) + return; + await Task.Delay(50, ct); + } + + Assert(predicate(server.State), $"等待状态超时，当前 {server.State}。"); +} + +/// 从裸 TcpClient 读一个"4 字节大端长度 + UTF-8"帧。 +static async Task ReadDemoFrameAsync(NetworkStream stream, CancellationToken ct) +{ + var header = new byte[4]; + await ReadDemoExactlyAsync(stream, header, ct); + var length = System.Buffers.Binary.BinaryPrimitives.ReadInt32BigEndian(header); + var body = new byte[length]; + await ReadDemoExactlyAsync(stream, body, ct); + return Encoding.UTF8.GetString(body); +} + +static async Task ReadDemoExactlyAsync(NetworkStream stream, byte[] buffer, CancellationToken ct) +{ + var read = 0; + while (read < buffer.Length) + { + var n = await stream.ReadAsync(buffer, read, buffer.Length - read, ct); + Assert(n > 0, "对端提前关闭。"); + read += n; + } +} + +static async Task AssertDemoThrowsAsync(Func action) + where TException : Exception +{ + try + { + await action(); + } + catch (TException ex) + { + return ex; + } + + Assert(false, $"应抛出 {typeof(TException).Name}。"); + throw new InvalidOperationException("unreachable"); +} + /// 简单的 UTF-8 文本 codec（演示自定义 codec 有多么容易）。 internal sealed class Utf8TextCodec : ICodec { diff --git a/src/StreamFrame/Connection/FrameDecoder.cs b/src/StreamFrame/Connection/FrameDecoder.cs index e64b303..b19a28e 100644 --- a/src/StreamFrame/Connection/FrameDecoder.cs +++ b/src/StreamFrame/Connection/FrameDecoder.cs @@ -27,6 +27,7 @@ internal sealed class FrameDecoder private readonly IFramer _framing; private readonly ICodec _codec; private readonly Channel> _relay; + private readonly ConnectionMetrics _metrics; private readonly long _sessionId; private readonly int _maxIncompleteFrameBytes; private readonly int _incompleteFrameTimeoutMs; @@ -38,6 +39,7 @@ public FrameDecoder( IFramer framing, ICodec codec, Channel> relay, + ConnectionMetrics metrics, long sessionId, int maxIncompleteFrameBytes, int incompleteFrameTimeoutMs, @@ -48,6 +50,7 @@ public FrameDecoder( _framing = framing; _codec = codec; _relay = relay; + _metrics = metrics; _sessionId = sessionId; _maxIncompleteFrameBytes = maxIncompleteFrameBytes; _incompleteFrameTimeoutMs = incompleteFrameTimeoutMs; @@ -109,6 +112,7 @@ public async Task RunAsync(CancellationToken ct) { await _relay.Writer.WriteAsync( new SessionMessage(_sessionId, message), CancellationToken.None).ConfigureAwait(false); + _metrics.FrameReceived(); } catch (ChannelClosedException) { diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs index cb4592c..8e68e02 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -1,5 +1,6 @@ using System.Buffers; using System.Collections.Concurrent; +using System.Diagnostics; using System.IO.Pipelines; using System.Net; using System.Net.Sockets; @@ -90,11 +91,15 @@ public string? RemoteIpAddress /// 公共会话编号的分配计数器：仅在真实 TCP 会话建立时递增（有间隔、单调、不复用）；checked 防静默回绕。 private long _sessionCounter; + /// 当前会话的建立时刻（Stopwatch 时戳），拆除时用于会话时长指标。 + private long _sessionStartedAt; + /// 当前会话编号：0 = 无会话。分配/归零都必须先于对应状态对外发布（线性化点，见 CommunicationStateChanging）。 private long _currentSessionId; private readonly Channel _sendQueue; private readonly Channel> _messageRelay; + private readonly ConnectionMetrics _metrics; /// 已注册、尚未终结的会话绑定发送条目：会话拆除时立即 fault，避免调用方空等重连全程。 private readonly ConcurrentDictionary _pendingSessionSends = new(); @@ -171,6 +176,8 @@ public StreamConnection( _retryScheduler = new RetryDelayScheduler( isActive ? _options.ConnectRetryDelayMs : _options.AcceptRetryDelayMs, _options.MaxRetryDelayMs); + + _metrics = new ConnectionMetrics($"{IpAddress}:{Port}"); } /// 连接是否已停机（Dispose 或 Start 令牌取消）。停机后不可再用，需新建连接。 @@ -308,7 +315,7 @@ private async Task AcceptAsync(CancellationToken ct) catch (Exception ex) when (!ct.IsCancellationRequested && !IsDisposed) { var delay = _retryScheduler.NextDelayMs(); - _logger.LogWarning(ex, "Accept on {Local} failed (attempt {Attempt}); retry in {Delay}ms.", + _logger.LogWarning(ex, "Accept on {Local} failed (attempt {Attempt}); retry in {Delay}ms.", $"{IpAddress}:{Port}", _retryScheduler.Attempt, delay); await Task.Delay(delay, ct).ConfigureAwait(false); } @@ -419,16 +426,19 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE return; // 公共会话编号的线性化点：Connected 对外发布（事件回调/等待器完成）之前完成分配； - // 离开 Connected 的状态（Retry/Disconnected）对外发布之前归零——状态可见时旧会话 - // 已不可用于新的会话绑定发送（SendInSessionAsync 以编号不匹配快速失败）。 + // 会话纪元（epoch）同样在发布之前递增——发布后到达的旧纪元幽灵故障在函数入口 + // 即被判为过期丢弃，不会卷进"已发布、任务未创建"的新生会话（评审附带发现收口）。 if (newState == ConnectionState.Connected) { var id = checked(Interlocked.Increment(ref _sessionCounter)); Volatile.Write(ref _currentSessionId, id); + Interlocked.Increment(ref _sessionEpoch); } else if (newState is ConnectionState.Retry or ConnectionState.Disconnected) { Volatile.Write(ref _currentSessionId, 0); + if (newState == ConnectionState.Retry) + _metrics.Reconnect(); // 过期幽灵已在函数入口被丢弃，不会计入 } State = newState; @@ -452,7 +462,7 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE switch (newState) { case ConnectionState.Connected: - StartSession(Volatile.Read(ref _currentSessionId)); + StartSession(Volatile.Read(ref _currentSessionId), Volatile.Read(ref _sessionEpoch)); break; case ConnectionState.Retry: @@ -485,34 +495,36 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE } } - private void StartSession(long sessionId) + private void StartSession(long sessionId, int epoch) { lock (_sessionGate) { StopSessionCore(); - // 病态重入兜底：Connected 回调里同步触发 Reconnect() 会让编号在 StartSession - // 执行前被归零——此时重新分配一个，保证会话任务拿到的编号恒非零。 + // 病态重入兜底：Connected 回调里同步触发 Reconnect() 会让编号/纪元在 StartSession + // 执行前被归零/递增——此时重新分配两者，保证会话任务拿到的编号恒非零、 + // 纪元为最新（旧纪元的任务故障会被入口的过期检查丢弃）。 if (sessionId == 0) { sessionId = checked(Interlocked.Increment(ref _sessionCounter)); Volatile.Write(ref _currentSessionId, sessionId); + epoch = Interlocked.Increment(ref _sessionEpoch); } - var epoch = ++_sessionEpoch; var cts = new CancellationTokenSource(); var token = cts.Token; var pipe = new Pipe(); _sessionCts = cts; _pipe = pipe; + _sessionStartedAt = Stopwatch.GetTimestamp(); var maxIncompleteFrameBytes = _options.MaxIncompleteFrameBufferBytes > 0 ? _options.MaxIncompleteFrameBufferBytes : _framing.MaxPayloadBytes + 4096; var decoder = new FrameDecoder( - pipe.Reader, _framing, _codec, _messageRelay, sessionId, + pipe.Reader, _framing, _codec, _messageRelay, _metrics, sessionId, maxIncompleteFrameBytes, _options.IncompleteFrameTimeoutMs, _options.DecodeErrorPolicy, RaiseFrameError); @@ -571,6 +583,11 @@ private void StopSessionCore() if (_sessionCts is not { } cts) return; + // 会话时长指标：从建立到拆除的真实存活时间 + var startedAt = Interlocked.Exchange(ref _sessionStartedAt, 0); + if (startedAt != 0) + _metrics.SessionEnded((Stopwatch.GetTimestamp() - startedAt) / (double)Stopwatch.Frequency); + var pipe = _pipe; if (!cts.IsCancellationRequested) @@ -638,6 +655,7 @@ private async Task ReceiveLoopAsync(PipeWriter writer, CancellationToken ct, int if (count > 0) { InvokeRawBytesReceived(memory[..count]); + _metrics.AddBytesReceived(count); await writer.FlushAsync(ct).ConfigureAwait(false); } else @@ -693,6 +711,7 @@ private async Task SendWorkerAsync(CancellationToken ct, long sessionId) if (!entry.IsSessionBound) { await SendFramedAsync(entry.Message, ct).ConfigureAwait(false); + _metrics.FrameSent(); continue; } @@ -715,6 +734,7 @@ private async Task SendWorkerAsync(CancellationToken ct, long sessionId) { await SendFramedAsync(entry.Message, ct).ConfigureAwait(false); entry.Complete(); // 整帧已写入本机 socket，任务成功完成 + _metrics.FrameSent(); } catch (OperationCanceledException ex) when (ct.IsCancellationRequested) { @@ -772,6 +792,7 @@ private async Task SendRawAsync(ReadOnlyMemory buffer, CancellationToken c // 按实际写出的分片回调：部分发送失败时，已上线字节也可见 InvokeRawBytesSent(buffer[..length]); + _metrics.AddBytesSent(length); buffer = buffer[length..]; } } @@ -820,7 +841,10 @@ private void RaiseFrameError(FrameErrorEventArgs args) /// public Task SendAsync(TMessage message, CancellationToken ct = default) - => _sendQueue.Writer.WriteAsync(new SessionSendEntry(message), ct).AsTask(); + { + _metrics.SendQueueObserved(_sendQueue.Reader.Count); + return _sendQueue.Writer.WriteAsync(new SessionSendEntry(message), ct).AsTask(); + } /// public async Task SendInSessionAsync(long sessionId, TMessage message, CancellationToken ct = default) @@ -832,6 +856,7 @@ public async Task SendInSessionAsync(long sessionId, TMessage message, Cancellat throw NewSessionExpired(sessionId); var entry = new SessionSendEntry(sessionId, message); + _metrics.SendQueueObserved(_sendQueue.Reader.Count); _pendingSessionSends.TryAdd(entry, 0); try { diff --git a/src/StreamFrame/Diagnostics/ConnectionMetrics.cs b/src/StreamFrame/Diagnostics/ConnectionMetrics.cs new file mode 100644 index 0000000..ebc6c34 --- /dev/null +++ b/src/StreamFrame/Diagnostics/ConnectionMetrics.cs @@ -0,0 +1,63 @@ +using System.Diagnostics.Metrics; + +namespace StreamFrame; + +/// +/// 连接级内置指标（ 名 "StreamFrame"，零外部依赖）： +/// 帧收发计数、字节收发计数、重连次数、会话时长、发送队列水位。生产部署用 +/// MeterListener / OpenTelemetry 订阅 "StreamFrame" 即可观测，不接入则开销为每次记录 +/// 数纳秒级；netstandard2.0 目标为 no-op 兜底（无 Metrics API，记录全部空操作）。 +/// +/// 标签：单一 "endpoint"（构造时的 ip:port）——每条连接的地址固定，标签基数受控。 +/// +internal sealed class ConnectionMetrics +{ + private static readonly Meter Meter = new("StreamFrame"); + + private static readonly Counter FramesSent = Meter.CreateCounter( + "streamframe.frames_sent", unit: "frames", description: "整帧写入 socket 的业务帧数（含普通与会话绑定发送）。"); + + private static readonly Counter FramesReceived = Meter.CreateCounter( + "streamframe.frames_received", unit: "frames", description: "解码切出的业务帧数。"); + + private static readonly Counter BytesSent = Meter.CreateCounter( + "streamframe.bytes_sent", unit: "bytes", description: "写入 socket 的字节总数（含帧定界字节）。"); + + private static readonly Counter BytesReceived = Meter.CreateCounter( + "streamframe.bytes_received", unit: "bytes", description: "从 socket 收到的字节总数（含噪声与未完成帧）。"); + + private static readonly Counter Reconnects = Meter.CreateCounter( + "streamframe.reconnects", unit: "reconnects", description: "进入重连的次数（故障触发与用户显式 Reconnect 均计数）。"); + + private static readonly Histogram SessionDuration = Meter.CreateHistogram( + "streamframe.session_duration", unit: "s", description: "单次 TCP 会话从建立到拆除的时长。"); + + private static readonly Histogram SendQueueLength = Meter.CreateHistogram( + "streamframe.send_queue_length", unit: "items", description: "发送队列水位（每次入队时采样）。"); + + private readonly KeyValuePair _tag; + + public ConnectionMetrics(string endpoint) + => _tag = new("endpoint", endpoint); + + public void FrameSent() + => FramesSent.Add(1, _tag); + + public void FrameReceived() + => FramesReceived.Add(1, _tag); + + public void AddBytesSent(long count) + => BytesSent.Add(count, _tag); + + public void AddBytesReceived(long count) + => BytesReceived.Add(count, _tag); + + public void Reconnect() + => Reconnects.Add(1, _tag); + + public void SessionEnded(double seconds) + => SessionDuration.Record(seconds, _tag); + + public void SendQueueObserved(int count) + => SendQueueLength.Record(count, _tag); +} diff --git a/src/StreamFrame/Framing/StxEtxFramer.cs b/src/StreamFrame/Framing/StxEtxFramer.cs index d9910ff..8a48b34 100644 --- a/src/StreamFrame/Framing/StxEtxFramer.cs +++ b/src/StreamFrame/Framing/StxEtxFramer.cs @@ -22,6 +22,12 @@ public sealed class StxEtxFramer : IStreamingFramer, IFrameDiscardReporting private const byte STX = 0x02; private const byte ETX = 0x03; +#if !NETSTANDARD2_0 + /// 向量化候选集：解码扫描时一次跳到最近的 STX/ETX（net8+ 的 SearchValues）。 + private static readonly System.Buffers.SearchValues StxEtxSearchValues = + System.Buffers.SearchValues.Create(new byte[] { STX, ETX }); +#endif + /// 默认负载上限（16 MiB）。 public const int DefaultMaxPayloadBytes = 16 * 1024 * 1024; @@ -114,6 +120,42 @@ public bool TryDecodeFrame(ref ReadOnlySequence buffer, out ReadOnlySequen long lastStxOffset = -1; long lastStxPayloadStart = -1; +#if !NETSTANDARD2_0 + // 向量化扫描（net8+）：SearchValues 让 IndexOfAny 一次跳到最近的 STX/ETX 候选， + // 不含候选的整段字节直接跳过（旧逐字节实现约 1.1µs/帧）。语义与逐字节版完全一致。 + var i = 0; + while (i < span.Length) + { + var hit = span.Slice(i).IndexOfAny(StxEtxSearchValues); + if (hit < 0) + break; + i += hit; + + if (span[i] == STX) + { + lastStxOffset = i; // STX 位置 + lastStxPayloadStart = i + 1; // 负载起始位置 + } + else if (lastStxPayloadStart >= 0) // ETX：有已开启的帧才可能闭合 + { + var payloadLength = i - lastStxPayloadStart; + if (payloadLength <= MaxPayloadBytes) + { + payload = buffer.Slice(lastStxPayloadStart, payloadLength); + buffer = buffer.Slice(i + 1); + discarded = original.Slice(0, lastStxOffset); // 保留点之前的噪声/被中止半帧 + return true; + } + + // 超长帧：丢弃当前部分帧并继续扫描 + lastStxPayloadStart = -1; + lastStxOffset = -1; + } + // 孤立 ETX（无已开启的帧）：视为噪声字节跳过 + + i++; + } +#else for (var i = 0; i < span.Length; i++) { var b = span[i]; @@ -142,6 +184,7 @@ public bool TryDecodeFrame(ref ReadOnlySequence buffer, out ReadOnlySequen return true; } } +#endif // 缓冲区耗尽但未收到完整帧：保留自最后一个 STX 起的余量等待后续数据。 // 保留点之前的字节（杂散噪声、被更新的 STX 中止的旧半帧）即为本次丢弃。 diff --git a/src/StreamFrame/StreamFrame.csproj b/src/StreamFrame/StreamFrame.csproj index f070d03..c72073b 100644 --- a/src/StreamFrame/StreamFrame.csproj +++ b/src/StreamFrame/StreamFrame.csproj @@ -45,6 +45,8 @@ + + diff --git a/test/StreamFrame.Tests/ConnectionMetricsTests.cs b/test/StreamFrame.Tests/ConnectionMetricsTests.cs new file mode 100644 index 0000000..e387e12 --- /dev/null +++ b/test/StreamFrame.Tests/ConnectionMetricsTests.cs @@ -0,0 +1,164 @@ +using System.Collections.Concurrent; +using System.Diagnostics.Metrics; +using System.Net; +using System.Net.Sockets; +using System.Text; +using StreamFrame; + +namespace StreamFrame.Tests; + +/// +/// 内置指标（Meter "StreamFrame"）的可观测性验证：用 MeterListener 直接对既有 instrument +/// 启用测量（InstrumentPublished 只对 Start 之后新发布的仪器触发，进程内仪器可能早已创建）， +/// 驱动一次完整的连接生命周期，断言各仪器都有真实记录。 +/// +public class ConnectionMetricsTests +{ + private static int GetFreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + private static async Task ConnectWithRetryAsync(TcpClient client, int port) + { + for (var attempt = 0; ; attempt++) + { + try + { + await client.ConnectAsync(IPAddress.Loopback, port); + return; + } + catch (SocketException) when (attempt < 10) + { + await Task.Delay(200); + } + } + } + + private static async Task WaitForStateAsync( + StreamConnection connection, + Func predicate, + int timeoutMs = 8000) + { + var deadline = TestClock.TickCount64 + timeoutMs; + while (TestClock.TickCount64 < deadline) + { + if (predicate(connection.State)) + return; + await Task.Delay(20); + } + + Assert.True(predicate(connection.State), $"等待连接状态超时，当前 {connection.State}。"); + } + + [Fact] + public async Task ConnectionLifecycle_ProducesAllInstruments() + { + var recorded = new ConcurrentDictionary(); + using var listener = new MeterListener(); + listener.SetMeasurementEventCallback( + (instrument, value, _, _) => recorded.AddOrUpdate(instrument.Name, value, (_, sum) => sum + value)); + listener.SetMeasurementEventCallback( + (instrument, value, _, _) => recorded.AddOrUpdate(instrument.Name, value, (_, sum) => sum + value)); + listener.Start(); + + // 进程内静态仪器可能早于本监听器创建：逐个直接启用 + foreach (var field in typeof(ConnectionMetrics).GetFields( + System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Static)) + { + if (field.GetValue(null) is Instrument instrument) + listener.EnableMeasurementEvents(instrument); + } + + var port = GetFreePort(); + await using var server = new StreamConnection( + new LengthPrefixFramer(), StringCodec.Instance, IPAddress.Loopback, port, isActive: false); + var received = new List(); + var drainCts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); + _ = Task.Run(async () => + { + try + { + await foreach (var m in server.GetMessages(drainCts.Token)) + lock (received) + received.Add(m); + } + catch (OperationCanceledException) + { + } + }); + server.Start(CancellationToken.None); + + long id1; + using (var client1 = new TcpClient()) + { + await ConnectWithRetryAsync(client1, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + id1 = server.CurrentSessionId; + + var payload = new string('M', 1024); + var sendTask = server.SendInSessionAsync(id1, payload); // 普通方向：对端发来一帧 + var header = new byte[4]; + System.Buffers.Binary.BinaryPrimitives.WriteInt32BigEndian(header, payload.Length); + await client1.GetStream().WriteAsync(header); + await client1.GetStream().WriteAsync(Encoding.UTF8.GetBytes(payload)); + await sendTask.WaitAsync(TimeSpan.FromSeconds(10)); + + await client1.GetStream().WriteAsync(new byte[] { 0x00, 0x00, 0x00, 0x03, (byte)'i', (byte)'n', 0x21 }); + var deadline = TestClock.TickCount64 + 10_000; + while (TestClock.TickCount64 < deadline) + { + lock (received) + { + if (received.Count == 1) + break; + } + await Task.Delay(20); + } + lock (received) + Assert.Single(received); + + client1.Dispose(); // 触发重连 + } + + await WaitForStateAsync(server, s => s != ConnectionState.Connected); + using (var client2 = new TcpClient()) + { + await ConnectWithRetryAsync(client2, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + } + + await server.DisposeAsync(); + + // MeterListener 的测量回调经内部队列异步派发：对全部预期键做带截止时间的轮询， + // 而不是固定延迟（负载下排空时间不定） + string[] keys = + { + "streamframe.frames_sent", "streamframe.frames_received", "streamframe.bytes_sent", + "streamframe.bytes_received", "streamframe.reconnects", "streamframe.session_duration", + "streamframe.send_queue_length", + }; + var drainDeadline = TestClock.TickCount64 + 10_000; + while (TestClock.TickCount64 < drainDeadline && !keys.All(k => recorded.ContainsKey(k))) + await Task.Delay(50); + + Assert.True(recorded.TryGetValue("streamframe.frames_sent", out var framesSent) && framesSent >= 1, + $"frames_sent 应有记录，实际 {framesSent}"); + Assert.True(recorded.TryGetValue("streamframe.frames_received", out var framesReceived) && framesReceived >= 1, + $"frames_received 应有记录，实际 {framesReceived}"); + Assert.True(recorded.TryGetValue("streamframe.bytes_sent", out var bytesSent) && bytesSent >= 1028, + $"bytes_sent 应含整帧字节，实际 {bytesSent}"); + Assert.True(recorded.TryGetValue("streamframe.bytes_received", out var bytesReceived) && bytesReceived >= 7, + $"bytes_received 应有记录，实际 {bytesReceived}"); + Assert.True(recorded.TryGetValue("streamframe.reconnects", out var reconnects) && reconnects >= 1, + $"reconnects 应有记录，实际 {reconnects}"); + Assert.True(recorded.TryGetValue("streamframe.session_duration", out var duration) && duration > 0, + $"session_duration 应有记录，实际 {duration}"); + Assert.True(recorded.TryGetValue("streamframe.send_queue_length", out var queue) && queue >= 0, + "send_queue_length 应有采样"); + } +} diff --git a/test/StreamFrame.Tests/FrameDecoderTests.cs b/test/StreamFrame.Tests/FrameDecoderTests.cs index 463a1c8..c517f0e 100644 --- a/test/StreamFrame.Tests/FrameDecoderTests.cs +++ b/test/StreamFrame.Tests/FrameDecoderTests.cs @@ -8,6 +8,9 @@ namespace StreamFrame.Tests; public class FrameDecoderTests { + /// 指标构造用的占位端点计数器（仅作标签，无实际意义）。 + private static int _portCounter; + /// /// 把一个 byte[] 拆成多个小块（模拟 TCP 分片），每次喂一块。 /// @@ -49,6 +52,7 @@ private static DecoderHarness CreateDecoder( var pipe = new Pipe(); var decoder = new FrameDecoder( pipe.Reader, framing, codec ?? StringCodec.Instance, relay, + metrics: new ConnectionMetrics($"test:{_portCounter++}"), sessionId: 1, maxIncompleteFrameBytes ?? framing.MaxPayloadBytes + 4096, incompleteFrameTimeoutMs, diff --git a/test/StreamFrame.Tests/SessionAwareConnectionTests.cs b/test/StreamFrame.Tests/SessionAwareConnectionTests.cs index 2e8e0be..4f9aa8e 100644 --- a/test/StreamFrame.Tests/SessionAwareConnectionTests.cs +++ b/test/StreamFrame.Tests/SessionAwareConnectionTests.cs @@ -278,6 +278,17 @@ public async Task SessionTeardown_PendingBoundSends_FailPromptly() Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException, $"失败类型应为会话失效/socket 故障，实际 {ex.GetType().Name}。"); } + + // 队满等待中的条目被 fault 后，其迟到的入队仍会成功（新 worker 腾出空位）—— + // worker 认领时的编号校验（跨会话残留防线）必须跳过它、绝不发送。 + // 连入新对端驱动排空，并断言三条旧消息一个字节都没有发到新会话 + using (var client2 = new TcpClient()) + { + await ConnectWithRetryAsync(client2, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected, timeoutMs: 8000); + await Task.Delay(500); // 留出排空窗口 + await AssertNoBytesAsync(client2.GetStream()); + } } [Fact] @@ -448,6 +459,52 @@ public async Task StaleGhostReconnect_DoesNotPolluteLiveSession() } } + [Fact] + public async Task GhostFault_DuringConnectedPublication_DoesNotKillNewbornSession() + { + // 评审"附带发现"的行为锁定：Connected 发布与 StartSession 之间到达的旧纪元幽灵故障 + // 必须被丢弃。入口的过期检查 + 纪元在发布点提前递增共同保证这一点—— + // 在发布线程上（StartSession 尚未执行）注入幽灵是最紧的时序构造 + var port = GetFreePort(); + await using var server = CreateServer(port); + + var injectGhost = false; + server.ConnectionChanged += (_, s) => + { + if (s != ConnectionState.Connected || !injectGhost) + return; + injectGhost = false; + typeof(StreamConnection).GetMethod( + "ScheduleReconnect", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)! + .Invoke(server, new object[] { 1 }); // 旧会话（epoch=1）的幽灵 + }; + server.Start(CancellationToken.None); + + using (var client1 = new TcpClient()) + { + await ConnectWithRetryAsync(client1, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + } + + await WaitForStateAsync(server, s => s != ConnectionState.Connected); + injectGhost = true; // 第二次 Connected 发布的回调里注入幽灵 + using (var client2 = new TcpClient()) + { + await ConnectWithRetryAsync(client2, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected, timeoutMs: 8000); + var id2 = server.CurrentSessionId; + Assert.NotEqual(0, id2); + + await Task.Delay(400); // 幽灵若未被丢弃，此刻已把会话 2 卷进 Retry + Assert.Equal(ConnectionState.Connected, server.State); + Assert.Equal(id2, server.CurrentSessionId); + + var task = server.SendInSessionAsync(id2, "alive"); + Assert.Equal("alive", await ReadFrameAsync(client2.GetStream())); + await AwaitDoneAsync(task, 3000, "新生会话发送"); + } + } + [Fact] public async Task CallerCancel_AfterCommit_HasNoEffect() { diff --git a/test/StreamFrame.Tests/SoakTests.cs b/test/StreamFrame.Tests/SoakTests.cs new file mode 100644 index 0000000..9b3a44b --- /dev/null +++ b/test/StreamFrame.Tests/SoakTests.cs @@ -0,0 +1,380 @@ +using System.Buffers.Binary; +using System.Collections.Concurrent; +using System.Net; +using System.Net.Sockets; +using System.Text; +using StreamFrame; + +namespace StreamFrame.Tests; + +/// +/// 浸泡/混沌测试（默认不运行）：真实 TCP 回环上的长时间运行与随机故障注入， +/// 验证消息完整性（顺序、无重复、无跨会话错发）与无悬挂。 +/// +/// 启用方式（本地手动触发，CI 与默认 dotnet test 均不运行）： +/// STREAMFRAME_SOAK_SECONDS=120 dotnet test -f net8.0 --filter FullyQualifiedName~Soak +/// 未设置环境变量时本套件立即返回（视为通过），不影响提交前的全量验证。 +/// +public class SoakTests +{ + private static int GetFreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + private static double SoakSeconds() + { + var raw = Environment.GetEnvironmentVariable("STREAMFRAME_SOAK_SECONDS"); + return double.TryParse(raw, out var value) && value > 0 ? value : 0; + } + + private static async Task ConnectWithRetryAsync(TcpClient client, int port) + { + // 混沌场景容忍慢收敛：服务端重连可能处于 2s 级的 accept 重试延迟（TIME_WAIT 竞争等）， + // 连接预算给足 60s，避免把"暂时拒绝"误判为测试失败 + for (var attempt = 0; ; attempt++) + { + try + { + await client.ConnectAsync(IPAddress.Loopback, port); + return; + } + catch (SocketException) when (attempt < 120) + { + await Task.Delay(500); + } + } + } + + private static async Task WaitForStateAsync( + StreamConnection connection, + Func predicate, + int timeoutMs = 15_000) + { + var deadline = TestClock.TickCount64 + timeoutMs; + while (TestClock.TickCount64 < deadline) + { + if (predicate(connection.State)) + return; + await Task.Delay(20); + } + + Assert.True(predicate(connection.State), $"等待连接状态超时（{timeoutMs}ms），当前 {connection.State}。"); + } + + [Fact] + public async Task Soak_LongRun_MessageIntegrity() + { + var seconds = SoakSeconds(); + if (seconds <= 0) + return; // 未启用浸泡模式：跳过（保持默认套件快速） + + var port = GetFreePort(); + await using var server = new StreamConnection( + new LengthPrefixFramer(), StringCodec.Instance, IPAddress.Loopback, port, isActive: false); + server.Start(CancellationToken.None); + + using var client = new TcpClient(); + await ConnectWithRetryAsync(client, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + + // 服务端 → 客户端方向：序列消息流；对端读尽并核对顺序与完整性 + var received = new ConcurrentQueue(); + var readCts = new CancellationTokenSource(TimeSpan.FromSeconds(seconds + 30)); + var readerTask = Task.Run(async () => + { + var stream = client.GetStream(); + var header = new byte[4]; + var body = new byte[4096]; + try + { + while (true) + { + var read = 0; + while (read < 4) + { + var n = await stream.ReadAsync(header, read, 4 - read, readCts.Token); + if (n == 0) + return; + read += n; + } + + var length = BinaryPrimitives.ReadInt32BigEndian(header); + var payload = new byte[length]; + read = 0; + while (read < length) + { + var n = await stream.ReadAsync(payload, read, length - read, readCts.Token); + if (n == 0) + return; + read += n; + } + + received.Enqueue(Encoding.UTF8.GetString(payload)); + } + } + catch (OperationCanceledException) + { + } + }); + + var deadline = TestClock.TickCount64 + (int)(seconds * 1000); + var sent = 0; + while (TestClock.TickCount64 < deadline) + { + await server.SendAsync($"msg-{sent}", CancellationToken.None); + sent++; + await Task.Delay(5); // ~200 msg/s + } + + readCts.Cancel(); + await readerTask; + + // 完整性：顺序、无丢失、无重复 + Assert.True(received.Count >= sent - 5, $"发送 {sent}，对端仅收 {received.Count}（允许尾部少数在途）"); + var index = 0; + foreach (var frame in received) + { + Assert.Equal($"msg-{index}", frame); + index++; + } + } + + [Fact] + public async Task Soak_Chaos_RandomFaults_NoHangAndIntegrity() + { + var seconds = SoakSeconds(); + if (seconds <= 0) + return; // 未启用浸泡模式：跳过 + + var seed = Environment.TickCount; + var random = new Random(seed); + var port = GetFreePort(); + await using var server = new StreamConnection( + new LengthPrefixFramer(), StringCodec.Instance, IPAddress.Loopback, port, + isActive: false, new StreamConnectionOptions { SendQueueCapacity = 256 }); + server.Start(CancellationToken.None); + + var peerFrames = new ConcurrentQueue(); + var pendingPlain = new ConcurrentQueue(); + var allTasks = new List(); + TcpClient? client = null; + NetworkStream? stream = null; + var readCts = new CancellationTokenSource(TimeSpan.FromSeconds(seconds + 30)); + var readerTask = Task.CompletedTask; + + var deadline = TestClock.TickCount64 + (int)(seconds * 1000); + var boundSeq = 0; + var plainSeq = 0; + + try + { + while (TestClock.TickCount64 < deadline) + { + var action = random.Next(100); + if (client is null) + { + // （重）连接对端并开始读 + client = new TcpClient(); + await ConnectWithRetryAsync(client, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + stream = client.GetStream(); + var currentStream = stream; + readerTask = Task.Run(async () => + { + var header = new byte[4]; + try + { + while (true) + { + var read = 0; + while (read < 4) + { + var n = await currentStream.ReadAsync(header, read, 4 - read, readCts.Token); + if (n == 0) + return; + read += n; + } + + var length = BinaryPrimitives.ReadInt32BigEndian(header); + var payload = new byte[length]; + read = 0; + while (read < length) + { + var n = await currentStream.ReadAsync(payload, read, length - read, readCts.Token); + if (n == 0) + return; + read += n; + } + + peerFrames.Enqueue(Encoding.UTF8.GetString(payload)); + } + } + catch (Exception ex) when (ex is ObjectDisposedException or IOException or OperationCanceledException) + { + } + }); + } + else if (action < 50) + { + // 会话绑定突发 + var id = server.CurrentSessionId; + if (id != 0) + { + for (var i = 0; i < random.Next(1, 5); i++) + { + var label = $"b{boundSeq++}"; + allTasks.Add(server.SendInSessionAsync(id, label)); + await Task.Delay(10); + } + } + } + else if (action < 70) + { + // 普通发送（跨会话续发）：记录标签，终局核对全部送达 + for (var i = 0; i < random.Next(1, 4); i++) + { + var label = $"p{plainSeq++}"; + pendingPlain.Enqueue(label); + await server.SendAsync(label); + } + } + else if (action < 80 && false && stream is not null) + { + // 半帧注入（未完成帧超时未启用：仅占住解码缓冲） + var junk = new byte[] { 0x00, 0x00, 0x0F, 0xA0 }; + await stream.WriteAsync(junk); + } + else if (action < 95) + { + // 杀对端（FIN 路径：正常释放） + if (client is not null) + { + client.Dispose(); + client = null; + stream = null; + await WaitForStateAsync(server, s => s != ConnectionState.Connected); + } + } + else if (false) + { + // 杀对端（RST 路径：Linger(true,0) 硬复位） + // 注：不混入用户显式 Reconnect()——用户重连与自动重连并发竞速会触发 + // 双 StartAsync 接受循环的存量脆弱点（监听状态机楔死，v2.3.0 评审 + // "附带发现"同源），属独立的状态过渡原子化设计任务，不在本套件覆盖 + if (client is not null) + { + client.LingerState = new LingerOption(true, 0); + client.Dispose(); + client = null; + stream = null; + await WaitForStateAsync(server, s => s != ConnectionState.Connected); + } + } + + await Task.Delay(random.Next(20, 120)); + else_placeholder: ; + } + } + finally + { + readCts.Cancel(); + + // 循环结束时对端可能仍存活：单客户端模式下监听已关闭（AcceptFirstClientOnly）， + // 必须先释放存量对端并等服务端离开 Connected，终局对端才能接入 + client?.Dispose(); + if (client is not null) + { + // 非断言式等待（尽力推进；真正的状态断言留给终局阶段） + var exitDeadline = TestClock.TickCount64 + 15_000; + while (TestClock.TickCount64 < exitDeadline && server.State == ConnectionState.Connected) + await Task.Delay(20); + } + } + + // 1) 所有会话绑定发送必须在时限内终结，失败类型合法；成功帧不得重复投递 + var boundSuccesses = new HashSet(); + foreach (var task in allTasks) + { + var done = await Task.WhenAny(task, Task.Delay(20_000)); + Assert.True(ReferenceEquals(task, done), "混沌后存在悬挂的会话绑定发送。"); + if (task.Status != TaskStatus.RanToCompletion) + { + var ex = task.Exception?.InnerExceptions[0]; + Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException, + $"混沌中失败类型意外：{ex?.GetType().Name}"); + } + } + + // 2) 重连一个干净对端，等普通消息全部续发送达（FIFO 完整性） + using (var finalClient = new TcpClient()) + { + await ConnectWithRetryAsync(finalClient, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + var finalStream = finalClient.GetStream(); + var finalCts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var finalReader = Task.Run(async () => + { + var header = new byte[4]; + try + { + while (true) + { + var read = 0; + while (read < 4) + { + var n = await finalStream.ReadAsync(header, read, 4 - read, finalCts.Token); + if (n == 0) + return; + read += n; + } + + var length = BinaryPrimitives.ReadInt32BigEndian(header); + var payload = new byte[length]; + read = 0; + while (read < length) + { + var n = await finalStream.ReadAsync(payload, read, length - read, finalCts.Token); + if (n == 0) + return; + read += n; + } + + peerFrames.Enqueue(Encoding.UTF8.GetString(payload)); + } + } + catch (Exception ex) when (ex is ObjectDisposedException or IOException or OperationCanceledException) + { + } + }); + + var finalDeadline = TestClock.TickCount64 + 20_000; + while (TestClock.TickCount64 < finalDeadline) + { + var plainSeen = 0; + foreach (var f in peerFrames) + if (f.StartsWith("p", StringComparison.Ordinal)) + plainSeen++; + if (plainSeen >= pendingPlain.Count) + break; + await Task.Delay(100); + } + + finalCts.Cancel(); + await finalReader; + } + + // 3) 终局核对：普通消息无丢失无重复（FIFO 序列完整） + var plainFrames = peerFrames.Where(f => f.StartsWith("p", StringComparison.Ordinal)).ToArray(); + Assert.Equal(pendingPlain.Count, plainFrames.Length); + Assert.Equal(pendingPlain.ToArray(), plainFrames); + + // 4) 绑定消息帧不重复（同一标签只允许出现一次——错发/重放探测器） + var boundFrames = peerFrames.Where(f => f.StartsWith("b", StringComparison.Ordinal)).ToArray(); + Assert.Equal(boundFrames.Length, boundFrames.Distinct().Count()); + } +} diff --git a/test/StreamFrame.Tests/XmlDocumentCodecTests.cs b/test/StreamFrame.Tests/XmlDocumentCodecTests.cs index a6db032..8a13030 100644 --- a/test/StreamFrame.Tests/XmlDocumentCodecTests.cs +++ b/test/StreamFrame.Tests/XmlDocumentCodecTests.cs @@ -1,5 +1,6 @@ using System.Buffers; using System.Text; +using System.Xml; using System.Xml.Linq; using StreamFrame.Protocols.Xml; @@ -43,4 +44,23 @@ public void Decode_EmptyFrame_Throws() var codec = new XmlDocumentCodec(); Assert.ThrowsAny(() => codec.Decode(new ReadOnlySequence(Array.Empty()))); } + + [Fact] + public void Decode_FrameExceedingMaxDocumentBytes_Throws() + { + var codec = new XmlDocumentCodec { MaxDocumentBytes = 16 }; + var payload = Encoding.UTF8.GetBytes("payload-longer-than-sixteen-bytes"); + + Assert.Throws( + () => codec.Decode(new ReadOnlySequence(payload))); + } + + [Fact] + public void Decode_MalformedXml_Throws() + { + var codec = new XmlDocumentCodec(); + var garbage = new byte[] { 0x3C, 0x78, 0x3E, 0xFF, 0xFE, 0x00, 0x7A }; // "" + 非法 UTF-8 序列 + + Assert.ThrowsAny(() => codec.Decode(new ReadOnlySequence(garbage))); + } } From b9e33cf2eccedf1d90340f5ce33c92af712280c5 Mon Sep 17 00:00:00 2001 From: Marci Date: Sat, 29 Aug 2026 03:09:57 +0800 Subject: [PATCH 2/2] =?UTF-8?q?test:=20=E6=8C=87=E6=A0=87=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E4=BF=AE=E6=AD=A3=E5=85=A5=E7=AB=99=E6=96=B9=E5=90=91?= =?UTF-8?q?=20+=20=E6=81=A2=E5=A4=8D=E5=AE=8C=E6=95=B4=E6=B7=B7=E6=B2=8C?= =?UTF-8?q?=E5=8A=A8=E4=BD=9C=E9=9B=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ConnectionLifecycle 测试误把 1024B 负载写成了入站方向（本地时序竞态 掩盖，ubuntu CI 暴露 Single 断言收到两帧）——出站仅 sendTask，入站 恰好一帧 - 恢复二分调试期间临时禁用的半帧注入与 RST 杀死动作，清理残留的 不可达分支与未引用标签；完整混沌 20s 冒烟通过 --- test/StreamFrame.Tests/ConnectionMetricsTests.cs | 7 ++----- test/StreamFrame.Tests/SoakTests.cs | 5 ++--- 2 files changed, 4 insertions(+), 8 deletions(-) diff --git a/test/StreamFrame.Tests/ConnectionMetricsTests.cs b/test/StreamFrame.Tests/ConnectionMetricsTests.cs index e387e12..51b9a5d 100644 --- a/test/StreamFrame.Tests/ConnectionMetricsTests.cs +++ b/test/StreamFrame.Tests/ConnectionMetricsTests.cs @@ -101,13 +101,10 @@ public async Task ConnectionLifecycle_ProducesAllInstruments() id1 = server.CurrentSessionId; var payload = new string('M', 1024); - var sendTask = server.SendInSessionAsync(id1, payload); // 普通方向：对端发来一帧 - var header = new byte[4]; - System.Buffers.Binary.BinaryPrimitives.WriteInt32BigEndian(header, payload.Length); - await client1.GetStream().WriteAsync(header); - await client1.GetStream().WriteAsync(Encoding.UTF8.GetBytes(payload)); + var sendTask = server.SendInSessionAsync(id1, payload); // 出站：回环缓冲即可写完，对端无需先读 await sendTask.WaitAsync(TimeSpan.FromSeconds(10)); + // 入站恰好一帧（长度 3 + "in!"）：驱动 frames_received / bytes_received await client1.GetStream().WriteAsync(new byte[] { 0x00, 0x00, 0x00, 0x03, (byte)'i', (byte)'n', 0x21 }); var deadline = TestClock.TickCount64 + 10_000; while (TestClock.TickCount64 < deadline) diff --git a/test/StreamFrame.Tests/SoakTests.cs b/test/StreamFrame.Tests/SoakTests.cs index 9b3a44b..144995c 100644 --- a/test/StreamFrame.Tests/SoakTests.cs +++ b/test/StreamFrame.Tests/SoakTests.cs @@ -243,7 +243,7 @@ public async Task Soak_Chaos_RandomFaults_NoHangAndIntegrity() await server.SendAsync(label); } } - else if (action < 80 && false && stream is not null) + else if (action < 80 && stream is not null) { // 半帧注入（未完成帧超时未启用：仅占住解码缓冲） var junk = new byte[] { 0x00, 0x00, 0x0F, 0xA0 }; @@ -260,7 +260,7 @@ public async Task Soak_Chaos_RandomFaults_NoHangAndIntegrity() await WaitForStateAsync(server, s => s != ConnectionState.Connected); } } - else if (false) + else { // 杀对端（RST 路径：Linger(true,0) 硬复位） // 注：不混入用户显式 Reconnect()——用户重连与自动重连并发竞速会触发 @@ -277,7 +277,6 @@ public async Task Soak_Chaos_RandomFaults_NoHangAndIntegrity() } await Task.Delay(random.Next(20, 120)); - else_placeholder: ; } } finally