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
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 轮服务端主动重连,端口须秒级可重新接受),混沌套件恢复用户重连动作。

Expand Down
16 changes: 16 additions & 0 deletions README.en.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<char>, IBufferWriter<byte>)` 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<byte> 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<byte> 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 |
Expand Down
16 changes: 16 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<char>, IBufferWriter<byte>)` 不产生中间数组(旧写法每条多一次全尺寸分配 + 拷贝):

```csharp
// 推荐:零中间数组
public void Encode(string message, IBufferWriter<byte> writer, CancellationToken ct)
=> Encoding.UTF8.GetBytes(message.AsSpan(), writer);
// 不推荐:writer.Write(Encoding.UTF8.GetBytes(message)); // 每条一次全尺寸 byte[]
```

2. **消息类型选 byte[] / ReadOnlyMemory<byte>**:string 消息每条固有 ≈2× 报文大小的 UTF-16 物化分配;字节负载 + 透传 codec 时框架距裸 TCP 仅 ≈20–30%;
3. 保持默认的**流式编码**开启(`UseStreamingEncode`),发送缓冲会按上一帧大小自适应起租(封顶 1MB)。

## 支持框架

| 包目标 | 运行环境 |
Expand Down
19 changes: 15 additions & 4 deletions bench/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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,两轮区间)
Expand All @@ -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,同上口径)

Expand Down
148 changes: 148 additions & 0 deletions bench/StreamFrame.Benchmarks/LargeMessageBenchmarks.cs
Original file line number Diff line number Diff line change
@@ -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;

/// <summary>
/// 64KB 大报文的分配归因基准:同一端到端管道(LengthPrefix + 计数消费,无回显)在三种
/// codec/消息类型下的表现——
/// ① String_Alloc:字符串 + 分配中间数组的 GetBytes(现状写法,demo/旧基准同款);
/// ② String_Span:字符串 + span 直写(Encoding.GetBytes(ReadOnlySpan&lt;char&gt;, IBufferWriter),零中间数组);
/// ③ ByteArray:byte[] 消息 + 透传 codec(编码零拷贝、解码一次 ToArray)。
/// 用于把"框架税"精确拆到 框架 / codec 写法 / 消息类型 三个口袋。
/// </summary>
[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);

/// <summary>单向吞吐:连发 1 万条 64KB。Mean 已折算为每消息。</summary>
[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));
}
}

/// <summary>byte[] 消息 + 透传 codec 的对照:编码零拷贝(同一数组实例发一万次),解码一次 ToArray。</summary>
[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<byte[]> _server = null!;
private StreamConnection<byte[]> _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<byte[]> 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();
}

/// <summary>单向吞吐:连发 1 万条(同一 byte[] 实例)。Mean 已折算为每消息。</summary>
[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;
}
}

/// <summary>span 直写的 UTF-8 文本 codec:Encode 不产生中间数组(大报文指南的推荐写法)。</summary>
internal sealed class SpanUtf8TextCodec : ICodec<string>
{
public static readonly SpanUtf8TextCodec Instance = new();

public string Decode(in ReadOnlySequence<byte> frame, CancellationToken ct = default)
=> Encoding.UTF8.GetString(frame);

public void Encode(string message, IBufferWriter<byte> writer, CancellationToken ct = default)
=> Encoding.UTF8.GetBytes(message.AsSpan(), writer);
}

/// <summary>byte[] 透传 codec:编码原样写入(零拷贝),解码一次 ToArray。</summary>
internal sealed class ByteArrayPassThroughCodec : ICodec<byte[]>
{
public static readonly ByteArrayPassThroughCodec Instance = new();

public byte[] Decode(in ReadOnlySequence<byte> frame, CancellationToken ct = default)
=> frame.ToArray();

public void Encode(byte[] message, IBufferWriter<byte> writer, CancellationToken ct = default)
=> writer.Write(message);
}
4 changes: 2 additions & 2 deletions bench/StreamFrame.Benchmarks/SessionAwareBenchmarks.cs
Original file line number Diff line number Diff line change
Expand Up @@ -35,13 +35,13 @@ protected Task RunServerLoop(Func<IAsyncEnumerable<string>> streamFactory)
}
});

protected void SetupCore(StreamConnectionOptions options, Func<IAsyncEnumerable<string>> streamFactory)
protected void SetupCore(StreamConnectionOptions options, Func<IAsyncEnumerable<string>> streamFactory, ICodec<string>? codec = null)
{
var port = GetFreePort();
StreamConnection<string> Create(bool isActive)
=> new(
new LengthPrefixFramer(),
Utf8TextCodec.Instance,
codec ?? Utf8TextCodec.Instance,
IPAddress.Loopback,
port,
isActive,
Expand Down
36 changes: 34 additions & 2 deletions src/StreamFrame/Connection/StreamConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -771,27 +771,59 @@ private async Task SendWorkerAsync(CancellationToken ct, long sessionId)
}
}

/// <summary>
/// 发送编码缓冲的自适应初始尺寸(高水位记忆):稳态消息尺寸相近的协议(设备周期上报)
/// 直接按上一帧大小起租,消掉从 EncodeBufferInitialSize(默认 1KB)起的几何增长梯子——
/// 大报文场景的反复跨桶租借与其中间拷贝显著减少。发送 worker 单线程访问,无需同步。
/// </summary>
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;
}
}

/// <summary>自适应缓冲封顶:超过此值回退到配置初始值(防御异常尺寸的偶发大帧抬高常态水位)。</summary>
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<byte> buffer, CancellationToken ct)
{
await _sendLock.WaitAsync(ct).ConfigureAwait(false);
Expand Down
Loading