From b413d3a1fab2eda77aebfc919967c1872f5255b7 Mon Sep 17 00:00:00 2001 From: Marci Date: Sat, 29 Aug 2026 13:42:19 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=B0=8F=E9=A1=B9=E6=B1=A0=E5=AD=90?= =?UTF-8?q?=E2=80=94=E2=80=94AOT=20=E9=97=A8=E7=A6=81=20+=20JSON=20codec?= =?UTF-8?q?=20=E7=A4=BA=E4=BE=8B=20+=20Pipe=20=E6=AE=B5=E5=B0=BA=E5=AF=B8?= =?UTF-8?q?=E5=AF=B9=E9=BD=90=20+=20=E8=A6=86=E7=9B=96=E7=8E=87=E8=A1=A5?= =?UTF-8?q?=E6=B5=8B=20+=20CHANGELOG=20=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - NativeAOT/裁剪门禁：samples/StreamFrame.AotSmoke 冒烟工程（引用库最小路径 含会话感知发送）+ ci.yml aot job（ubuntu PublishAot，警告即错误；非必需 检查不影响分支保护）。注意：PublishAot 在工程内开启而非命令行传递—— 全局属性会传播到库的 ns2.0 目标触发 NETSDK1124 - README 双语新增 System.Text.Json 最简 codec 示例（span 直写、AOT 安全） - 接收 Pipe 段尺寸对齐 SocketReceiveBufferSize（64KB 整帧常驻单段，Kestrel 同款）；本机两轮基准在噪声内无法确立效果，保留（98×3 无回归），待 soak 复检 - 覆盖率补测：+13 个守卫/边界测试（StxEtx/LengthPrefix 守卫与超限重同步、 PooledBufferWriter/BufferWriterStream 参数守卫），分支 78.9%→82.3% - CHANGELOG 修复：v2.5.0 归档脚本锚点未匹配（无 assert 静默失败）导致 2.5.0 内容错挂 2.4.0 段落、Release 正文回退自动生成——按两 tag 实际内容还原拆分， 并已回填 v2.5.0 Release 正文 --- .github/workflows/ci.yml | 19 +++ CHANGELOG.md | 23 ++- README.en.md | 19 +++ README.md | 19 ++- StreamFrame.slnx | 1 + samples/StreamFrame.AotSmoke/Program.cs | 64 ++++++++ .../StreamFrame.AotSmoke.csproj | 20 +++ .../Connection/StreamConnection.cs | 4 +- .../GuardAndFramerEdgeTests.cs | 138 ++++++++++++++++++ 9 files changed, 301 insertions(+), 6 deletions(-) create mode 100644 samples/StreamFrame.AotSmoke/Program.cs create mode 100644 samples/StreamFrame.AotSmoke/StreamFrame.AotSmoke.csproj create mode 100644 test/StreamFrame.Tests/GuardAndFramerEdgeTests.cs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 707e5ed..45c2881 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -101,3 +101,22 @@ jobs: name: coverage-${{ github.run_number }} path: coverage/ retention-days: 14 + + # NativeAOT/裁剪兼容性门禁（非必需检查）：发布引用库的冒烟工程（PublishAot 在工程内开启， + # 涵盖裁剪分析）。IL2026/IL3050 等警告经 TreatWarningsAsErrors 变为错误，防止库悄悄 + # 引入反射/动态代码。ubuntu runner 自带 AOT 原生工具链。 + aot: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v7 + + - name: Set up .NET + uses: actions/setup-dotnet@v6 + with: + dotnet-version: | + 8.0.x + 10.0.x + + - name: PublishAot（NativeAOT + 裁剪警告门禁） + run: dotnet publish samples/StreamFrame.AotSmoke -c Release -r linux-x64 diff --git a/CHANGELOG.md b/CHANGELOG.md index ce2d353..3d0bec6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,9 +5,28 @@ ## [Unreleased] +### 新增 +- **NativeAOT/裁剪兼容性门禁**：`samples/StreamFrame.AotSmoke` 冒烟工程（引用库的最小可运行路径，含会话感知发送）+ ci.yml 新增 `aot` job（ubuntu 上 PublishAot 发布，IL2026/IL3050 等警告经 TreatWarningsAsErrors 直接红掉；非必需检查，不影响分支保护）。本地 managed 侧已验证零裁剪/AOT 警告。 +- README 双语新增 System.Text.Json 最简 codec 示例（span 直写、AOT 安全）。 + ### 变更 +- 接收 Pipe 段尺寸对齐 `SocketReceiveBufferSize`（大报文整帧常驻单段，Kestrel 同款做法）；本机基准两轮 48.75µs / 93µs，效果在噪声内无法确立，保留改动（语义合理、98×3 测试无回归），待每夜 soak 与后续基准复检。 +- CHANGELOG 修复：v2.5.0 归档时脚本锚点未匹配导致 2.5.0 内容错挂在 2.4.0 段落（Release 正文回退为自动生成）——已按两个 tag 的实际内容拆分还原（监听加固归 2.4.0，代次门控/大报文/观察基建归 2.5.0）。 + +### 修复 - 暂无 +## [2.5.0] - 2026-08-29 + +### 新增 +- **重连竞速长期观察基建**：`StateTransitionRecorder`（状态机合法转移、非 Connected 态编号归零、Connected 态编号严格递增、终态收敛的全程校验 + Connected→Connected 直连迁移计数）、`Soak_ReconnectRacing_LongRun` 竞速专用混沌（60% 动作并发竞速 + FIFO 完整性探针）、每夜 600s 双平台 soak CI（`.github/workflows/soak.yml`，北京时间 03:00，失败自动通知）。 + +### 变更 +- **大报文优化与归因**：发送编码缓冲自适应初始尺寸（按连接记忆上一帧高水位、封顶 1MB——稳态尺寸相近的协议不再从 1KB 起爬几何增长梯子）。新增 64KB 三 codec 归因基准（byte[] 透传 / string+span / string+分配式），结论：框架纯成本距裸 TCP 仅 ≈20–30%，其余为 codec 写法税（span 重载可消除中间数组）与 string 消息固有税（UTF-16 物化 ≈2× 报文）。README 双语新增"大报文指南"，bench/README 附归因表。 + +### 修复 +- **接受循环代次门控（#47 阶段二，竞速泄漏监听器 bug）**：用户重连与自动重连竞速产生多个并发 StartAsync 接受循环时，被取代的"僵尸"循环完成 accept 后可能看到 `_server` 已被新循环替换而跳过单客户端的监听器关闭——**泄漏一个仍在监听但永无人 accept 的 socket**，客户端 SYN 进入无人消费的 backlog 后 `ConnectAsync` 永久挂起（远比"拒绝"严重）。新增 `_acceptLoopId` 代次：每次 StartCore 递增，被取代的循环在取得 `_acceptLock` 后静默退出，监听器的创建/关闭/accept 只归属当代循环。由新增的重连竞速混沌（`Soak_ReconnectRacing_LongRun`，60% 动作占比并发竞速）复现并锁定。 + ## [2.4.0] - 2026-08-29 ### 新增 @@ -20,11 +39,7 @@ - demo 新增场景 5：会话感知收发（编号生命周期、整帧写完才完成、旧会话发送失效不重放、新会话照常收发）。 - **基准矩阵扩容 + 性能文档重写**：新增内置指标开销微基准（0.5–1.0 ns/次、零分配）、裸 TCP 地板基准（框架税：小消息反而快于裸写 13–52%，64KB 约 3–4×）、会话感知/接收视图/未完成帧超时的三轮对比（SendInSessionAsync ≈2× SendAsync、+≈470 B/条；其余无可测差异）；端到端吞吐参数化到 64B/1KB/64KB 两轮区间。bench/README 全量刷新（环境、噪声披露、框架税绝对+百分比并列），README 双语性能摘要同步。 -### 变更（2.4.0 后新增） -- **大报文优化与归因**：发送编码缓冲自适应初始尺寸（按连接记忆上一帧高水位、封顶 1MB——稳态尺寸相近的协议不再从 1KB 起爬几何增长梯子）。新增 64KB 三 codec 归因基准（byte[] 透传 / string+span / string+分配式），结论：框架纯成本距裸 TCP 仅 ≈20–30%，其余为 codec 写法税（span 重载可消除中间数组）与 string 消息固有税（UTF-16 物化 ≈2× 报文）。README 双语新增"大报文指南"，bench/README 附归因表。 - ### 修复 -- **接受循环代次门控（#47 阶段二，竞速泄漏监听器 bug）**：用户重连与自动重连竞速产生多个并发 StartAsync 接受循环时，被取代的"僵尸"循环完成 accept 后可能看到 `_server` 已被新循环替换而跳过单客户端的监听器关闭——**泄漏一个仍在监听但永无人 accept 的 socket**，客户端 SYN 进入无人消费的 backlog 后 `ConnectAsync` 永久挂起（远比"拒绝"严重）。新增 `_acceptLoopId` 代次：每次 StartCore 递增，被取代的循环在取得 `_acceptLock` 后静默退出，监听器的创建/关闭/accept 只归属当代循环。由新增的重连竞速混沌（`Soak_ReconnectRacing_LongRun`，60% 动作占比并发竞速）复现并锁定。 - **被动端监听韧性加固（#47，防御性）**：排查确认原始"用户 `Reconnect()` 竞速楔死监听"**无法复现**（顺序/并发两组确定性复现 + 混沌套件恢复该动作后多轮全绿，此前观察到的楔死系当时测试自身缺陷），但修复排查中识别的两个真实弱点——① 接受循环的重试延迟改为锁外等待（旧实现整个重试循环持有 `_acceptLock`，绑定/接受持续失败期间会堵死其它接受循环）；② 监听 socket 设置 `SO_REUSEADDR`（服务端主动关闭后立即重绑不受 2MSL/TIME_WAIT 限制；Windows 实测宽松，Linux 上无此选项会遇 `EADDRINUSE`，一并设置保持跨平台一致）。新增确定性回归测试（8 轮服务端主动重连，端口须秒级可重新接受），混沌套件恢复用户重连动作。 ## [2.3.1] - 2026-08-28 diff --git a/README.en.md b/README.en.md index 49b1bf5..54c5bf5 100644 --- a/README.en.md +++ b/README.en.md @@ -34,6 +34,25 @@ XML message driver (optional): dotnet add package StreamFrame.Protocols.Xml ``` +Minimal JSON driver (System.Text.Json, span-written, AOT-safe): + +```csharp +sealed class SystemTextJsonCodec : ICodec +{ + public static readonly SystemTextJsonCodec Instance = new(); + + public JsonElement Decode(in ReadOnlySequence frame, CancellationToken ct = default) + => JsonDocument.Parse(frame.ToArray()).RootElement; + + public void Encode(JsonElement message, IBufferWriter writer, CancellationToken ct = default) + { + var raw = message.GetRawText(); // serialized form (non-ASCII already escaped) + var span = writer.GetSpan(Encoding.UTF8.GetMaxByteCount(raw.Length)); + writer.Advance(Encoding.UTF8.GetBytes(raw, span)); + } +} +``` + ## Quick start One connection = one **framer** + one **codec** + address/port/mode. Framing and codec are both **fixed per connection**: a connection carries exactly one message type in exactly one frame format. diff --git a/README.md b/README.md index a5e2c54..defe51c 100644 --- a/README.md +++ b/README.md @@ -87,7 +87,24 @@ public interface ICodec } ``` -**接新设备 = 写一个驱动**：实现 `ICodec` + 定义业务消息类，选一个帧策略，其余全部复用。官方示例见 `StreamFrame.Protocols.Xml`。 +**接新设备 = 写一个驱动**：实现 `ICodec` + 定义业务消息类，选一个帧策略，其余全部复用。官方示例见 `StreamFrame.Protocols.Xml`。JSON 报文最简驱动（System.Text.Json，span 直写、AOT 安全）： + +```csharp +sealed class SystemTextJsonCodec : ICodec +{ + public static readonly SystemTextJsonCodec Instance = new(); + + public JsonElement Decode(in ReadOnlySequence frame, CancellationToken ct = default) + => JsonDocument.Parse(frame.ToArray()).RootElement; + + public void Encode(JsonElement message, IBufferWriter writer, CancellationToken ct = default) + { + var raw = message.GetRawText(); // 序列化产物（已转义的非 ASCII 安全直写） + var span = writer.GetSpan(Encoding.UTF8.GetMaxByteCount(raw.Length)); + writer.Advance(Encoding.UTF8.GetBytes(raw, span)); + } +} +``` ### 连接（IStreamConnection<T>）— 传输层 diff --git a/StreamFrame.slnx b/StreamFrame.slnx index 127037a..e2a3fcf 100644 --- a/StreamFrame.slnx +++ b/StreamFrame.slnx @@ -4,6 +4,7 @@ + diff --git a/samples/StreamFrame.AotSmoke/Program.cs b/samples/StreamFrame.AotSmoke/Program.cs new file mode 100644 index 0000000..8bbb807 --- /dev/null +++ b/samples/StreamFrame.AotSmoke/Program.cs @@ -0,0 +1,64 @@ +using System.Buffers; +using System.Net; +using System.Text.Json; +using StreamFrame; + +// AOT/裁剪兼容性冒烟：引用 StreamFrame 的最小可运行路径（连接生命周期 + 定界 + codec + +// 会话感知 + 指标记录调用），确保库不依赖被裁剪掉的能力。CI 以 PublishAot 发布本工程， +// 裁剪/AOT 警告会因 TreatWarningsAsErrors 直接红掉。 +var port = 19_002; + +var server = new StreamConnection( + new LengthPrefixFramer(), SystemTextJsonCodec.Instance, + IPAddress.Loopback, port, isActive: false, + options: new StreamConnectionOptions { IncompleteFrameTimeoutMs = 5_000 }); +var client = new StreamConnection( + new LengthPrefixFramer(), SystemTextJsonCodec.Instance, + IPAddress.Loopback, port, isActive: true, + options: new StreamConnectionOptions { ConnectRetryDelayMs = 200 }); + +var received = 0; +_ = Task.Run(async () => +{ + await foreach (var message in server.GetMessages()) + { + if (Interlocked.Increment(ref received) == 3) + break; + } +}); + +server.Start(default); +client.Start(default); +await client.WaitForConnectedAsync(); + +if (client is ISessionAwareStreamConnection sessionAware) +{ + var sessionId = sessionAware.CurrentSessionId; + await sessionAware.SendInSessionAsync(sessionId, JsonDocument.Parse("""{"type":"aot","n":1}""").RootElement); + await sessionAware.SendInSessionAsync(sessionId, JsonDocument.Parse("""{"type":"aot","n":2}""").RootElement); +} + +await client.SendAsync(JsonDocument.Parse("""{"type":"aot","n":3}""").RootElement); +await Task.Delay(3_000); + +await server.DisposeAsync(); +await client.DisposeAsync(); +Console.WriteLine(received == 3 ? "AOT smoke OK" : $"unexpected received={received}"); + +/// System.Text.Json 的 span 直写 codec（AOT 安全：无反射序列化）。 +internal sealed class SystemTextJsonCodec : ICodec +{ + public static readonly SystemTextJsonCodec Instance = new(); + + public JsonElement Decode(in ReadOnlySequence frame, CancellationToken ct = default) + => JsonDocument.Parse(frame.ToArray()).RootElement; + + public void Encode(JsonElement message, IBufferWriter writer, CancellationToken ct = default) + { + // span 直写（大报文指南推荐写法）：编码器直写 IBufferWriter，无中间数组 + var raw = message.GetRawText(); + var span = writer.GetSpan(System.Text.Encoding.UTF8.GetMaxByteCount(raw.Length)); + var written = System.Text.Encoding.UTF8.GetBytes(raw, span); + writer.Advance(written); + } +} diff --git a/samples/StreamFrame.AotSmoke/StreamFrame.AotSmoke.csproj b/samples/StreamFrame.AotSmoke/StreamFrame.AotSmoke.csproj new file mode 100644 index 0000000..90c5717 --- /dev/null +++ b/samples/StreamFrame.AotSmoke/StreamFrame.AotSmoke.csproj @@ -0,0 +1,20 @@ + + + + Exe + net8.0 + enable + enable + true + + true + + + + + + + diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs index f2986af..ac9eebb 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -550,7 +550,9 @@ private void StartSession(long sessionId, int epoch) var cts = new CancellationTokenSource(); var token = cts.Token; - var pipe = new Pipe(); + // 段尺寸对齐接收缓冲：大报文（64KB 级）在默认 4KB 段下跨 16 段重组， + // 对齐后整帧常驻单段（64KB 基准实测见 bench/README） + var pipe = new Pipe(new PipeOptions(minimumSegmentSize: _options.SocketReceiveBufferSize)); _sessionCts = cts; _pipe = pipe; diff --git a/test/StreamFrame.Tests/GuardAndFramerEdgeTests.cs b/test/StreamFrame.Tests/GuardAndFramerEdgeTests.cs new file mode 100644 index 0000000..a800220 --- /dev/null +++ b/test/StreamFrame.Tests/GuardAndFramerEdgeTests.cs @@ -0,0 +1,138 @@ +using System.Buffers; +using StreamFrame; + +namespace StreamFrame.Tests; + +/// +/// 定向守卫/边界分支测试（覆盖率补测）：帧定界器与缓冲写入器的参数守卫、超限拒绝路径。 +/// +public class GuardAndFramerEdgeTests +{ + // ---- StxEtxFramer ---- + + [Fact] + public void StxEtx_Ctor_NonPositiveMax_Throws() + => Assert.Throws(() => new StxEtxFramer(maxPayloadBytes: 0)); + + [Fact] + public void StxEtx_EncodeFrame_OversizePayload_Throws() + { + var framer = new StxEtxFramer(maxPayloadBytes: 8); + Assert.Throws(() => framer.EncodeFrame(new byte[9], new TestWrittenBufferWriter())); + } + + [Fact] + public void StxEtx_EncodeFrame_PayloadContainsStxOrEtx_Throws() + { + var framer = new StxEtxFramer(); + using var writer = new TestWrittenBufferWriter(); + + Assert.Throws(() => framer.EncodeFrame(new byte[] { 0x02 }, writer)); + Assert.Throws(() => framer.EncodeFrame(new byte[] { 0x03 }, writer)); + } + + [Fact] + public void StxEtx_EndFrame_OversizePayload_Throws() + { + // BeginFrame(STX) + 20 字节负载 + EndFrame：EndFrame 发现负载超 MaxPayloadBytes=8 + var framer = new StxEtxFramer(maxPayloadBytes: 8); + using var writer = new TestWrittenBufferWriter(); + framer.BeginFrame(writer); + writer.GetSpan(20); // 触发扩容到 20+ + writer.Advance(20); + + Assert.Throws(() => framer.EndFrame(writer)); + } + + [Fact] + public void StxEtx_EndFrame_PayloadContainsStxOrEtx_Throws() + { + var framer = new StxEtxFramer(); + using var writer = new TestWrittenBufferWriter(); + framer.BeginFrame(writer); + writer.GetSpan(1)[0] = 0x02; // 负载内的 STX + writer.Advance(1); + + Assert.Throws(() => framer.EndFrame(writer)); + } + + [Fact] + public void StxEtx_Decode_OversizeFrame_DiscardsAndRescans() + { + // STX + 9 字节负载 + ETX，MaxPayloadBytes=8：超长帧被丢弃，扫描继续到下一个合法帧 + var framer = new StxEtxFramer(maxPayloadBytes: 8); + var buffer = new ReadOnlySequence(new byte[] + { + 0x02, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x03, // 超长帧（9B 负载） + 0x02, 0x4F, 0x4B, 0x03, // 合法帧（2B 负载） + }); + + Assert.True(framer.TryDecodeFrame(ref buffer, out var payload)); + Assert.Equal(new byte[] { 0x4F, 0x4B }, payload.ToArray()); + Assert.True(buffer.IsEmpty); // 两个帧都被消化 + } + + // ---- LengthPrefixFramer ---- + + [Fact] + public void LengthPrefix_Ctor_NonPositiveMax_Throws() + => Assert.Throws(() => new LengthPrefixFramer(maxPayloadBytes: 0)); + + [Fact] + public void LengthPrefix_EncodeFrame_OversizePayload_Throws() + { + var framer = new LengthPrefixFramer(maxPayloadBytes: 8); + Assert.Throws(() => framer.EncodeFrame(new byte[9], new TestWrittenBufferWriter())); + } + + [Fact] + public void LengthPrefix_Decode_NegativeLength_DiscardsHeader() + { + // 长度头高位为 1 → 负数 → 非法，丢弃 4 字节重同步；后续合法帧可解 + var framer = new LengthPrefixFramer(); + var bytes = new byte[4 + 4 + 2 + 4 + 2]; + System.Buffers.Binary.BinaryPrimitives.WriteInt32BigEndian(bytes.AsSpan(0, 4), -1); // 非法头 + System.Buffers.Binary.BinaryPrimitives.WriteInt32BigEndian(bytes.AsSpan(4, 4), 2); // 合法帧头 + bytes[8] = (byte)'o'; + bytes[9] = (byte)'k'; + _ = bytes.AsSpan(10); // （占位：保持数组形状可读） + + var buffer = new ReadOnlySequence(bytes); + // 第一次调用：丢弃非法头后本次无完整帧（数据不足路径） + if (framer.TryDecodeFrame(ref buffer, out _)) + Assert.Fail("负长度头不应解出帧。"); + // 第二次调用：合法帧就位 + Assert.True(framer.TryDecodeFrame(ref buffer, out var payload)); + Assert.Equal("ok"u8.ToArray(), payload.ToArray()); + } + + // ---- PooledBufferWriter / BufferWriterStream ---- + + [Fact] + public void PooledBufferWriter_AdvanceNegative_Throws() + { + using var writer = new PooledBufferWriter(16); + Assert.Throws(() => writer.Advance(-1)); + } + + [Fact] + public void PooledBufferWriter_AdvanceBeyondCapacity_Throws() + { + using var writer = new PooledBufferWriter(16); + Assert.Throws(() => writer.Advance(17)); + } + + [Fact] + public void BufferWriterStream_NullWriter_Throws() + => Assert.Throws(() => new BufferWriterStream(null!)); + + [Fact] + public void BufferWriterStream_Write_AppendsBytes() + { + using var target = new TestWrittenBufferWriter(); + using var stream = new BufferWriterStream(target); + stream.Write(new byte[] { 1, 2, 3 }, 0, 3); + + Assert.Equal(new byte[] { 1, 2, 3 }, target.WrittenSpan.ToArray()); + } +}