diff --git a/CHANGELOG.md b/CHANGELOG.md index 8274541..82f572d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ - **Soak 目标矩阵（[#66](https://github.com/CSJ608/StreamFrame/issues/66)）**：显式选择 Ubuntu net8.0/net10.0、Windows net8.0/net10.0/net48，避免 Ubuntu 启动 .NET Framework 测试；job 名、详细日志与始终尝试上传的 TRX 产物按 OS/TFM 区分，测试断言失败继续使 job 失败。普通 CI 必需检查名不变。 ### 变更 +- **驱动所有权与并发文档（[#69](https://github.com/CSJ608/StreamFrame/issues/69)）**：补全 ICodec/IFramer/流式定界的缓冲借用及同连接收发、新旧会话、跨连接共享约束；双语 README 增加拥有独立数据的 byte[]/ReadOnlyMemory 示例、有界队列生产配置及单客户端边界，澄清终态等待、KeepAlive 粒度和入队非确认。示例纳入三目标编译并以主动覆写缓冲验证异步消费；不改变默认值或实现公共契约。 - **Soak 可观测协议（[#67](https://github.com/CSJ608/StreamFrame/issues/67)）**：保持 SendAsync 仅入队契约，记录可重放 seed、动作、尝试 ID、会话及发送/读取阶段；业务测试使用 ACK、重试、去重并精确校验，重试前独立检查队列续发与会话边界。所有阶段有界等待，未启用长测报告 Skip；完整 600 秒五目标连续三次验收另行跟踪，见 docs/SOAK.md。 ## [2.6.0] - 2026-08-29 diff --git a/README.en.md b/README.en.md index db031dd..8b05c09 100644 --- a/README.en.md +++ b/README.en.md @@ -42,7 +42,10 @@ sealed class SystemTextJsonCodec : ICodec public static readonly SystemTextJsonCodec Instance = new(); public JsonElement Decode(in ReadOnlySequence frame, CancellationToken ct = default) - => JsonDocument.Parse(frame.ToArray()).RootElement; + { + using var document = JsonDocument.Parse(frame.ToArray()); + return document.RootElement.Clone(); // Own data beyond document disposal. + } public void Encode(JsonElement message, IBufferWriter writer, CancellationToken ct = default) { @@ -55,6 +58,8 @@ sealed class SystemTextJsonCodec : ICodec ## Quick start +This is a **single-connection device integration library**. A passive listener serves one client at a time; `AcceptFirstClientOnly = false` keeps the listener open without adding multi-client handling. The snippets below represent separate processes/roles, with a host-provided `ct`. See the production configuration below and dispose connections with `await using`. + 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. ```csharp @@ -113,7 +118,7 @@ public interface ICodec - **Client/server dual mode**: `isActive: true` connects out, `false` listens; IPv4/IPv6 dual-stack (listening on `IPAddress.Any` is normalized to dual-stack; IPv4 client addresses are displayed as plain IPv4) - **Auto-reconnect**: `Connecting → Connected → Retry` state machine; optional exponential backoff (`MaxRetryDelayMs`, doubling with cap + ±20% jitter, auto-reset on success); `GetMessages` is a stable stream across reconnections — received messages are not lost and enumeration does not break when the connection drops and recovers - **Start & stop**: the `ct` passed to `Start(ct)` is the connection's **lifetime token** — cancelling it stops connection/reconnection and tears the link down (state enters the terminal `Disconnected`, `GetMessages` completes naturally; create a new connection afterwards). `DisposeAsync` does the same. After shutdown, `SendAsync` throws `ChannelClosedException` -- **Waiting for readiness**: `await conn.WaitForConnectedAsync(ct)` — completes immediately when connected; otherwise waits for the next successful connection (or cancellation/dispose). No state polling, no `Task.Delay` guessing +- **Waiting for readiness**: `await conn.WaitForConnectedAsync(ct)` — completes immediately when connected; otherwise waits for the next successful connection (or caller cancellation, lifetime cancellation/disposal). Future waits after terminal shutdown are cancelled too, even without a caller token - **Robustness**: payload decode failures, incomplete-frame overflows/timeouts, send failures, and receive idle timeouts all invalidate the session and rebuild it automatically (no more "connection looks alive while messages silently vanish") - **Liveness detection (optional)**: TCP KeepAlive and receive idle timeout to catch half-open connections (peer power loss / unplugged cable) - **Events**: `ConnectionChanged` state changes, `FrameError` frame-level diagnostics, `RawBytesReceived/Sent` raw bytes (HEX debugging) @@ -121,6 +126,44 @@ public interface ICodec - **Backpressure**: bounded send queue — `SendAsync` waits when full; the receive side buffers without limit by default, or set `ReceiveQueueCapacity` — decoding pauses when the consumer lags, propagating TCP backpressure to the peer and preventing unbounded memory growth - **Session-aware messaging (optional, advanced)**: `CurrentSessionId` / `SendInSessionAsync` (completes only after the whole frame is written to the socket; fails on session loss and never replays across sessions) / `GetSessionMessages` (messages carry their session id) — for protocols with strict session boundaries such as HSMS; see the dedicated section below +## Buffer ownership and concurrency + +`Decode` borrows `frame` only during the synchronous call. The returned message, including nested fields, must independently own retained data. Never return `frame.First`, slices or `ReadOnlyMemory` backed by pipeline storage: the pipe advances/releases its buffers, while business processing may happen after an `await`. Read-only memory does not imply ownership. This interface offers no transfer of pooled pipeline storage. + +`Encode` must finish writing before returning. Do not retain/dispose `writer` or retain its buffers for asynchronous use. The send queue holds message objects: do not mutate or return their backing arrays after enqueue. Plain `SendAsync` completion means enqueue only; encoding may not have started. A dedicated array that is never modified afterwards is the simplest approach. + +A session's send worker serializes its own encoding calls, **not the codec instance**. `Encode` and `Decode` may run concurrently. Session teardown waits at most two seconds: old decoders (for example, stalled by receive backpressure) may overlap new session decoders; a blocked old encoder cannot be assumed to have exited either. Use call-local state or synchronization and avoid blocking synchronous methods. Callers sharing a codec/framer across connections must ensure concurrency safety. One instance per connection still needs to support concurrent read/write and old/new session calls. + +The same rules apply to `IFramer`: do not retain input sequences; output `payload` may borrow input for the immediately following synchronous codec call. Use the encoding writer only within the call. Other calls may interleave between `IStreamingFramer.BeginFrame` and `EndFrame`; keep pair state in the supplied writer rather than an instance's "current frame" field. + +Complete, compilable [`OwnedBytesCodec`](samples/StreamFrame.Ownership/OwnedBytesCodec.cs) / [`OwnedMemoryCodec`](samples/StreamFrame.Ownership/OwnedMemoryCodec.cs) examples return `frame.ToArray()` for both message types, including single-segment input. Their encoders synchronously call `writer.Write(message)` or `writer.Write(message.Span)`. These exact sources are compiled in the test project for net8.0/net10.0/net48. `OwnershipExampleTests` overwrites single/multi-segment input and runs the real FrameDecoder with a pool that fills returned storage with `0xDD`; asynchronous consumption after pipe disposal must still produce the original bytes. + +## Production configuration starting point + +The complete [`ProductionExample.RunAsync`](samples/StreamFrame.Ownership/ProductionExample.cs) demonstrates bounded queues, a ten-second readiness timeout, asynchronous consumption, lifetime cancellation and `await using` disposal, and is compiled for all three test targets. These values assume payloads up to 64 KiB: tune them to the protocol, peak traffic and processing time. Library defaults remain unchanged. + +```csharp +var framer = new LengthPrefixFramer(64 * 1024); +var options = new StreamConnectionOptions +{ + SendQueueCapacity = 128, + ReceiveQueueCapacity = 128, // Default 0 is unbounded. + MaxIncompleteFrameBufferBytes = 64 * 1024 + 4, // Includes length header. + IncompleteFrameTimeoutMs = 5_000, + ReceiveIdleTimeoutMs = 0, // Silence is legal; consider 15_000 with a 5s heartbeat. + TcpKeepAlive = true, + KeepAliveTimeMs = 30_000, + KeepAliveIntervalMs = 1_500, // Modern .NET rounds up to 2 seconds (#62). + AcceptFirstClientOnly = true, +}; +``` + +Capacity counts **messages**, not bytes: 128 × 64 KiB is approximately 8 MiB of payload per queue, plus object overhead, encoding/pipe/socket buffers, in-flight processing, waiting producers and lingering old-session work. Decoded objects may exceed their wire size. Bound producer concurrency and await sends sequentially rather than creating unlimited waiting tasks. Bounded queues are not a hard process-memory limit. + +Slow consumers block receive-channel writes and pause decoding; backpressure propagates through the pipe and TCP to the peer. The incomplete-frame byte limit checks unconsumed partial frames, not complete messages or total memory. While the channel is blocked, the incomplete-frame timer does not run and the byte check must wait for decoding to resume. Receive idle timeout suits periodic traffic; keep it at zero when silence is valid and choose a partial-frame timeout from the protocol's transfer deadline. KeepAlive uses positive milliseconds, rounded up to seconds on modern .NET; netstandard2.0 IOControl retains milliseconds. + +A caller timeout for `WaitForConnectedAsync` cancels only that wait. Lifetime cancellation/disposal stops the connection and cancels future waits too, even without a caller token (#65). Plain `SendAsync` guarantees neither socket write nor peer acknowledgement: queued entries can continue in a new session, but a failed dequeued send cannot be treated as delivered (#67). Reliable business delivery needs protocol ACKs, retries and deduplication. Session-bound send success also means only that the entire frame reached the local socket. + ## Session-aware messaging (advanced) For typical business traffic that tolerates replay, `SendAsync` (completes on enqueue; queued messages continue on the next session) + `GetMessages` (a stable stream across reconnects) is enough. Some protocols have strict session boundaries — re-handshake after reconnect, no replay of old-session messages, protocol timers starting when the frame is actually written (HSMS Select/T3/T6/T8). For those there is an optional capability interface `ISessionAwareStreamConnection` (implemented by `StreamConnection`; upper layers depending on the interface abstraction detect it with `is`): @@ -275,7 +318,7 @@ For 64KB-class messages the dominant costs are the codec style and the message t // 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; +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 an [owning codec](#buffer-ownership-and-concurrency) 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 diff --git a/README.md b/README.md index 0ee1a36..28fa196 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,8 @@ dotnet add package StreamFrame.Protocols.Xml ## 快速上手 +本库面向**单连接设备对接**。被动监听同时只服务一个客户端；`AcceptFirstClientOnly = false` 仅保持监听，不增加多客户端处理能力。以下为两个独立进程/角色的入门片段，`ct` 由宿主提供；部署前参考下方生产配置，并用 `await using` 释放连接。 + 一条连接 = 一个**帧定界策略**（framing）+ 一个**编解码器**（codec）+ 地址/端口/模式。framing 与 codec 均**连接级固定**，一条连接只流通一种消息类型、一种帧格式。 ```csharp @@ -95,7 +97,10 @@ sealed class SystemTextJsonCodec : ICodec public static readonly SystemTextJsonCodec Instance = new(); public JsonElement Decode(in ReadOnlySequence frame, CancellationToken ct = default) - => JsonDocument.Parse(frame.ToArray()).RootElement; + { + using var document = JsonDocument.Parse(frame.ToArray()); + return document.RootElement.Clone(); // Own data beyond document disposal. + } public void Encode(JsonElement message, IBufferWriter writer, CancellationToken ct = default) { @@ -111,7 +116,7 @@ sealed class SystemTextJsonCodec : ICodec - **客户端/服务端双模式**：`isActive: true` 主动连远端，`false` 被动监听；IPv4/IPv6 双栈（监听 `IPAddress.Any` 自动按双栈处理，IPv4 客户端地址归一显示为 IPv4） - **自动重连**：`Connecting → Connected → Retry` 状态机；可选指数退避（`MaxRetryDelayMs`，连续失败倍增封顶 + ±20% 抖动，连接成功自动复位）；`GetMessages` 是跨重连的稳定消息流——断线重连后已收消息不丢、枚举不中断 - **启动与停止**：`Start(ct)` 的 `ct` 是连接的**生命周期令牌**——取消它会停止连接/重连并拆线（状态进入 `Disconnected` 终态，`GetMessages` 自然结束，之后需新建连接）；`DisposeAsync` 与之等效。停机后 `SendAsync` 抛 `ChannelClosedException` -- **等待连接就绪**：`await conn.WaitForConnectedAsync(ct)`——已连接立即完成，未连接时等到下次连接成功（或取消/Dispose），不用轮询状态、不用 `Task.Delay` 盲等 +- **等待连接就绪**：`await conn.WaitForConnectedAsync(ct)`——已连接立即完成，未连接时等到下次连接成功（或调用方取消、生命周期取消/Dispose）；终态后的新等待也会取消，即使未传令牌，不用轮询状态 - **健壮性**：帧内容解码失败、未完成帧超限/超时、发送失败、接收空闲超时都会判定会话失效并自动重建（不再产生"连接看似存活、消息静默消失"的假活） - **活性探测（可选）**：TCP KeepAlive 与接收空闲超时，兜底半开连接（对端断电/拔线） - **事件**：`ConnectionChanged` 状态变化、`FrameError` 帧层诊断、`RawBytesReceived/Sent` 原始字节（HEX 调试） @@ -119,6 +124,44 @@ sealed class SystemTextJsonCodec : ICodec - **发送背压**：有界发送队列，队列满时 `SendAsync` 自动等待；接收侧默认无上限缓冲，可用 `ReceiveQueueCapacity` 设上限——消费慢时解码暂停、TCP 背压自然传导到对端，防内存无限增长 - **会话感知收发（可选高级）**：`CurrentSessionId` / `SendInSessionAsync`（整帧写入 socket 才完成、会话失效即失败、绝不跨会话重放）/ `GetSessionMessages`（消息带会话编号）——为有严格会话边界的协议（如 HSMS）准备，详见下文 +## 缓冲所有权与并发约束 + +`Decode` 的 `frame` 仅在同步调用期间借用。返回消息及其嵌套字段必须独立拥有所需数据，不能返回 `frame.First`、切片或引用管线内存的 `ReadOnlyMemory`。管线随后推进/释放，业务可能跨 `await` 才消费。`ReadOnlyMemory` 不代表所有权转移；本接口没有转移管线池化内存所有权的机制。 + +`Encode` 必须在返回前完成写入，不得保留/释放 `writer`，也不得保留其缓冲供异步使用。发送队列存消息对象，入队后不要修改或归还底层数组：普通 `SendAsync` 成功只表示入队，编码可能还没开始。最简单的做法是发送独立数组且之后不再修改。 + +同一会话的发送 worker 只串行化自己的编码调用，**不是整个 Codec 的锁**：`Encode` 与 `Decode` 可并行；会话拆除最多等待 2 秒，旧解码任务（例如被接收背压阻塞）可与新会话解码重叠，阻塞的旧编码也不能假定已退出。使用调用局部状态或自行同步，避免阻塞同步方法。跨连接共享 Codec/Framer 实例前，由调用方确认其并发安全；每连接一个实例仍须满足收发和新旧会话约束。 + +`IFramer` 同样适用：不能留存输入序列；输出 `payload` 可借用输入，供连接紧接着同步调用 Codec。编码 writer 只能在本次调用内使用。`IStreamingFramer.BeginFrame`/`EndFrame` 之间可穿插其它调用，配对状态应放在传入 writer 中，不能依赖实例的“当前帧”字段。 + +完整可编译的 [`OwnedBytesCodec`](samples/StreamFrame.Ownership/OwnedBytesCodec.cs) / [`OwnedMemoryCodec`](samples/StreamFrame.Ownership/OwnedMemoryCodec.cs) 对两种消息类型均返回 `frame.ToArray()`，包括单段输入；`Encode` 同步 `writer.Write(message)` 或 `writer.Write(message.Span)`。源文件直接纳入测试工程，在 net8.0/net10.0/net48 编译。`OwnershipExampleTests` 主动覆盖单段/多段输入，并用归还时填充 `0xDD` 的内存池运行真实 FrameDecoder，等管线释放后再异步消费，断言原始数据仍正确。 + +## 生产配置起点 + +完整 [`ProductionExample.RunAsync`](samples/StreamFrame.Ownership/ProductionExample.cs) 展示有界收发、10 秒连接等待超时、异步消费、生命周期取消及 `await using` Dispose，直接参与三目标编译。以下参数假设负载最多 64 KiB，须按协议、峰值流量和处理时间调整；不改变库默认值： + +```csharp +var framer = new LengthPrefixFramer(64 * 1024); +var options = new StreamConnectionOptions +{ + SendQueueCapacity = 128, + ReceiveQueueCapacity = 128, // 默认 0 为无界 + MaxIncompleteFrameBufferBytes = 64 * 1024 + 4, // 含长度头 + IncompleteFrameTimeoutMs = 5_000, + ReceiveIdleTimeoutMs = 0, // 允许静默；有 5 秒心跳时可考虑 15_000 + TcpKeepAlive = true, + KeepAliveTimeMs = 30_000, + KeepAliveIntervalMs = 1_500, // 现代 .NET 向上取整为 2 秒（#62） + AcceptFirstClientOnly = true, +}; +``` + +容量按**消息数**而非字节预算：128 条 × 64 KiB 约为每队列 8 MiB 负载，还需计算对象开销、编码/管线/socket 缓冲、处理中消息、等待入队的生产者及旧会话残留工作。解码对象也可能比线上负载大。限制生产者并发并逐条 `await SendAsync`，避免无限创建等待任务；有界队列不是进程内存硬上限。 + +慢消费者使接收通道写入等待，解码暂停，背压经管线和 TCP 逐步传向对端。半帧字节限制只检查未消费半帧，不是完整消息或总内存预算；通道堵塞期间半帧超时不计时，字节检查也要等解码循环继续才能执行。接收空闲超时适合周期流量；允许静默时保留 0，并按帧传输时限选择半帧超时。KeepAlive 参数为正毫秒，现代 .NET 向上取整为秒，netstandard2.0 IOControl 保留毫秒。 + +`WaitForConnectedAsync` 的调用方超时只取消本次等待；生命周期取消/Dispose 停止连接，终态后的新等待也取消，即使未传令牌（#65）。普通 `SendAsync` 不保证 socket 写出或远端确认，未出队条目可跨会话续发，已出队失败不能视为已交付（#67）；可靠业务交付需要协议 ACK、重试和去重。会话绑定发送成功也只表示整帧交给本机 socket。 + ## 会话感知收发（高级） 对允许消息重放的一般业务，`SendAsync`（入队即完成、断线后由新会话续发）+ `GetMessages`（跨重连稳定流）就够了。部分协议有严格的会话边界——重连后必须重新握手、旧会话消息禁止重放、协议计时器要从"整帧实际写出"起算（HSMS 的 Select/T3/T6/T8 即是）。为此提供可选能力接口 `ISessionAwareStreamConnection`（`StreamConnection` 已实现；依赖接口抽象的上层用 `is` 探测）： @@ -269,7 +312,7 @@ BenchmarkDotNet 实测（详见 [bench/README.md](bench/README.md)，可本地 // 不推荐：writer.Write(Encoding.UTF8.GetBytes(message)); // 每条一次全尺寸 byte[] ``` -2. **消息类型选 byte[] / ReadOnlyMemory**：string 消息每条固有 ≈2× 报文大小的 UTF-16 物化分配；字节负载 + 透传 codec 时框架距裸 TCP 仅 ≈20–30%； +2. **消息类型选 byte[] / ReadOnlyMemory**：string 消息每条固有 ≈2× 报文大小的 UTF-16 物化分配；字节负载 + [拥有数据的 codec](#缓冲所有权与并发约束) 时框架距裸 TCP 仅 ≈20–30%； 3. 保持默认的**流式编码**开启（`UseStreamingEncode`），发送缓冲会按上一帧大小自适应起租（封顶 1MB）。 ## 支持框架 diff --git a/samples/StreamFrame.Ownership/OwnedBytesCodec.cs b/samples/StreamFrame.Ownership/OwnedBytesCodec.cs new file mode 100644 index 0000000..3365924 --- /dev/null +++ b/samples/StreamFrame.Ownership/OwnedBytesCodec.cs @@ -0,0 +1,20 @@ +using System.Buffers; +using StreamFrame; + +namespace StreamFrame.Ownership; + +// 无共享可变状态 / No shared mutable state. +public sealed class OwnedBytesCodec : ICodec +{ + public byte[] Decode(in ReadOnlySequence frame, CancellationToken ct = default) + { + ct.ThrowIfCancellationRequested(); + return frame.ToArray(); // Own the bytes, including for single-segment frames. + } + + public void Encode(byte[] message, IBufferWriter writer, CancellationToken ct = default) + { + ct.ThrowIfCancellationRequested(); + writer.Write(message); + } +} diff --git a/samples/StreamFrame.Ownership/OwnedMemoryCodec.cs b/samples/StreamFrame.Ownership/OwnedMemoryCodec.cs new file mode 100644 index 0000000..d103ad0 --- /dev/null +++ b/samples/StreamFrame.Ownership/OwnedMemoryCodec.cs @@ -0,0 +1,19 @@ +using System.Buffers; +using StreamFrame; + +namespace StreamFrame.Ownership; + +public sealed class OwnedMemoryCodec : ICodec> +{ + public ReadOnlyMemory Decode(in ReadOnlySequence frame, CancellationToken ct = default) + { + ct.ThrowIfCancellationRequested(); + return frame.ToArray(); // ReadOnlyMemory alone would not establish ownership. + } + + public void Encode(ReadOnlyMemory message, IBufferWriter writer, CancellationToken ct = default) + { + ct.ThrowIfCancellationRequested(); + writer.Write(message.Span); + } +} diff --git a/samples/StreamFrame.Ownership/ProductionExample.cs b/samples/StreamFrame.Ownership/ProductionExample.cs new file mode 100644 index 0000000..f0758dc --- /dev/null +++ b/samples/StreamFrame.Ownership/ProductionExample.cs @@ -0,0 +1,55 @@ +using System.Net; +using StreamFrame; + +namespace StreamFrame.Ownership; + +public static class ProductionExample +{ + // Starting values for <= 64 KiB payloads; tune to the device protocol and workload. + public static StreamConnectionOptions CreateOptions() => new() + { + SendQueueCapacity = 128, + ReceiveQueueCapacity = 128, + MaxIncompleteFrameBufferBytes = 64 * 1024 + 4, + IncompleteFrameTimeoutMs = 5_000, + ReceiveIdleTimeoutMs = 0, // Idle is legal; use e.g. 15_000 only with a 5s heartbeat. + TcpKeepAlive = true, + KeepAliveTimeMs = 30_000, + KeepAliveIntervalMs = 1_500, // Modern .NET: rounded up to 2 seconds. + AcceptFirstClientOnly = true, + }; + + public static async Task RunAsync(IPAddress address, int port, CancellationToken stoppingToken) + { + using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken); + await using var connection = new StreamConnection>( + new LengthPrefixFramer(64 * 1024), new OwnedMemoryCodec(), + address, port, isActive: true, options: CreateOptions()); + connection.Start(lifetime.Token); + try + { + using (var ready = CancellationTokenSource.CreateLinkedTokenSource(lifetime.Token)) + { + ready.CancelAfter(TimeSpan.FromSeconds(10)); + await connection.WaitForConnectedAsync(ready.Token); + } + + // Owned immutable data: do not change/recycle this array after enqueue. + await connection.SendAsync(new byte[] { 1, 2, 3 }, lifetime.Token); + // SendAsync means enqueue only. Protocol ACK/retry/deduplication belongs to the application. + await foreach (var message in connection.GetMessages(lifetime.Token)) + { + await Task.Delay(20, lifetime.Token); // Simulated asynchronous business processing. + Console.WriteLine(BitConverter.ToString(message.ToArray())); // Still valid after await. + } + } + catch (OperationCanceledException) when (lifetime.IsCancellationRequested) + { + // Requested shutdown. A readiness timeout otherwise propagates to the caller. + } + finally + { + lifetime.Cancel(); // Stop reconnects; await using also disposes on timeout/failure. + } + } +} diff --git a/src/StreamFrame/Codec/ICodec.cs b/src/StreamFrame/Codec/ICodec.cs index dc25259..01e293f 100644 --- a/src/StreamFrame/Codec/ICodec.cs +++ b/src/StreamFrame/Codec/ICodec.cs @@ -7,12 +7,33 @@ namespace StreamFrame; /// /// 由驱动实现——例如 XML 驱动把负载解析为强类型消息，自定义二进制协议驱动手写字节布局。 /// +/// +/// 并发 / Concurrency: 同一会话的发送 worker 串行编码，但 Encode 与 Decode 可并行。 +/// 会话拆除只作有界等待，旧任务可能与新会话重叠（包括 Decode/Decode，阻塞的 Encode 也不能假定已退出）。 +/// 实现应无共享可变状态或自行同步；跨连接共享实例的调用方负责确认实现支持并发，框架不提供实例级锁。 +/// A send worker serializes its own calls, not the codec instance. Encode/Decode and old/new +/// session calls can overlap. Use stateless implementations or synchronization; callers sharing +/// an instance across connections must ensure it is safe. Avoid blocking synchronous methods. +/// /// 业务消息类型，贯穿一条连接。 public interface ICodec { /// 把一帧负载（不含定界字节）解析为一条业务消息。 + /// + /// frame 仅供同步调用期间借用；返回对象（包括嵌套字段）必须独立拥有数据。 + /// 不得未经拷贝保留 frame、其切片或底层 Memory；管线随后推进/释放，异步消费者可能更晚读取。 + /// ReadOnlyMemory 只限制写入接口，不转移所有权。本接口不转移管线内存所有权。 + /// Borrow frame only during this synchronous call. Copy retained bytes into message-owned storage; + /// never return pipeline slices, even as ReadOnlyMemory. No pipeline ownership transfer is provided. + /// TMessage Decode(in ReadOnlySequence frame, CancellationToken ct = default); /// 把一条业务消息编码为帧内负载字节，写入 。 + /// + /// 必须同步完成写入；不得保留 writer 或其缓冲供返回后/异步使用，也不得释放框架拥有的 writer。 + /// 调用方应保持已入队消息及其底层数据不变，普通 SendAsync 完成仅表示入队，编码可能尚未开始。 + /// Complete writes synchronously. Do not retain/dispose the writer or retain its buffers after return. + /// Queued messages must remain unchanged: SendAsync completion does not mean encoding has finished. + /// void Encode(TMessage message, IBufferWriter writer, CancellationToken ct = default); } diff --git a/src/StreamFrame/Framing/IFramer.cs b/src/StreamFrame/Framing/IFramer.cs index f4c2dd6..3f705e8 100644 --- a/src/StreamFrame/Framing/IFramer.cs +++ b/src/StreamFrame/Framing/IFramer.cs @@ -7,6 +7,12 @@ namespace StreamFrame; /// /// 发送与接收共用同一实现，保证一条连接上帧格式永远一致。 /// +/// +/// 同连接编码与切帧可并行，新旧会话也可能重叠；跨连接共享实例前须确认线程安全。 +/// 应无共享可变解析状态或自行同步，框架不提供实例级锁。 +/// Encode and decode may overlap, as may old/new sessions. Shared instances must be thread-safe; +/// keep parsing state local or synchronize it yourself. The framework does not lock the instance. +/// public interface IFramer { /// 单帧负载允许的最大字节数，用于防御超长帧撑爆缓冲区。 @@ -14,6 +20,8 @@ public interface IFramer /// /// 对完整的帧内负载（不含定界字节）加帧定界并写入 。 + /// payload、writer 及其缓冲仅供同步调用期间借用，不得留存供异步使用或释放 writer。 + /// Borrow payload/writer only for this call; do not retain buffers or dispose the writer. /// void EncodeFrame(ReadOnlySpan payload, IBufferWriter writer); @@ -27,6 +35,9 @@ public interface IFramer /// buffer 必须保留原缓冲的未消费后缀。返回 false 时也可通过消费前缀进行重同步； /// 连接层在长度减少时继续解析剩余缓冲，未消费任何字节时等待更多输入。 /// 返回 true 时必须消费完整帧的线上字节（包括空负载帧的定界字节）。 + /// payload 可以是输入切片，供连接层紧接着同步 Decode；实现不得留存 buffer/payload。 + /// Payload may borrow the input for the immediately following synchronous codec call; + /// the framer must not retain either sequence beyond this call. /// bool TryDecodeFrame(ref ReadOnlySequence buffer, out ReadOnlySequence payload); } diff --git a/src/StreamFrame/Framing/IStreamingFramer.cs b/src/StreamFrame/Framing/IStreamingFramer.cs index 6e89d71..585d721 100644 --- a/src/StreamFrame/Framing/IStreamingFramer.cs +++ b/src/StreamFrame/Framing/IStreamingFramer.cs @@ -15,6 +15,13 @@ namespace StreamFrame; /// /// 协议语义（帧边界、超长防御）与对应 完全一致。 /// +/// +/// 沿用 IFramer 的并发与借用约束。BeginFrame/EndFrame 的配对状态应存于本次 writer， +/// 不应使用实例字段记录当前帧；两个调用之间可穿插解码或其它会话/连接的调用。 +/// writer 及缓冲不得被保留供方法返回后使用；EndFrame 使用调用方重新传入的 writer。 +/// IFramer concurrency rules apply. Keep pair state in the supplied writer, not a current-frame field: +/// other decode/session/connection calls may interleave. Never retain writer/buffers after a method returns. +/// public interface IStreamingFramer : IFramer { /// 帧编码的起始；BeginFrame 后 writer 的任何写入都属于帧内容。 diff --git a/src/StreamFrame/IStreamConnection.cs b/src/StreamFrame/IStreamConnection.cs index c78dd05..4fe6a4a 100644 --- a/src/StreamFrame/IStreamConnection.cs +++ b/src/StreamFrame/IStreamConnection.cs @@ -72,12 +72,20 @@ public interface IStreamConnection : IAsyncDisposable /// /// 等待连接进入 ：已连接时立即完成；否则等到 - /// 下一次连接成功、 取消或连接 Dispose（任务以取消结束）。 + /// 下一次连接成功、 取消、生命周期取消或连接 Dispose（任务以取消结束）。 + /// 终态后的新等待同样以取消结束，即使未传调用方令牌。 + /// Lifetime cancellation/disposal also cancels future waits, even without a caller token. /// 替代轮询 或 Task.Delay 式等待。 /// Task WaitForConnectedAsync(CancellationToken ct = default); /// 发送一条业务消息。仅入队，由发送 worker 编码加帧后写出；队列满时背压等待。 + /// + /// 完成不代表已编码、写入 socket 或远端确认。消息及底层数据应在入队后保持不变。 + /// 未出队普通消息可由新会话续发；已出队发送失败的远端结果未知，不保证交付。 + /// Completion means enqueue only, not encoding, socket write or peer acknowledgement. + /// Keep queued data immutable; queued entries can continue in a new session, but delivery is not guaranteed. + /// Task SendAsync(TMessage message, CancellationToken ct = default); /// diff --git a/src/StreamFrame/StreamConnectionOptions.cs b/src/StreamFrame/StreamConnectionOptions.cs index 7084eb8..a5cd5a4 100644 --- a/src/StreamFrame/StreamConnectionOptions.cs +++ b/src/StreamFrame/StreamConnectionOptions.cs @@ -22,7 +22,8 @@ public sealed class StreamConnectionOptions /// Socket 接收缓冲区大小（字节）。 public int SocketReceiveBufferSize { get; set; } = 65536; - /// 发送队列容量；超过容量时 SendAsync 将等待（背压）。 + /// 发送队列容量（消息数，非字节预算）；超过容量时 SendAsync 将等待（背压）。 + /// Capacity counts messages, not bytes; full queues backpressure SendAsync. public int SendQueueCapacity { get; set; } = 1024; /// 编码缓冲区的初始大小（字节）。 @@ -75,6 +76,8 @@ public sealed class StreamConnectionOptions /// 接收消息通道容量（默认 0 = 不限制）：解码出的业务消息先入通道，再由 GetMessages 消费。 /// 设为正数后，消费慢时解码循环暂停读取（TCP 背压自然传导到对端），防止慢消费者撑爆内存。 /// 注意：消费完全停滞期间，会话拆除最多等待 2 秒（内部超时）后继续。 + /// 容量按消息数而非字节计量；还需预算单条对象大小、处理中消息、等待入队的生产者及管线/socket 缓冲。 + /// Capacity counts messages, not bytes. Budget message size, in-flight work, waiting producers and I/O buffers too. /// public int ReceiveQueueCapacity { get; set; } diff --git a/test/StreamFrame.Tests/OwnershipExampleTests.cs b/test/StreamFrame.Tests/OwnershipExampleTests.cs new file mode 100644 index 0000000..22520b8 --- /dev/null +++ b/test/StreamFrame.Tests/OwnershipExampleTests.cs @@ -0,0 +1,89 @@ +using System.Buffers; +using System.IO.Pipelines; +using System.Threading.Channels; +using StreamFrame; +using StreamFrame.Ownership; + +namespace StreamFrame.Tests; + +public class OwnershipExampleTests +{ + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task DecodedMessages_OwnBytes_AfterSourceIsOverwritten(bool segmented) + { + var source = new byte[] { 1, 2, 3, 4 }; + var sequence = segmented ? Segment.Create(source) : new ReadOnlySequence(source); + var bytes = new OwnedBytesCodec().Decode(sequence); + var memory = new OwnedMemoryCodec().Decode(sequence); + Array.Clear(source, 0, source.Length); // Deterministic invalidation, no pool timing assumption. + await Task.Yield(); + Assert.Equal(new byte[] { 1, 2, 3, 4 }, bytes); + Assert.Equal(bytes, memory.ToArray()); + using var writer = new PooledBufferWriter(16); + new OwnedBytesCodec().Encode(bytes, writer); + new OwnedMemoryCodec().Encode(memory, writer); + Assert.Equal(new byte[] { 1, 2, 3, 4, 1, 2, 3, 4 }, writer.WrittenMemory.ToArray()); + ProductionExample.CreateOptions().Validate(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Decoder_AsyncConsumer_AfterPipeStorageIsPoisoned(bool memoryMessages) + { + if (memoryMessages) + await VerifyPipe(new OwnedMemoryCodec(), static value => value.ToArray()); + else + await VerifyPipe(new OwnedBytesCodec(), static value => value); + } + + private static async Task VerifyPipe(ICodec codec, Func getBytes) + { + using var pool = new PoisonPool(); + var pipe = new Pipe(new PipeOptions(pool: pool)); + var relay = Channel.CreateBounded>(2); + var metrics = new ConnectionMetrics("ownership-test"); + var decoder = new FrameDecoder(pipe.Reader, new LengthPrefixFramer(), codec, + relay, metrics, 1, 1024, 0, DecodeErrorPolicy.Disconnect, null); + await pipe.Writer.WriteAsync(new byte[] { 0, 0, 0, 4, 1, 2, 3, 4 }); + await pipe.Writer.CompleteAsync(); + await decoder.RunAsync(CancellationToken.None); // CompleteAsync returns all borrowed segments. + Assert.True(pool.Returned > 0); + await Task.Yield(); + Assert.True(relay.Reader.TryRead(out var received)); + Assert.Equal(new byte[] { 1, 2, 3, 4 }, getBytes(received.Message)); + } + + private sealed class Segment : ReadOnlySequenceSegment + { + private Segment(ReadOnlyMemory memory) => Memory = memory; + public static ReadOnlySequence Create(byte[] bytes) + { + var first = new Segment(bytes.AsMemory(0, 2)); + var last = new Segment(bytes.AsMemory(2)) { RunningIndex = 2 }; + first.Next = last; + return new ReadOnlySequence(first, 0, last, 2); + } + } + + private sealed class PoisonPool : MemoryPool + { + public int Returned { get; private set; } + public override int MaxBufferSize => int.MaxValue; + public override IMemoryOwner Rent(int minBufferSize = -1) + => new Owner(this, new byte[Math.Max(4096, minBufferSize)]); + protected override void Dispose(bool disposing) { } + + private sealed class Owner(PoisonPool pool, byte[] bytes) : IMemoryOwner + { + public Memory Memory => bytes; + public void Dispose() + { + for (var i = 0; i < bytes.Length; i++) bytes[i] = 0xDD; + pool.Returned++; + } + } + } +} diff --git a/test/StreamFrame.Tests/StreamFrame.Tests.csproj b/test/StreamFrame.Tests/StreamFrame.Tests.csproj index ed36b57..00655a5 100644 --- a/test/StreamFrame.Tests/StreamFrame.Tests.csproj +++ b/test/StreamFrame.Tests/StreamFrame.Tests.csproj @@ -29,6 +29,8 @@ + +