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
26 changes: 25 additions & 1 deletion .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -119,4 +119,28 @@ jobs:
10.0.x

- name: PublishAot(NativeAOT + 裁剪警告门禁)
run: dotnet publish samples/StreamFrame.AotSmoke -c Release -r linux-x64
run: dotnet publish samples/StreamFrame.AotSmoke -c Release -r linux-x64 -o artifacts/aot

- name: Run NativeAOT smoke(连接、会话发送、普通发送与内容校验)
run: timeout 40s ./artifacts/aot/StreamFrame.AotSmoke

- name: Verify NativeAOT failure exit codes
shell: bash
run: |
set -euo pipefail
for fault in connect missing consumer content; do
category="$fault"
case "$fault" in
connect) category=connection ;;
missing) category=messages ;;
esac
status=0
timeout 40s ./artifacts/aot/StreamFrame.AotSmoke "--fault=$fault" > "$fault.log" 2>&1 || status=$?
cat "$fault.log"
# 只接受程序自身报告的指定故障:0、崩溃、启动失败或 timeout 的 124 均失败。
if [ "$status" -ne 1 ] || ! grep -Fq "AOT smoke FAILED [$category]:" "$fault.log"; then
echo "::error::Fault $fault was not detected correctly (exit=$status)"
exit 1
fi
echo "Verified expected failure: $fault (exit=$status)" | tee -a "$GITHUB_STEP_SUMMARY"
done
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
## [Unreleased]

### 修复
- **NativeAOT 运行验收([#71](https://github.com/CSJ608/StreamFrame/issues/71))**:CI 发布后执行 linux-x64 原生产物;有界等待连接与三条消息,校验会话绑定/普通发送及内容,观察消费任务并释放资源。动态选择端口并仅在连接阶段有限重选;连接失败、少收、消费异常和内容错误以非零退出,通过受控故障反证检查有效。保留警告为错误和非必需 AOT 检查。
- **Soak 队列续发与收尾([#67](https://github.com/CSJ608/StreamFrame/issues/67))**:发送 worker 在每次出队前检查会话取消,避免上一帧写完后拆除的旧 worker 取走并丢弃待续发普通条目;新增确定性门控回归。长流测试精确排空后关闭读 socket 并观察任务,修复 net48 仅取消令牌时读任务不收敛。
- **TCP KeepAlive 单位([#62](https://github.com/CSJ608/StreamFrame/issues/62))**:现代 .NET 将正毫秒值安全向上取整为秒,默认 30000/5000ms 正确设置为 30/5s;保留 netstandard2.0 IOControl 毫秒语义及默认关闭。增加真实 Socket 配置读回与边界回归,双语文档明确粒度差异。
- **终态连接等待([#65](https://github.com/CSJ608/StreamFrame/issues/65))**:取得连接等待器后复查停机状态,关闭首次注册与 Shutdown 的竞态;Dispose 或生命周期取消后的新等待也以任务取消结束,无需调用方令牌。补充终态调用、1000 次首次注册竞速及调用方取消隔离回归。
Expand Down
4 changes: 4 additions & 0 deletions README.en.md
Original file line number Diff line number Diff line change
Expand Up @@ -347,6 +347,10 @@ dotnet run -c Release --project bench/StreamFrame.Benchmarks # benchmarks (~5-
dotnet test -f net8.0 --collect:"XPlat Code Coverage" # coverage (CI also collects & summarizes)
```

## NativeAOT validation

The non-required `aot` CI check separately validates **NativeAOT publishing/trimming analysis** and **execution of the linux-x64 native binary**. The TCP loopback smoke verifies two session-bound sends, one ordinary send, and message contents; controlled faults verify nonzero exits. A successful publish alone does not prove runtime success. See the [AotSmoke guide](samples/StreamFrame.AotSmoke/README.md) for commands, deadlines, fault modes, and future transport coverage under [#55](https://github.com/CSJ608/StreamFrame/issues/55).

## Project layout

```
Expand Down
4 changes: 4 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -341,6 +341,10 @@ dotnet run -c Release --project bench/StreamFrame.Benchmarks # 性能基准(
dotnet test -f net8.0 --collect:"XPlat Code Coverage" # 覆盖率(CI 亦自动收集并写入运行摘要)
```

## NativeAOT 验证

CI 的非必需 `aot` 检查分别验证 **NativeAOT 发布/裁剪分析** 与 **linux-x64 原生产物运行**:真实 TCP 回环校验两条会话绑定发送、一条普通发送和消息内容,并用受控故障验证非零退出。发布成功不等于运行通过。命令、期限、故障模式及 [#55](https://github.com/CSJ608/StreamFrame/issues/55) 未来传输扩展边界见 [AotSmoke 说明](samples/StreamFrame.AotSmoke/README.md)。

## 项目结构

```
Expand Down
146 changes: 113 additions & 33 deletions samples/StreamFrame.AotSmoke/Program.cs
Original file line number Diff line number Diff line change
@@ -1,61 +1,141 @@
using System.Buffers;
using System.Net;
using System.Net.Sockets;
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 });
// 发布验证与原生运行验证是两道独立检查;见 README.md 和 ci.yml。
var fault = args.Length == 0 ? "none" : args.Length == 1 ? args[0] : "invalid";
if (fault is not ("none" or "--fault=connect" or "--fault=missing" or "--fault=consumer" or "--fault=content"))
{
Console.Error.WriteLine("Usage: StreamFrame.AotSmoke [--fault=connect|missing|consumer|content]");
return 2;
}

var received = 0;
_ = Task.Run(async () =>
var phase = "connection";
try
{
await foreach (var message in server.GetMessages())
// 公共 API 尚不暴露 port=0 的实际监听地址。先向 OS 申请端口;释放到绑定之间
// 存在竞争,因此仅在连接阶段最多重选三次,消息阶段绝不重试。
for (var attempt = 1; attempt <= 3; attempt++)
{
if (Interlocked.Increment(ref received) == 3)
break;
using var reservation = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
reservation.ExclusiveAddressUse = true;
reservation.Bind(new IPEndPoint(IPAddress.Loopback, 0));
var port = ((IPEndPoint)reservation.LocalEndPoint!).Port;
await using var server = CreateConnection(port, isActive: false);
await using var client = CreateConnection(port, isActive: true);
using var connectDeadline = new CancellationTokenSource(TimeSpan.FromSeconds(5));

Console.WriteLine($"Connecting attempt={attempt}, port={port}, fault={fault}");
if (fault != "--fault=connect")
{
reservation.Dispose();
server.Start(default);
}
// connect 故障保留已绑定但未监听的 socket,避免误连别的服务。
client.Start(default);
try
{
await Task.WhenAll(client.WaitForConnectedAsync(connectDeadline.Token),
server.WaitForConnectedAsync(connectDeadline.Token));
}
catch (OperationCanceledException) when (connectDeadline.IsCancellationRequested)
{
Console.Error.WriteLine($"Connection deadline exceeded (attempt={attempt}, port={port}).");
if (fault == "--fault=connect" || attempt == 3)
throw new SmokeFailure("connection", "Both peers must connect within the deadline.");
continue;
}

phase = "messages";
using var messageDeadline = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var consumer = ConsumeAsync(server, fault, messageDeadline.Token);
var sender = SendAsync(client, fault, messageDeadline.Token);
var work = Task.WhenAll(sender, consumer);
try
{
await work.WaitAsync(messageDeadline.Token);
}
finally
{
// 无论成功、超时还是消费抛错,都先停机,再观察所有已启动的应用任务。
// 只忽略收尾取消;消费/发送的业务异常仍传播至进程非零退出。
messageDeadline.Cancel();
await client.DisposeAsync();
await server.DisposeAsync();
try { await work; }
catch (OperationCanceledException) when (messageDeadline.IsCancellationRequested) { }
}

Console.WriteLine("AOT smoke OK: received=3, session-bound=2, ordinary=1, content verified");
return 0;
}
});
throw new SmokeFailure("connection", "Connection attempts exhausted.");
}
catch (Exception ex)
{
Console.Error.WriteLine($"AOT smoke FAILED [{(ex is SmokeFailure failure ? failure.Category : phase)}]: {ex}");
return 1;
}

server.Start(default);
client.Start(default);
await client.WaitForConnectedAsync();
static StreamConnection<JsonElement> CreateConnection(int port, bool isActive)
=> new(new LengthPrefixFramer(), SystemTextJsonCodec.Instance, IPAddress.Loopback, port, isActive,
new StreamConnectionOptions { ConnectRetryDelayMs = 200, IncompleteFrameTimeoutMs = 5_000 });

if (client is ISessionAwareStreamConnection<JsonElement> sessionAware)
static async Task SendAsync(StreamConnection<JsonElement> client, string fault, CancellationToken ct)
{
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);
var sessionId = client.CurrentSessionId;
if (sessionId <= 0)
throw new SmokeFailure("session", "Connected client must have a valid session ID.");
using var first = JsonDocument.Parse("""{"type":"aot","n":1}""");
using var second = JsonDocument.Parse("""{"type":"aot","n":2}""");
using var third = JsonDocument.Parse(fault == "--fault=content"
? """{"type":"wrong","n":3}""" : """{"type":"aot","n":3}""");
await client.SendInSessionAsync(sessionId, first.RootElement.Clone(), ct);
await client.SendInSessionAsync(sessionId, second.RootElement.Clone(), ct);
if (fault != "--fault=missing")
await client.SendAsync(third.RootElement.Clone(), ct);
}

await client.SendAsync(JsonDocument.Parse("""{"type":"aot","n":3}""").RootElement);
await Task.Delay(3_000);
static async Task ConsumeAsync(StreamConnection<JsonElement> server, string fault, CancellationToken ct)
{
var received = 0;
await foreach (var message in server.GetMessages(ct))
{
if (fault == "--fault=consumer")
throw new SmokeFailure("consumer", "Injected background consumer failure.");
var expected = ++received;
if (message.ValueKind != JsonValueKind.Object ||
!message.TryGetProperty("type", out var type) || type.ValueKind != JsonValueKind.String ||
type.GetString() != "aot" || !message.TryGetProperty("n", out var number) ||
number.ValueKind != JsonValueKind.Number || !number.TryGetInt32(out var n) || n != expected)
throw new SmokeFailure("content", $"Unexpected message #{expected}: {message.GetRawText()}");
Console.WriteLine($"Received and verified message {received}/3");
if (received == 3)
return;
}
throw new SmokeFailure("messages", $"Message stream ended early: received={received}, expected=3.");
}

await server.DisposeAsync();
await client.DisposeAsync();
Console.WriteLine(received == 3 ? "AOT smoke OK" : $"unexpected received={received}");
internal sealed class SmokeFailure(string category, string message) : Exception(message)
{
public string Category { get; } = category;
}

/// <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;
{
using var document = JsonDocument.Parse(frame.ToArray());
return document.RootElement.Clone();
}

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);
Expand Down
28 changes: 28 additions & 0 deletions samples/StreamFrame.AotSmoke/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
# NativeAOT 冒烟验证

`ci.yml` 的 `aot` job 保持非必需检查,不改变分支保护或普通 `build-test` 检查名。

- **可发布**:`dotnet publish` 成功,证明当前引用路径通过 NativeAOT 编译及裁剪分析;工程保留 `PublishAot` 和 `TreatWarningsAsErrors`,包括 IL2026/IL3050 等警告门禁。
- **运行冒烟通过**:在 Linux x64 上直接执行生成的原生程序,退出 0 并报告 `AOT smoke OK`。它验证真实 TCP 回环连接、两条会话绑定发送、一条普通发送,以及接收顺序和 JSON 内容。发布成功本身不代表运行通过。

在具有 .NET 8/10 SDK 和 NativeAOT 工具链的 Linux x64 环境执行:

```bash
dotnet publish samples/StreamFrame.AotSmoke -c Release -r linux-x64 -o artifacts/aot
timeout 40s ./artifacts/aot/StreamFrame.AotSmoke
```

连接每次最多等待 5 秒,消息收发整体最多等待 10 秒。使用 OS 分配的空闲端口;由于当前公共 API 未暴露被动端 `port=0` 的实际地址,探测释放与框架绑定存在竞争,连接超时可重选端口,最多三次。双方都连接成功才发送,消息或消费失败绝不重试。结束时取消任务、释放双方连接,并观察发送/消费任务;没有用固定睡眠判断成功。JSON 解码结果拥有独立数据,临时 `JsonDocument` 及时释放。

CI 先执行无故障的正常运行(任何非零直接使 job 失败),再执行以下受控故障。每次必须退出 **1** 且日志包含对应 `AOT smoke FAILED [类别]:` 才算反证通过;退出 0、启动错误、崩溃和外层超时 124 都不能当成有效故障。业务异常原文保留在日志中。无效命令行退出 2。

| 参数 | 注入方式 | 错误类别 |
|---|---|---|
| `--fault=connect` | 保留已绑定但未监听的端口,不启动服务端 | `connection` |
| `--fault=missing` | 不发送第三条消息,实际等待接收期限 | `messages` |
| `--fault=consumer` | 消费任务收到首条后抛错 | `consumer` |
| `--fault=content` | 第三条消息携带错误内容 | `content` |

例如单独执行 `timeout 40s ./artifacts/aot/StreamFrame.AotSmoke --fault=consumer` 应返回 1;直接将此命令作为 CI 运行步骤会使步骤失败。CI 中只有专门的反证步骤检查“期望非零”,正常路径没有容错或 `continue-on-error`。

本项覆盖现有 TCP + LengthPrefixFramer + JSON 路径,并不证明所有 codec/传输均兼容 AOT。[#55](https://github.com/CSJ608/StreamFrame/issues/55) 将来扩展 TLS/串口时,需要分别扩展发布与原生运行验证;本项不提前实现这些传输。
Loading