diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c90065..13ce24e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ - **`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：会话感知收发（编号生命周期、整帧写完才完成、旧会话发送失效不重放、新会话照常收发）。 +- **基准矩阵扩容 + 性能文档重写**：新增内置指标开销微基准（0.5–1.0 ns/次、零分配）、裸 TCP 地板基准（框架税：小消息反而快于裸写 13–52%，64KB 约 3–4×）、会话感知/接收视图/未完成帧超时的三轮对比（SendInSessionAsync ≈2× SendAsync、+≈470 B/条；其余无可测差异）；端到端吞吐参数化到 64B/1KB/64KB 两轮区间。bench/README 全量刷新（环境、噪声披露、框架税绝对+百分比并列），README 双语性能摘要同步。 ### 修复 - **被动端监听韧性加固（#47，防御性）**：排查确认原始"用户 `Reconnect()` 竞速楔死监听"**无法复现**（顺序/并发两组确定性复现 + 混沌套件恢复该动作后多轮全绿，此前观察到的楔死系当时测试自身缺陷），但修复排查中识别的两个真实弱点——① 接受循环的重试延迟改为锁外等待（旧实现整个重试循环持有 `_acceptLock`，绑定/接受持续失败期间会堵死其它接受循环）；② 监听 socket 设置 `SO_REUSEADDR`（服务端主动关闭后立即重绑不受 2MSL/TIME_WAIT 限制；Windows 实测宽松，Linux 上无此选项会遇 `EADDRINUSE`，一并设置保持跨平台一致）。新增确定性回归测试（8 轮服务端主动重连，端口须秒级可重新接受），混沌套件恢复用户重连动作。 diff --git a/README.en.md b/README.en.md index 62df38a..4caade5 100644 --- a/README.en.md +++ b/README.en.md @@ -229,7 +229,8 @@ Measured with BenchmarkDotNet (details and how to reproduce in [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` ≈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. +- **End-to-end** (real TCP loopback, 2026-08-29 two-round ranges, built-in metrics on): one-way throughput ≈130–135k msgs/s at 1KB and ≈210–240k at 64B (LengthPrefix); round-trip latency ≈66–74 µs; the XML codec costs 2–16 µs per message (400B–4KB) — serialization dominates, not framing. **Framework tax depends on message size**: for small messages the bounded-queue pipeline is even faster than the naive serialized-NetworkStream-write baseline (13–52%); at 64KB the cost is ≈3–4× (encoding-buffer allocations dominate — recorded as an optimization direction); +- **Cost of the new features, measured**: built-in metrics cost 0.5–1.0 ns per record with zero allocation when nobody subscribes (fine to leave on); `SendInSessionAsync` is roughly 2× the time of `SendAsync` with +≈470 B/msg allocated (the price of session semantics — use `SendAsync` when you don't need them); the receive views and the disabled incomplete-frame timeout show no measurable difference. Absolute values vary by machine; full data and noise disclosure in [bench/README.md](bench/README.md). ## Supported frameworks @@ -252,7 +253,7 @@ The `netstandard2.0` asset is validated by the **full net48 test suite** (real T ```bash dotnet build StreamFrame.slnx dotnet test -dotnet run --project samples/StreamFrame.Demo # 4-scenario end-to-end demo +dotnet run --project samples/StreamFrame.Demo # 5-scenario end-to-end demo dotnet run -c Release --project bench/StreamFrame.Benchmarks # benchmarks (~5-10 min) dotnet test -f net8.0 --collect:"XPlat Code Coverage" # coverage (CI also collects & summarizes) ``` diff --git a/README.md b/README.md index 7702cfc..b5b555f 100644 --- a/README.md +++ b/README.md @@ -226,7 +226,8 @@ BenchmarkDotNet 实测（详见 [bench/README.md](bench/README.md)，可本地 - **流式编码**：每帧堆分配减半（单缓冲 vs 双缓冲），小负载耗时快 25–35%，大负载持平； - **切帧吞吐**：`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)。 +- **端到端**（真实 TCP 回环，2026-08-29 两轮区间，含内置指标）：单向吞吐 1KB ≈13–13.5 万条/秒、64B ≈21–24 万条/秒（LengthPrefix）；往返延迟 ≈66–74 µs；XML codec 每条报文 2–16 µs（400B–4KB）——序列化开销远大于定界层。**框架税分尺寸**：小消息下队列流水线反而快于裸 NetworkStream 逐条 Write 的对照（13–52%），64KB 大消息约 3–4×（编码缓冲分配主导，优化方向已记录）； +- **新特性成本实测**：内置指标无监听时 0.5–1.0 ns/次、零分配（可常开）；`SendInSessionAsync` 约为 `SendAsync` 的 2 倍耗时、+≈470 B/条分配（会话语义的代价，不需要时用 `SendAsync`）；接收视图与未完成帧超时（关闭态）均无可测差异。绝对值随机器浮动，完整数据与噪声披露见 [bench/README.md](bench/README.md)。 ## 支持框架 @@ -249,7 +250,7 @@ BenchmarkDotNet 实测（详见 [bench/README.md](bench/README.md)，可本地 ```bash dotnet build StreamFrame.slnx dotnet test -dotnet run --project samples/StreamFrame.Demo # 四场景端到端 demo +dotnet run --project samples/StreamFrame.Demo # 五场景端到端 demo dotnet run -c Release --project bench/StreamFrame.Benchmarks # 性能基准（约 5-10 分钟） dotnet test -f net8.0 --collect:"XPlat Code Coverage" # 覆盖率（CI 亦自动收集并写入运行摘要） ``` diff --git a/bench/README.md b/bench/README.md index e6aaa7a..0d90f2b 100644 --- a/bench/README.md +++ b/bench/README.md @@ -18,7 +18,25 @@ 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 表为历史数据） +## 结果 + +### 测量环境（2026-08-29，宏基准为两轮区间） + +- CPU：12th Gen Intel Core i5-12500H（笔记本）；内存 32GB；Windows 10 19045；.NET 10.0.11 +- BenchmarkDotNet SimpleJob：微基准 warmup 3 / iteration 15，宏基准 warmup 2 / iteration 10；真实 TCP 回环 +- **噪声披露**：测量在长时间高负载的开发机上完成，宏基准（回环端到端）绝对值两轮漂移可达 ±30%，下表以区间标注、仅供量级参考；**关键差值（会话发送成本、接收视图、超时开销）经第三轮复核稳定**。微基准与差值结论可信度高。请以自己环境的复现为准。 + +### 内置指标开销（微基准，无监听者的生产默认态） + +| 调用 | 耗时 | 分配 | +|---|---:|---:| +| 单次计数/直方图记录 | 0.5–1.0 ns | 0 | +| 接收路径每消息合计（字节块 + 帧） | ≈1.6 ns | 0 | +| 发送路径每消息合计（入队采样 + 字节块 + 帧） | ≈2.6 ns | 0 | + +结论：不订阅 Meter 时指标开销为**亚纳秒到个位纳秒级、零分配**——按 10 万条/秒的流量折算，发送路径合计 ≈0.026 µs/秒，完全可忽略。 + +### 帧定界（微基准，2026-08-28 数据仍有效） | 方法 | 负载 | 耗时 | 分配 | |---|---|---:|---:| @@ -40,7 +58,43 @@ dotnet run -c Release --project bench/StreamFrame.Benchmarks -- --filter "*EndTo ¹ 恰好压在 `EncodeBufferInitialSize`（默认 1024）的缓冲增长边界上：流式路径触发一次扩容租借， 抵消了省下的那次拷贝。负载远大于初始缓冲后两条路径的拷贝量趋同（几何扩容摊销）。 -### Codec（XmlDocumentCodec） +### 端到端单向吞吐（真实 TCP 回环，1 万条/轮，两轮区间） + +| Framer | 64B | 1KB | 64KB | +|---|---:|---:|---:| +| LengthPrefix | 4.1–4.7 µs/条 ≈ 21–24 万条/秒 | 7.4–7.7 µs/条 ≈ 13–13.5 万条/秒 | 134–188 µs/条 ≈ 0.53–0.75 万条/秒 | +| StxEtx | 5.5–6.9 µs/条 ≈ 15–18 万条/秒 | 4.3–8.7 µs/条 ≈ 12–23 万条/秒 | 151–157 µs/条 ≈ 0.64–0.66 万条/秒 | +| 每消息堆分配（LengthPrefix） | 0.8 KB | 6.7 KB | 394–409 KB | + +### 裸 TCP 地板（同口径回环：预生成帧字节逐条 NetworkStream.Write，两轮区间） + +| 负载 | 地板 | StreamFrame（LengthPrefix） | 框架税（绝对 / 百分比并列） | +|---|---:|---:|---| +| 64B | 10.1–13.0 µs/条 | 4.1–4.7 µs/条 | **-6~-8 µs/条（快 48–63%）**² | +| 1KB | 9.4–16.0 µs/条 | 7.4–7.7 µs/条 | **-2~-8 µs/条（快 13–52%）**² | +| 64KB | 45–50 µs/条 | 134–188 µs/条 | **+89~138 µs/条（慢 198~276%）** | + +² 小消息下框架反而**快于**裸 NetworkStream 逐条 Write 的写法：StreamFrame 的有界队列把 +"基准线程的逐条入队"与"worker 的 socket 写出"解耦成流水线，而裸写法把每次内核写串行在 +发送线程上。框架的真实代价体现在 64KB 大消息：每条约 +0.1 ms 与 ~400 KB 分配（编码缓冲 +几何扩容 + 池租借未命中），高吞吐大报文场景是明确的优化方向（见读数要点）。 + +### 会话感知与新特性开销（三轮，1KB LengthPrefix，同上口径） + +| 对比 | 基线 | 变体 | 结论 | +|---|---:|---:|---| +| 发送方式 | SendAsync 13.2–17.2 µs/条 | SendInSessionAsync 24.7–31.6 µs/条 | **成本 ≈ +10~18 µs/条（中位 +11）**，分配 +≈470 B/条（信封 + 完成源 + 注册表；两轮区间内 SendInSession 误差棒较大，三轮方向与量级一致） | +| 接收视图 | GetMessages 13.7–15.8 µs/条 | GetSessionMessages 12.7–15.8 µs/条 | **无可测差异**（同一通道，信封解包平凡）；分配同为 ≈3.4 KB/条 | +| 未完成帧超时 | 关闭 13.5–20.9 µs/条 | 开启未触发 13.3–18.5 µs/条 | **无可测差异**（默认关闭为真零开销；开启后仅在半帧等待时才有计时令牌开销） | + +### 往返延迟（乒乓串行，两轮区间） + +| Framer | RTT | +|---|---:| +| LengthPrefix | 71.8–73.1 µs | +| StxEtx | 65.6–73.8 µs | + +### Codec（XmlDocumentCodec，历史数据） | 操作 | 报文 | 耗时 | 分配 | |---|---|---:|---:| @@ -49,20 +103,12 @@ dotnet run -c Release --project bench/StreamFrame.Benchmarks -- --filter "*EndTo | Encode | ~400B | 1.9 µs | 7.8 KB | | Encode | ~4KB | 10.9 µs | 10.4 KB | -### 端到端（真实 TCP 回环，1KB 文本消息） - -| 指标 | LengthPrefix | StxEtx | -|---|---:|---:| -| 单向吞吐（每消息折算） | 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)。 -数值仅供量级参考，请以自己环境的复现为准。 - ## 读数要点 -- **流式编码的实测收益主要是"每帧堆分配减半"（一个缓冲对象 vs 两个）**，小负载（≤ 百字节）耗时也快 25–35%； - 大负载下省掉的那次 memcpy 被缓冲增长摊销，耗时基本持平。"零拷贝"指的是帧路径不再整体搬运负载，不是字面意义的零成本。 -- **切帧吞吐相差约两个数量级**：`LengthPrefixFramer` 读长度头直接跳转（≈10 ns/帧）； - `StxEtxFramer` 在 net8+ 用 `SearchValues` 向量化定位边界（≈50 ns/帧；netstandard2.0 目标回退逐字节实现）。长度前缀在超大帧上仍占优，STX/ETX 适合既有协议兼容。 +- **框架税是分尺寸的**：小消息（≤1KB）下有界队列的流水线让 StreamFrame 反而快于"裸 NetworkStream 逐条 Write"的对照写法；大消息（64KB）真实代价显现（≈3–4×、每条 ~400 KB 分配）——高吞吐大报文是后续优化方向（编码缓冲按负载预扩容 / 池化命中率）。 +- **会话感知发送的诚实成本**：`SendInSessionAsync` 约为普通发送的 2 倍耗时、+≈470 B/条分配（信封、完成源、注册表、状态字 CAS）。对协议计时器语义（T3/T6 从整帧写出起算）而言通常值得，但高频小消息场景若不需要会话语义，用 `SendAsync`。 +- **内置指标可以常开**：无监听时 0.5–1.0 ns/次、零分配，接收/发送路径每消息合计 ≈1.6/2.6 ns。 +- **新特性默认关闭即零开销**：未完成帧超时关闭时无可测差异；接收视图切换也无成本。 +- **流式编码的实测收益主要是"每帧堆分配减半"**（一个缓冲对象 vs 两个），小负载耗时也快 24–33%；大负载下省掉的那次 memcpy 被缓冲增长摊销。"零拷贝"指帧路径不再整体搬运负载，不是字面意义的零成本。 +- **切帧吞吐**：`LengthPrefixFramer` 读长度头直接跳转（≈8.3 ns/帧）；`StxEtxFramer` 在 net8+ 用 `SearchValues` 向量化定位边界（≈50 ns/帧；netstandard2.0 目标回退逐字节实现）。 +- **回环数据的边界**：以上宏基准全部来自同机回环，仅代表同机进程通信的量级；真实网络的延迟/带宽会改变各分量的占比。复现命令见文首。 diff --git a/bench/StreamFrame.Benchmarks/EndToEndBenchmarks.cs b/bench/StreamFrame.Benchmarks/EndToEndBenchmarks.cs index e120919..e812367 100644 --- a/bench/StreamFrame.Benchmarks/EndToEndBenchmarks.cs +++ b/bench/StreamFrame.Benchmarks/EndToEndBenchmarks.cs @@ -1,126 +1,150 @@ using System.Net; using System.Net.Sockets; -using System.Threading.Channels; using BenchmarkDotNet.Attributes; namespace StreamFrame.Benchmarks; /// -/// 端到端管道基准（真实 TCP 回环）：完整连接（收发队列 + Pipe + 定界 + codec + socket）的 -/// 单向吞吐（按 framer 分组）与往返延迟。回答"整库每秒能打多少消息、一条消息来回要多久"。 +/// 端到端管道基准（真实 TCP 回环）：完整连接（收发队列 + Pipe + 定界 + codec + socket）。 +/// 原单类按关注点拆分：单向吞吐（Framer × 负载尺寸）与往返延迟（Framer）分开参数化， +/// 避免延迟基准在尺寸参数下空转重复。 /// -[MemoryDiagnoser] -[SimpleJob(warmupCount: 2, iterationCount: 10)] -public class EndToEndBenchmarks +public abstract class EndToEndBenchmarkBase { - private const int ThroughputMessages = 10_000; - private const int PingPongRounds = 2_000; - - [Params("LengthPrefix", "StxEtx")] - public string Framer { get; set; } = "LengthPrefix"; + protected const int ThroughputMessages = 10_000; + protected const int PingPongRounds = 2_000; - private StreamConnection _server = null!; - private StreamConnection _client = null!; - private CancellationTokenSource _cts = null!; + protected StreamConnection Server { get; private set; } = null!; + protected StreamConnection Client { get; private set; } = null!; + protected CancellationTokenSource Cts { get; private set; } = null!; - // 完成信号：消费端收到目标条数时置位（吞吐）/客户端收到回显时置位（乒乓） - private long _serverReceived; - private long _clientReceived; - private volatile TaskCompletionSource _serverDrained = NewTcs(); - private volatile TaskCompletionSource _clientDrained = NewTcs(); + protected long ServerReceived; + protected long ClientReceived; + protected volatile TaskCompletionSource ServerDrained = NewTcs(); + protected volatile TaskCompletionSource ClientDrained = NewTcs(); - private static TaskCompletionSource NewTcs() + protected static TaskCompletionSource NewTcs() => new(TaskCreationOptions.RunContinuationsAsynchronously); - [GlobalSetup] - public async Task Setup() + protected void SetupCore(IFramer framer) { var port = GetFreePort(); var options = new StreamConnectionOptions { ConnectRetryDelayMs = 200 }; StreamConnection Create(bool isActive) - => new( - Framer == "LengthPrefix" ? new LengthPrefixFramer() : new StxEtxFramer(), - Utf8TextCodec.Instance, - IPAddress.Loopback, - port, - isActive, - options); - - _server = Create(isActive: false); - _client = Create(isActive: true); - _cts = new CancellationTokenSource(); + => new(framer, Utf8TextCodec.Instance, IPAddress.Loopback, port, isActive, options); + + Server = Create(isActive: false); + Client = Create(isActive: true); + Cts = new CancellationTokenSource(); // 服务端：回显（乒乓用）；同时计数（吞吐用） _ = Task.Run(async () => { - await foreach (var message in _server.GetMessages(_cts.Token)) + await foreach (var message in Server.GetMessages(Cts.Token)) { - if (Interlocked.Increment(ref _serverReceived) == ThroughputMessages) - _serverDrained.TrySetResult(); - await _server.SendAsync(message, _cts.Token); + if (Interlocked.Increment(ref ServerReceived) == ThroughputMessages) + ServerDrained.TrySetResult(); + await Server.SendAsync(message, Cts.Token); } }); // 客户端：只计数收到的回显 _ = Task.Run(async () => { - await foreach (var _ in _client.GetMessages(_cts.Token)) + await foreach (var _ in Client.GetMessages(Cts.Token)) { - if (Interlocked.Increment(ref _clientReceived) == PingPongRounds) - _clientDrained.TrySetResult(); + if (Interlocked.Increment(ref ClientReceived) == PingPongRounds) + ClientDrained.TrySetResult(); } }); - _server.Start(default); - _client.Start(default); - await Task.WhenAll(_server.WaitForConnectedAsync(), _client.WaitForConnectedAsync()); + Server.Start(default); + Client.Start(default); + Task.WhenAll(Server.WaitForConnectedAsync(), Client.WaitForConnectedAsync()).GetAwaiter().GetResult(); } [GlobalCleanup] public void Cleanup() { - _cts.Cancel(); - _server.DisposeAsync().AsTask().GetAwaiter().GetResult(); - _client.DisposeAsync().AsTask().GetAwaiter().GetResult(); - _cts.Dispose(); + Cts.Cancel(); + Server.DisposeAsync().AsTask().GetAwaiter().GetResult(); + Client.DisposeAsync().AsTask().GetAwaiter().GetResult(); + Cts.Dispose(); + } + + protected static IFramer CreateFramer(string name) + => name == "LengthPrefix" ? new LengthPrefixFramer() : new StxEtxFramer(); + + protected static int PayloadLength(string size) + => size switch + { + "64B" => 64, + "64KB" => 64 * 1024, + _ => 1024, + }; + + private static int GetFreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; } +} + +/// 单向吞吐：客户端连发 1 万条消息，计到服务端全部收到。Mean 已按 OperationsPerInvoke 折算为每消息。 +[MemoryDiagnoser] +[SimpleJob(warmupCount: 2, iterationCount: 10)] +public class OneWayThroughputBenchmarks : EndToEndBenchmarkBase +{ + [Params("LengthPrefix", "StxEtx")] + public string Framer { get; set; } = "LengthPrefix"; + + [Params("64B", "1KB", "64KB")] + public string PayloadSize { get; set; } = "1KB"; + + [GlobalSetup] + public void Setup() + => SetupCore(CreateFramer(Framer)); - /// 单向吞吐：客户端连发 1 万条 1KB 消息，计到服务端全部收到。Mean 已按 OperationsPerInvoke 折算为每消息。 [Benchmark(OperationsPerInvoke = ThroughputMessages)] - public async Task OneWayThroughput_10k_1KB() + public async Task OneWayThroughput() { - Interlocked.Exchange(ref _serverReceived, 0); - _serverDrained = NewTcs(); + Interlocked.Exchange(ref ServerReceived, 0); + ServerDrained = NewTcs(); - var payload = new string('x', 1024); + var payload = new string('x', PayloadLength(PayloadSize)); for (var i = 0; i < ThroughputMessages; i++) - await _client.SendAsync(payload); + await Client.SendAsync(payload); - // 等全部送达（30 秒兜底，异常时让基准响亮失败而不是挂死） - await _serverDrained.Task.WaitAsync(TimeSpan.FromSeconds(30)); + await ServerDrained.Task.WaitAsync(TimeSpan.FromSeconds(30)); } +} + +/// 往返延迟：客户端逐条发送并等回显（串行化每一条的完整 RTT）。Mean 已折算为每次往返。 +[MemoryDiagnoser] +[SimpleJob(warmupCount: 2, iterationCount: 10)] +public class RoundTripLatencyBenchmarks : EndToEndBenchmarkBase +{ + [Params("LengthPrefix", "StxEtx")] + public string Framer { get; set; } = "LengthPrefix"; + + [GlobalSetup] + public void Setup() + => SetupCore(CreateFramer(Framer)); - /// 往返延迟：客户端逐条发送并等回显（串行化每一条的完整 RTT）。Mean 已折算为每次往返。 [Benchmark(OperationsPerInvoke = PingPongRounds)] public async Task RoundTripLatency() { - Interlocked.Exchange(ref _clientReceived, 0); - _clientDrained = NewTcs(); + Interlocked.Exchange(ref ClientReceived, 0); + ClientDrained = NewTcs(); for (var i = 0; i < PingPongRounds; i++) { - await _client.SendAsync("ping"); + await Client.SendAsync("ping"); // 串行等待这一条的回显：单条完整往返 - while (Volatile.Read(ref _clientReceived) <= i) + while (Volatile.Read(ref ClientReceived) <= i) await Task.Yield(); } } - - private static int GetFreePort() - { - var listener = new TcpListener(IPAddress.Loopback, 0); - listener.Start(); - var port = ((IPEndPoint)listener.LocalEndpoint).Port; - listener.Stop(); - return port; - } } diff --git a/bench/StreamFrame.Benchmarks/MetricsOverheadBenchmarks.cs b/bench/StreamFrame.Benchmarks/MetricsOverheadBenchmarks.cs new file mode 100644 index 0000000..969e2ff --- /dev/null +++ b/bench/StreamFrame.Benchmarks/MetricsOverheadBenchmarks.cs @@ -0,0 +1,91 @@ +using BenchmarkDotNet.Attributes; +using StreamFrame; + +namespace StreamFrame.Benchmarks; + +/// +/// 内置指标（ConnectionMetrics）的无监听开销（生产默认态：无人订阅 Meter）。 +/// 回答：每次记录到底几纳秒、单条消息的收/发路径合计加多少。 +/// +[MemoryDiagnoser] +[SimpleJob(warmupCount: 3, iterationCount: 15)] +public class MetricsOverheadBenchmarks +{ + private const int Calls = 1_000; + + // 无监听者（生产默认）：ConnectionMetrics 的记录全部走"无订阅"短路 + private readonly ConnectionMetrics _metrics = new("127.0.0.1:0"); + + [Benchmark(Baseline = true, OperationsPerInvoke = Calls)] + public void EmptyLoop_Calibration() + { + for (var i = 0; i < Calls; i++) + { + } + } + + [Benchmark(OperationsPerInvoke = Calls)] + public void FrameSent() + { + for (var i = 0; i < Calls; i++) + _metrics.FrameSent(); + } + + [Benchmark(OperationsPerInvoke = Calls)] + public void FrameReceived() + { + for (var i = 0; i < Calls; i++) + _metrics.FrameReceived(); + } + + [Benchmark(OperationsPerInvoke = Calls)] + public void AddBytesReceived_1Chunk_1028B() + { + for (var i = 0; i < Calls; i++) + _metrics.AddBytesReceived(1028); + } + + [Benchmark(OperationsPerInvoke = Calls)] + public void SendQueueObserved() + { + for (var i = 0; i < Calls; i++) + _metrics.SendQueueObserved(0); + } + + [Benchmark(OperationsPerInvoke = Calls)] + public void Reconnect() + { + for (var i = 0; i < Calls; i++) + _metrics.Reconnect(); + } + + [Benchmark(OperationsPerInvoke = Calls)] + public void SessionEnded_1s() + { + for (var i = 0; i < Calls; i++) + _metrics.SessionEnded(1.0); + } + + /// 单条消息的接收路径合计：一个字节块 + 一帧。 + [Benchmark(OperationsPerInvoke = Calls)] + public void ReceivePath_PerMessage() + { + for (var i = 0; i < Calls; i++) + { + _metrics.AddBytesReceived(1028); + _metrics.FrameReceived(); + } + } + + /// 单条消息的发送路径合计：入队采样 + 一个字节块 + 一帧。 + [Benchmark(OperationsPerInvoke = Calls)] + public void SendPath_PerMessage() + { + for (var i = 0; i < Calls; i++) + { + _metrics.SendQueueObserved(0); + _metrics.AddBytesSent(1028); + _metrics.FrameSent(); + } + } +} diff --git a/bench/StreamFrame.Benchmarks/RawTcpFloorBenchmarks.cs b/bench/StreamFrame.Benchmarks/RawTcpFloorBenchmarks.cs new file mode 100644 index 0000000..ed6947d --- /dev/null +++ b/bench/StreamFrame.Benchmarks/RawTcpFloorBenchmarks.cs @@ -0,0 +1,99 @@ +using System.Buffers.Binary; +using System.Net; +using System.Net.Sockets; +using BenchmarkDotNet.Attributes; + +namespace StreamFrame.Benchmarks; + +/// +/// 裸 TCP 回环地板值：预生成"4 字节大端长度 + 负载"的帧字节，直接在 NetworkStream 上 +/// 逐条 Write/Read（无定界、无 codec、无队列）。用于计算 StreamFrame 的框架税 +/// （绝对差值与百分比并列，口径与 OneWayThroughputBenchmarks 完全一致）。 +/// 接收循环常驻、每次调用重置计数并换完成源（与 e2e 基准同款模式——初版接收只读一轮， +/// 第二次调用起写侧无人读导致 WriteAsync 永久阻塞，已修正）。 +/// +[MemoryDiagnoser] +[SimpleJob(warmupCount: 2, iterationCount: 10)] +public class RawTcpFloorBenchmarks +{ + private const int Messages = 10_000; + + [Params("64B", "1KB", "64KB")] + public string PayloadSize { get; set; } = "1KB"; + + private TcpListener _listener = null!; + private TcpClient _sender = null!; + private TcpClient _receiver = null!; + private byte[] _frame = null!; + private long _totalBytesPerRound; + private long _receivedBytes; + private volatile TaskCompletionSource _drained = NewTcs(); + + private static TaskCompletionSource NewTcs() + => new(TaskCreationOptions.RunContinuationsAsynchronously); + + [GlobalSetup] + public async Task Setup() + { + _listener = new TcpListener(IPAddress.Loopback, 0); + _listener.Start(); + var port = ((IPEndPoint)_listener.LocalEndpoint).Port; + + _sender = new TcpClient(); + await _sender.ConnectAsync(IPAddress.Loopback, port); + _receiver = await _listener.AcceptTcpClientAsync(); + _listener.Stop(); + + var payloadLength = PayloadSize switch + { + "64B" => 64, + "64KB" => 64 * 1024, + _ => 1024, + }; + _frame = new byte[4 + payloadLength]; + BinaryPrimitives.WriteInt32BigEndian(_frame, payloadLength); + for (var i = 4; i < _frame.Length; i++) + _frame[i] = (byte)'x'; + _totalBytesPerRound = (4L + payloadLength) * Messages; + + // 常驻接收：本轮读满目标字节数即完成当前调用（计数滚动抵扣跨轮余量） + _ = Task.Run(async () => + { + var stream = _receiver.GetStream(); + var buffer = new byte[64 * 1024]; + while (true) + { + var n = await stream.ReadAsync(buffer); + if (n == 0) + return; + if (Interlocked.Add(ref _receivedBytes, n) >= Volatile.Read(ref _totalBytesPerRound)) + { + Interlocked.Add(ref _receivedBytes, -Volatile.Read(ref _totalBytesPerRound)); + _drained.TrySetResult(); + } + } + }); + } + + [GlobalCleanup] + public void Cleanup() + { + _sender.Dispose(); + _receiver.Dispose(); + _listener.Stop(); + } + + /// 单向吞吐地板：连发 1 万条帧字节。Mean 已按 OperationsPerInvoke 折算为每消息。 + [Benchmark(OperationsPerInvoke = Messages)] + public async Task OneWayThroughput() + { + Interlocked.Exchange(ref _receivedBytes, 0); + _drained = NewTcs(); + + var stream = _sender.GetStream(); + for (var i = 0; i < Messages; i++) + await stream.WriteAsync(_frame); + + await _drained.Task.WaitAsync(TimeSpan.FromSeconds(30)); + } +} diff --git a/bench/StreamFrame.Benchmarks/SessionAwareBenchmarks.cs b/bench/StreamFrame.Benchmarks/SessionAwareBenchmarks.cs new file mode 100644 index 0000000..23727e9 --- /dev/null +++ b/bench/StreamFrame.Benchmarks/SessionAwareBenchmarks.cs @@ -0,0 +1,191 @@ +using System.Net; +using System.Net.Sockets; +using BenchmarkDotNet.Attributes; +using StreamFrame; + +namespace StreamFrame.Benchmarks; + +/// +/// 会话感知/新特性开销的端到端量化（LengthPrefix + 1KB，与单向吞吐同口径）： +/// ① SendInSessionAsync vs SendAsync（信封 + 完成源 + 注册表的每消息成本）； +/// ② GetSessionMessages vs GetMessages（接收视图的成本）； +/// ③ 未完成帧超时 关闭 vs 开启未触发（默认关闭路径是否零开销）。 +/// 三个关注点拆成三个基准类，避免参数交叉积膨胀。 +/// +public abstract class SessionAwareBenchmarkBase +{ + protected const int Messages = 10_000; + + protected StreamConnection Server { get; private set; } = null!; + protected StreamConnection Client { get; private set; } = null!; + protected CancellationTokenSource Cts { get; private set; } = null!; + protected long ServerReceived; + protected volatile TaskCompletionSource ServerDrained = NewTcs(); + + protected static TaskCompletionSource NewTcs() + => new(TaskCreationOptions.RunContinuationsAsynchronously); + + protected Task RunServerLoop(Func> streamFactory) + => Task.Run(async () => + { + await foreach (var _ in streamFactory()) + { + if (Interlocked.Increment(ref ServerReceived) == Messages) + ServerDrained.TrySetResult(); + } + }); + + protected void SetupCore(StreamConnectionOptions options, Func> streamFactory) + { + var port = GetFreePort(); + StreamConnection Create(bool isActive) + => new( + new LengthPrefixFramer(), + Utf8TextCodec.Instance, + IPAddress.Loopback, + port, + isActive, + options); + + Server = Create(isActive: false); + Client = Create(isActive: true); + Cts = new CancellationTokenSource(); + + _ = RunServerLoop(streamFactory); + + Server.Start(default); + Client.Start(default); + Task.WhenAll(Server.WaitForConnectedAsync(), Client.WaitForConnectedAsync()).GetAwaiter().GetResult(); + } + + [GlobalCleanup] + public void Cleanup() + { + Cts.Cancel(); + Server.DisposeAsync().AsTask().GetAwaiter().GetResult(); + Client.DisposeAsync().AsTask().GetAwaiter().GetResult(); + Cts.Dispose(); + } + + protected static int GetFreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } +} + +/// ① 发送方式：会话绑定发送 vs 普通发送（服务端经 GetMessages 消费）。 +[MemoryDiagnoser] +[SimpleJob(warmupCount: 2, iterationCount: 10)] +public class SessionAwareSendBenchmarks : SessionAwareBenchmarkBase +{ + private long _sessionId; + + [Params("SendAsync", "SendInSessionAsync")] + public string SendMode { get; set; } = "SendAsync"; + + [GlobalSetup] + public void Setup() + => SetupCore(new StreamConnectionOptions { ConnectRetryDelayMs = 200 }, () => Server.GetMessages(Cts.Token)); + + /// 单向吞吐：连发 1 万条 1KB。Mean 已折算为每消息。 + [Benchmark(OperationsPerInvoke = Messages)] + public async Task OneWayThroughput() + { + Interlocked.Exchange(ref ServerReceived, 0); + ServerDrained = NewTcs(); + + var payload = new string('x', 1024); + if (SendMode == "SendInSessionAsync") + { + _sessionId = ((ISessionAwareStreamConnection)Client).CurrentSessionId; + for (var i = 0; i < Messages; i++) + await Client.SendInSessionAsync(_sessionId, payload); + } + else + { + for (var i = 0; i < Messages; i++) + await Client.SendAsync(payload); + } + + await ServerDrained.Task.WaitAsync(TimeSpan.FromSeconds(30)); + } +} + +/// ② 接收视图：GetSessionMessages vs GetMessages（普通发送驱动）。 +[MemoryDiagnoser] +[SimpleJob(warmupCount: 2, iterationCount: 10)] +public class SessionAwareReceiveBenchmarks : SessionAwareBenchmarkBase +{ + [Params("GetMessages", "GetSessionMessages")] + public string ReceiveView { get; set; } = "GetMessages"; + + [GlobalSetup] + public void Setup() + => SetupCore( + new StreamConnectionOptions { ConnectRetryDelayMs = 200 }, + () => ReceiveView == "GetMessages" + ? Server.GetMessages(Cts.Token) + : ((ISessionAwareStreamConnection)Server).GetSessionMessages(Cts.Token).Select(m => m.Message)); + + /// 单向吞吐：连发 1 万条 1KB。Mean 已折算为每消息。 + [Benchmark(OperationsPerInvoke = Messages)] + public async Task OneWayThroughput() + { + Interlocked.Exchange(ref ServerReceived, 0); + ServerDrained = NewTcs(); + + var payload = new string('x', 1024); + for (var i = 0; i < Messages; i++) + await Client.SendAsync(payload); + + await ServerDrained.Task.WaitAsync(TimeSpan.FromSeconds(30)); + } +} + +/// ③ 未完成帧超时：关闭（默认） vs 开启未触发（500ms，帧内不会超时）。 +[MemoryDiagnoser] +[SimpleJob(warmupCount: 2, iterationCount: 10)] +public class IncompleteFrameTimeoutBenchmarks : SessionAwareBenchmarkBase +{ + [Params("Off", "On_Untriggered")] + public string TimeoutMode { get; set; } = "Off"; + + [GlobalSetup] + public void Setup() + => SetupCore( + new StreamConnectionOptions + { + ConnectRetryDelayMs = 200, + IncompleteFrameTimeoutMs = TimeoutMode == "On_Untriggered" ? 500 : 0, + }, + () => Server.GetMessages(Cts.Token)); + + /// 单向吞吐：连发 1 万条 1KB。Mean 已折算为每消息。 + [Benchmark(OperationsPerInvoke = Messages)] + public async Task OneWayThroughput() + { + Interlocked.Exchange(ref ServerReceived, 0); + ServerDrained = NewTcs(); + + var payload = new string('x', 1024); + for (var i = 0; i < Messages; i++) + await Client.SendAsync(payload); + + await ServerDrained.Task.WaitAsync(TimeSpan.FromSeconds(30)); + } +} + +file static class AsyncEnumerableLinq +{ + // System.Linq.Async 依赖避免：仅用于把会话视图投影回消息流 + public static async IAsyncEnumerable Select( + this IAsyncEnumerable source, Func selector) + { + await foreach (var item in source) + yield return selector(item); + } +}