Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
23 changes: 19 additions & 4 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

### 新增
Expand All @@ -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
Expand Down
19 changes: 19 additions & 0 deletions README.en.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<JsonElement>
{
public static readonly SystemTextJsonCodec Instance = new();

public JsonElement Decode(in ReadOnlySequence<byte> frame, CancellationToken ct = default)
=> JsonDocument.Parse(frame.ToArray()).RootElement;

public void Encode(JsonElement message, IBufferWriter<byte> 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.
Expand Down
19 changes: 18 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,24 @@ public interface ICodec<TMessage>
}
```

**接新设备 = 写一个驱动**:实现 `ICodec<TMessage>` + 定义业务消息类,选一个帧策略,其余全部复用。官方示例见 `StreamFrame.Protocols.Xml`。
**接新设备 = 写一个驱动**:实现 `ICodec<TMessage>` + 定义业务消息类,选一个帧策略,其余全部复用。官方示例见 `StreamFrame.Protocols.Xml`。JSON 报文最简驱动(System.Text.Json,span 直写、AOT 安全):

```csharp
sealed class SystemTextJsonCodec : ICodec<JsonElement>
{
public static readonly SystemTextJsonCodec Instance = new();

public JsonElement Decode(in ReadOnlySequence<byte> frame, CancellationToken ct = default)
=> JsonDocument.Parse(frame.ToArray()).RootElement;

public void Encode(JsonElement message, IBufferWriter<byte> 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&lt;T&gt;)— 传输层

Expand Down
1 change: 1 addition & 0 deletions StreamFrame.slnx
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
</Folder>
<Folder Name="/samples/">
<Project Path="samples/StreamFrame.Demo/StreamFrame.Demo.csproj" />
<Project Path="samples/StreamFrame.AotSmoke/StreamFrame.AotSmoke.csproj" />
</Folder>
<Folder Name="/src/">
<Project Path="src/StreamFrame.Protocols.Xml/StreamFrame.Protocols.Xml.csproj" />
Expand Down
64 changes: 64 additions & 0 deletions samples/StreamFrame.AotSmoke/Program.cs
Original file line number Diff line number Diff line change
@@ -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<JsonElement>(
new LengthPrefixFramer(), SystemTextJsonCodec.Instance,
IPAddress.Loopback, port, isActive: false,
options: new StreamConnectionOptions { IncompleteFrameTimeoutMs = 5_000 });
var client = new StreamConnection<JsonElement>(
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<JsonElement> 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}");

/// <summary>System.Text.Json 的 span 直写 codec(AOT 安全:无反射序列化)。</summary>
internal sealed class SystemTextJsonCodec : ICodec<JsonElement>
{
public static readonly SystemTextJsonCodec Instance = new();

public JsonElement Decode(in ReadOnlySequence<byte> frame, CancellationToken ct = default)
=> JsonDocument.Parse(frame.ToArray()).RootElement;

public void Encode(JsonElement message, IBufferWriter<byte> 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);
}
}
20 changes: 20 additions & 0 deletions samples/StreamFrame.AotSmoke/StreamFrame.AotSmoke.csproj
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
<Project Sdk="Microsoft.NET.Sdk">

<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<!-- CI 以 RID publish 本工程(见 ci.yml 的 aot job):NativeAOT 发布涵盖裁剪分析,
IL2026/IL3050 等警告因 TreatWarningsAsErrors 直接红掉——防止库悄悄引入
反射/动态代码等不兼容依赖。注意:不要用命令行 -p:PublishAot=true 传递
(全局属性会传播到库的 netstandard2.0 目标,触发 NETSDK1124) -->
<PublishAot>true</PublishAot>
</PropertyGroup>

<ItemGroup>
<ProjectReference Include="..\..\src\StreamFrame\StreamFrame.csproj" />
</ItemGroup>

</Project>
4 changes: 3 additions & 1 deletion src/StreamFrame/Connection/StreamConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
138 changes: 138 additions & 0 deletions test/StreamFrame.Tests/GuardAndFramerEdgeTests.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,138 @@
using System.Buffers;
using StreamFrame;

namespace StreamFrame.Tests;

/// <summary>
/// 定向守卫/边界分支测试(覆盖率补测):帧定界器与缓冲写入器的参数守卫、超限拒绝路径。
/// </summary>
public class GuardAndFramerEdgeTests
{
// ---- StxEtxFramer ----

[Fact]
public void StxEtx_Ctor_NonPositiveMax_Throws()
=> Assert.Throws<ArgumentOutOfRangeException>(() => new StxEtxFramer(maxPayloadBytes: 0));

[Fact]
public void StxEtx_EncodeFrame_OversizePayload_Throws()
{
var framer = new StxEtxFramer(maxPayloadBytes: 8);
Assert.Throws<InvalidOperationException>(() => 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<InvalidOperationException>(() => framer.EncodeFrame(new byte[] { 0x02 }, writer));
Assert.Throws<InvalidOperationException>(() => 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<InvalidOperationException>(() => 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<InvalidOperationException>(() => 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<byte>(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<ArgumentOutOfRangeException>(() => new LengthPrefixFramer(maxPayloadBytes: 0));

[Fact]
public void LengthPrefix_EncodeFrame_OversizePayload_Throws()
{
var framer = new LengthPrefixFramer(maxPayloadBytes: 8);
Assert.Throws<InvalidOperationException>(() => 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<byte>(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<ArgumentOutOfRangeException>(() => writer.Advance(-1));
}

[Fact]
public void PooledBufferWriter_AdvanceBeyondCapacity_Throws()
{
using var writer = new PooledBufferWriter(16);
Assert.Throws<ArgumentOutOfRangeException>(() => writer.Advance(17));
}

[Fact]
public void BufferWriterStream_NullWriter_Throws()
=> Assert.Throws<ArgumentNullException>(() => 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());
}
}
Loading