diff --git a/CHANGELOG.md b/CHANGELOG.md index bbfc83d..1fde7cb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,9 @@ - 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，防御性）**：排查确认原始"用户 `Reconnect()` 竞速楔死监听"**无法复现**（顺序/并发两组确定性复现 + 混沌套件恢复该动作后多轮全绿，此前观察到的楔死系当时测试自身缺陷），但修复排查中识别的两个真实弱点——① 接受循环的重试延迟改为锁外等待（旧实现整个重试循环持有 `_acceptLock`，绑定/接受持续失败期间会堵死其它接受循环）；② 监听 socket 设置 `SO_REUSEADDR`（服务端主动关闭后立即重绑不受 2MSL/TIME_WAIT 限制；Windows 实测宽松，Linux 上无此选项会遇 `EADDRINUSE`，一并设置保持跨平台一致）。新增确定性回归测试（8 轮服务端主动重连，端口须秒级可重新接受），混沌套件恢复用户重连动作。 diff --git a/README.en.md b/README.en.md index 4caade5..49b1bf5 100644 --- a/README.en.md +++ b/README.en.md @@ -232,6 +232,22 @@ Measured with BenchmarkDotNet (details and how to reproduce in [bench/README.md] - **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). +### Large-message guide (≥64KB) + +For 64KB-class messages the dominant costs are the codec style and the message type, not the framework (attribution data in [bench/README.md](bench/README.md)). Three recommendations: + +1. **Use the span-based write overload in your codec** — `Encoding.GetBytes(ReadOnlySpan, IBufferWriter)` produces no intermediate array (the naive style adds a full-size allocation + copy per message): + + ```csharp + // Recommended: zero intermediate arrays + public void Encode(string message, IBufferWriter writer, CancellationToken ct) + => Encoding.UTF8.GetBytes(message.AsSpan(), writer); + // Not recommended: writer.Write(Encoding.UTF8.GetBytes(message)); // full-size byte[] per message + ``` + +2. **Pick byte[] / ReadOnlyMemory as the message type**: string messages inherently allocate ≈2× the payload size per message for UTF-16 materialization; with byte payloads and a pass-through codec the framework sits within ≈20–30% of raw TCP; +3. Keep **streaming encode** on (default, `UseStreamingEncode`) — send buffers now start at the previous frame's size (adaptive, capped at 1MB). + ## Supported frameworks | Package target | Runtime | diff --git a/README.md b/README.md index b5b555f..a5e2c54 100644 --- a/README.md +++ b/README.md @@ -229,6 +229,22 @@ BenchmarkDotNet 实测（详见 [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)。 +### 大报文指南（≥64KB） + +64KB 级消息下，开销大头在 codec 写法与消息类型而非框架（归因数据见 [bench/README.md](bench/README.md)）。三条建议： + +1. **codec 用 span 直写重载**——`Encoding.GetBytes(ReadOnlySpan, IBufferWriter)` 不产生中间数组（旧写法每条多一次全尺寸分配 + 拷贝）： + + ```csharp + // 推荐：零中间数组 + public void Encode(string message, IBufferWriter writer, CancellationToken ct) + => Encoding.UTF8.GetBytes(message.AsSpan(), writer); + // 不推荐：writer.Write(Encoding.UTF8.GetBytes(message)); // 每条一次全尺寸 byte[] + ``` + +2. **消息类型选 byte[] / ReadOnlyMemory**：string 消息每条固有 ≈2× 报文大小的 UTF-16 物化分配；字节负载 + 透传 codec 时框架距裸 TCP 仅 ≈20–30%； +3. 保持默认的**流式编码**开启（`UseStreamingEncode`），发送缓冲会按上一帧大小自适应起租（封顶 1MB）。 + ## 支持框架 | 包目标 | 运行环境 | diff --git a/bench/README.md b/bench/README.md index 0d90f2b..2dfc00a 100644 --- a/bench/README.md +++ b/bench/README.md @@ -62,8 +62,8 @@ dotnet run -c Release --project bench/StreamFrame.Benchmarks -- --filter "*EndTo | 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 | 4.1–4.7 µs/条 ≈ 21–24 万条/秒 | 4.1–7.7 µs/条 ≈ 13–24 万条/秒 | 134–188 µs/条 ≈ 0.53–0.75 万条/秒 | +| StxEtx | 2.2–6.9 µs/条 ≈ 15–46 万条/秒 | 3.8–8.7 µs/条 ≈ 12–26 万条/秒 | 127–157 µs/条 ≈ 0.64–0.79 万条/秒 | | 每消息堆分配（LengthPrefix） | 0.8 KB | 6.7 KB | 394–409 KB | ### 裸 TCP 地板（同口径回环：预生成帧字节逐条 NetworkStream.Write，两轮区间） @@ -72,12 +72,23 @@ dotnet run -c Release --project bench/StreamFrame.Benchmarks -- --filter "*EndTo |---|---:|---:|---| | 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%）** | +| 64KB | 45–50 µs/条 | 134–188 µs/条³ | **+89~138 µs/条（慢 198~276%）**³ | ² 小消息下框架反而**快于**裸 NetworkStream 逐条 Write 的写法：StreamFrame 的有界队列把 "基准线程的逐条入队"与"worker 的 socket 写出"解耦成流水线，而裸写法把每次内核写串行在 发送线程上。框架的真实代价体现在 64KB 大消息：每条约 +0.1 ms 与 ~400 KB 分配（编码缓冲 -几何扩容 + 池租借未命中），高吞吐大报文场景是明确的优化方向（见读数要点）。 +几何扩容 + 池租借未命中）。³ 该行使用"分配式 GetBytes 的字符串 codec + 服务端回显"，归因表明其中大头是 codec 与消息类型成本（见"大报文归因"）：byte[] 负载 + 透传 codec 时框架距地板仅 59–73 µs（≈20–45%）。 + +### 大报文归因（64KB，计数消费无回显，2026-08-29，含自适应发送缓冲） + +| 变体 | 每消息耗时 | 每消息分配 | 归因 | +|---|---:|---:|---| +| byte[] 消息 + 透传 codec | 59–73 µs | 64 KB | **框架纯成本基线**：分配仅为解码侧 ToArray（codec 固有），框架自身零分配；耗时较裸 TCP 地板（45–50 µs）高 ≈20–30% | +| string 消息 + span 直写 codec | 70–104 µs（误差大，见噪声披露） | 129 KB | +64 KB = **消息类型固有税**（UTF-8 → UTF-16 string 物化， unavoidable for TMessage=string） | +| string 消息 + 分配中间数组的 GetBytes（旧写法） | 73–92 µs | 193 KB | 再 +65 KB = **codec 写法税**（GetBytes 中间数组，可用 span 重载消除） | + +结论：64KB 场景的"框架税"大部分是 codec 写法与消息类型的成本——byte[] 负载 + 透传/低拷贝 +codec 时框架距裸 TCP 仅 ≈20–30%。使用建议见 README"大报文指南"。 ### 会话感知与新特性开销（三轮，1KB LengthPrefix，同上口径） diff --git a/bench/StreamFrame.Benchmarks/LargeMessageBenchmarks.cs b/bench/StreamFrame.Benchmarks/LargeMessageBenchmarks.cs new file mode 100644 index 0000000..9e134f5 --- /dev/null +++ b/bench/StreamFrame.Benchmarks/LargeMessageBenchmarks.cs @@ -0,0 +1,148 @@ +using System.Buffers; +using System.Net; +using System.Net.Sockets; +using System.Text; +using BenchmarkDotNet.Attributes; +using StreamFrame; + +namespace StreamFrame.Benchmarks; + +/// +/// 64KB 大报文的分配归因基准：同一端到端管道（LengthPrefix + 计数消费，无回显）在三种 +/// codec/消息类型下的表现—— +/// ① String_Alloc：字符串 + 分配中间数组的 GetBytes（现状写法，demo/旧基准同款）； +/// ② String_Span：字符串 + span 直写（Encoding.GetBytes(ReadOnlySpan<char>, IBufferWriter)，零中间数组）； +/// ③ ByteArray：byte[] 消息 + 透传 codec（编码零拷贝、解码一次 ToArray）。 +/// 用于把"框架税"精确拆到 框架 / codec 写法 / 消息类型 三个口袋。 +/// +[MemoryDiagnoser] +[SimpleJob(warmupCount: 2, iterationCount: 10)] +public class LargeMessageStringBenchmarks : SessionAwareBenchmarkBase +{ + [Params("String_Alloc", "String_Span")] + public string CodecMode { get; set; } = "String_Alloc"; + + [GlobalSetup] + public void Setup() + => SetupCore( + new StreamConnectionOptions { ConnectRetryDelayMs = 200 }, + () => Server.GetMessages(Cts.Token), + CodecMode == "String_Span" ? SpanUtf8TextCodec.Instance : Utf8TextCodec.Instance); + + /// 单向吞吐：连发 1 万条 64KB。Mean 已折算为每消息。 + [Benchmark(OperationsPerInvoke = Messages)] + public async Task OneWayThroughput_64KB() + { + Interlocked.Exchange(ref ServerReceived, 0); + ServerDrained = NewTcs(); + + var payload = new string('x', 64 * 1024); + for (var i = 0; i < Messages; i++) + await Client.SendAsync(payload); + + await ServerDrained.Task.WaitAsync(TimeSpan.FromSeconds(60)); + } +} + +/// byte[] 消息 + 透传 codec 的对照：编码零拷贝（同一数组实例发一万次），解码一次 ToArray。 +[MemoryDiagnoser] +[SimpleJob(warmupCount: 2, iterationCount: 10)] +public class LargeMessageByteArrayBenchmarks +{ + private const int Messages = 10_000; + + [Params("64KB", "1KB")] + public string PayloadSize { get; set; } = "64KB"; + + private StreamConnection _server = null!; + private StreamConnection _client = null!; + private CancellationTokenSource _cts = null!; + private byte[] _payload = null!; + private long _received; + private volatile TaskCompletionSource _drained = NewTcs(); + + private static TaskCompletionSource NewTcs() + => new(TaskCreationOptions.RunContinuationsAsynchronously); + + [GlobalSetup] + public void Setup() + { + var port = GetFreePort(); + var options = new StreamConnectionOptions { ConnectRetryDelayMs = 200 }; + StreamConnection Create(bool isActive) + => new(new LengthPrefixFramer(), ByteArrayPassThroughCodec.Instance, IPAddress.Loopback, port, isActive, options); + + _server = Create(isActive: false); + _client = Create(isActive: true); + _cts = new CancellationTokenSource(); + _payload = new byte[PayloadSize == "1KB" ? 1024 : 64 * 1024]; + + _ = Task.Run(async () => + { + await foreach (var _ in _server.GetMessages(_cts.Token)) + { + if (Interlocked.Increment(ref _received) == Messages) + _drained.TrySetResult(); + } + }); + + _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(); + } + + /// 单向吞吐：连发 1 万条（同一 byte[] 实例）。Mean 已折算为每消息。 + [Benchmark(OperationsPerInvoke = Messages)] + public async Task OneWayThroughput() + { + Interlocked.Exchange(ref _received, 0); + _drained = NewTcs(); + + for (var i = 0; i < Messages; i++) + await _client.SendAsync(_payload); + + await _drained.Task.WaitAsync(TimeSpan.FromSeconds(60)); + } + + private static int GetFreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } +} + +/// span 直写的 UTF-8 文本 codec：Encode 不产生中间数组（大报文指南的推荐写法）。 +internal sealed class SpanUtf8TextCodec : ICodec +{ + public static readonly SpanUtf8TextCodec Instance = new(); + + public string Decode(in ReadOnlySequence frame, CancellationToken ct = default) + => Encoding.UTF8.GetString(frame); + + public void Encode(string message, IBufferWriter writer, CancellationToken ct = default) + => Encoding.UTF8.GetBytes(message.AsSpan(), writer); +} + +/// byte[] 透传 codec：编码原样写入（零拷贝），解码一次 ToArray。 +internal sealed class ByteArrayPassThroughCodec : ICodec +{ + public static readonly ByteArrayPassThroughCodec Instance = new(); + + public byte[] Decode(in ReadOnlySequence frame, CancellationToken ct = default) + => frame.ToArray(); + + public void Encode(byte[] message, IBufferWriter writer, CancellationToken ct = default) + => writer.Write(message); +} diff --git a/bench/StreamFrame.Benchmarks/SessionAwareBenchmarks.cs b/bench/StreamFrame.Benchmarks/SessionAwareBenchmarks.cs index 23727e9..07900fe 100644 --- a/bench/StreamFrame.Benchmarks/SessionAwareBenchmarks.cs +++ b/bench/StreamFrame.Benchmarks/SessionAwareBenchmarks.cs @@ -35,13 +35,13 @@ protected Task RunServerLoop(Func> streamFactory) } }); - protected void SetupCore(StreamConnectionOptions options, Func> streamFactory) + protected void SetupCore(StreamConnectionOptions options, Func> streamFactory, ICodec? codec = null) { var port = GetFreePort(); StreamConnection Create(bool isActive) => new( new LengthPrefixFramer(), - Utf8TextCodec.Instance, + codec ?? Utf8TextCodec.Instance, IPAddress.Loopback, port, isActive, diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs index e0b58a9..5fd7e4e 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -771,27 +771,59 @@ private async Task SendWorkerAsync(CancellationToken ct, long sessionId) } } + /// + /// 发送编码缓冲的自适应初始尺寸（高水位记忆）：稳态消息尺寸相近的协议（设备周期上报） + /// 直接按上一帧大小起租，消掉从 EncodeBufferInitialSize（默认 1KB）起的几何增长梯子—— + /// 大报文场景的反复跨桶租借与其中间拷贝显著减少。发送 worker 单线程访问，无需同步。 + /// + private int _encodeBufferHighWater; + + private int AdaptiveEncodeBufferSize + { + get + { + // netstandard2.0 无 Math.Clamp：显式收拢到 [初始配置, 1MB] + var size = _encodeBufferHighWater; + if (size < _options.EncodeBufferInitialSize) + size = _options.EncodeBufferInitialSize; + if (size > MaxAdaptiveEncodeBufferBytes) + size = MaxAdaptiveEncodeBufferBytes; + return size; + } + } + + /// 自适应缓冲封顶：超过此值回退到配置初始值（防御异常尺寸的偶发大帧抬高常态水位）。 + private const int MaxAdaptiveEncodeBufferBytes = 1024 * 1024; + private async Task SendFramedAsync(TMessage message, CancellationToken ct) { // 流式编码：单缓冲、零 memcpy（BeginFrame 占位 → codec 直接写 → EndFrame 回填/收尾） if (_options.UseStreamingEncode && _framing is IStreamingFramer streaming) { - using var frame = new PooledBufferWriter(_options.EncodeBufferInitialSize); + using var frame = new PooledBufferWriter(AdaptiveEncodeBufferSize); streaming.BeginFrame(frame); _codec.Encode(message, frame, ct); streaming.EndFrame(frame); + RememberEncodeHighWater(frame.WrittenCount); await SendRawAsync(frame.WrittenMemory, ct).ConfigureAwait(false); return; } // 纯函数编码：序列化产物 → 帧，两段缓冲（含一次 memcpy） - using var payload = new PooledBufferWriter(_options.EncodeBufferInitialSize); + using var payload = new PooledBufferWriter(AdaptiveEncodeBufferSize); _codec.Encode(message, payload, ct); using var frame2 = new PooledBufferWriter(payload.WrittenCount + 16); _framing.EncodeFrame(payload.WrittenSpan, frame2); + RememberEncodeHighWater(frame2.WrittenCount); await SendRawAsync(frame2.WrittenMemory, ct).ConfigureAwait(false); } + private void RememberEncodeHighWater(int frameBytes) + { + if (frameBytes > _encodeBufferHighWater && frameBytes <= MaxAdaptiveEncodeBufferBytes) + _encodeBufferHighWater = frameBytes; + } + private async Task SendRawAsync(ReadOnlyMemory buffer, CancellationToken ct) { await _sendLock.WaitAsync(ct).ConfigureAwait(false);