diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0f01d31..b949dfd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -18,6 +18,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 必需检查名不变。
### 变更
+- **会话对象化与拆除分段（[#68](https://github.com/CSJ608/StreamFrame/issues/68) PR-2）**：会话五元组（socket/Pipe/取消源/接收、解码、发送三任务/编号纪元/建立时刻）收敛为不可变的内部 `Session` 对象——会话任务只操作所属会话的资源（接收循环与发送 worker 绑定本会话 socket，不再读"当前值"字段），直连迁移与拆除竞速中新旧会话互不干扰；拆除只关本会话的 socket（`_socket` 字段降级为诊断/交接镜像）。会话拆除分段：锁内仅快速中止（取消令牌 + **先关 socket 解锁阻塞 I/O**——netfx 上取消令牌无法中止已提交的接收），有界等待（≤2s）移到锁外——高并发重连/停机不再在会话锁上排队，netfx 拆除收敛加快；停机/重连的总阻塞时长与公共 API 的同步/异步行为不变。全套测试（139+139+131）多轮全绿；同机同口径基准抽检（OneWayThroughput vs issue-70 基线）：框架路径与 DirectTcp 对照同向漂移（环境噪声），未观测框架特有劣化，每操作分配无系统性增长。
- **状态机与所有权规范（[#68](https://github.com/CSJ608/StreamFrame/issues/68) PR-1）**：DESIGN 增补权威清单——状态转换全表与发布段/通知段规则、资源所有权表（会话 socket/监听器/会话五元组/消息通道）、锁与同步原语清单、允许重入点（回调内 Reconnect/Dispose、停机后调用、迟到旧会话任务）及已知限制（拆除同步等待持锁、`_socket` 字段语义——指向后续内部 Session 对象重构）；与新增的 `StateOwnershipBaselineTests` 共同构成 #68 会话所有权重构的行为基线。
- 依赖升级（Dependabot，[PR #59](https://github.com/CSJ608/StreamFrame/pull/59)）：soak 工作流 `actions/setup-dotnet` v4→v6（Node 24 运行时，与其余工作流对齐）。
- **驱动所有权与并发文档（[#69](https://github.com/CSJ608/StreamFrame/issues/69)）**：补全 ICodec/IFramer/流式定界的缓冲借用及同连接收发、新旧会话、跨连接共享约束；双语 README 增加拥有独立数据的 byte[]/ReadOnlyMemory 示例、有界队列生产配置及单客户端边界，澄清终态等待、KeepAlive 粒度和入队非确认。示例纳入三目标编译并以主动覆写缓冲验证异步消费；不改变默认值或实现公共契约。
diff --git a/docs/DESIGN.md b/docs/DESIGN.md
index 3a7daa1..00c044f 100644
--- a/docs/DESIGN.md
+++ b/docs/DESIGN.md
@@ -134,7 +134,7 @@ socket --ReceiveAsync--> pipe.Writer --[Pipe]--> FrameDecoder.RunAsync
**等待连接就绪**：`WaitForConnectedAsync` 基于"当前代"的 TaskCompletionSource——已连接立即完成；否则挂起，进入 `Connected` 时完成并清空（供下轮复用），`Shutdown` 时以取消收尾。
-**会话模型（1.2.0）**：一次 TCP 连接 = 一个会话（Pipe + 接收/解码/发送三任务，各自带守护）。会话因对端断开、socket 故障、帧解码失败（`DecodeErrorPolicy.Disconnect`）、未完成帧超限、发送失败或接收空闲超时而终结时，守护任务经 `ScheduleReconnect` 调度重建——不能在任务内联调用 `Reconnect`（StopSession 会等待调用方自身，自等待 2s）。每个会话持有单调递增的 epoch；故障重连只在 epoch 仍为最新时执行，迟到的过期故障不会误杀已重建的新会话。
+**会话模型（1.2.0；Session 对象化见 #68）**：一次 TCP 连接 = 一个会话（内部 Session 对象集中持有 socket、Pipe、取消源与接收/解码/发送三任务，各自带守护）。会话因对端断开、socket 故障、帧解码失败（`DecodeErrorPolicy.Disconnect`）、未完成帧超限、发送失败或接收空闲超时而终结时，守护任务经 `ScheduleReconnect` 调度重建——不能在任务内联调用 `Reconnect`（StopSession 会等待调用方自身，自等待 2s）。每个会话持有单调递增的 epoch；故障重连只在 epoch 仍为最新时执行，迟到的过期故障不会误杀已重建的新会话。
**消息通道所有权（1.2.0）**：`_messageRelay` 归连接所有、跨会话复用，仅停机（Shutdown）时完成——`GetMessages` 是跨重连的稳定消息流。默认无上限；`ReceiveQueueCapacity > 0` 时改为有界通道（FullMode=Wait），消费慢时解码循环暂停读取，TCP 背压自然传导到对端。`FrameDecoder` 不完成通道（1.1.0 在此把取消异常写进通道完成状态，导致重连后消息永不到达的假活）。解码循环的 `ReadAsync` 不绑定取消令牌（取消中的 ReadAsync 会把 Pipe 留在"读取进行中"状态，无法再 TryRead）；退出由 `writer.CompleteAsync` / `CancelPendingRead` 驱动，退出前投递所有已缓冲的完整帧——"已收到的字节必达"。
@@ -158,19 +158,21 @@ socket --ReceiveAsync--> pipe.Writer --[Pipe]--> FrameDecoder.RunAsync
发布段规则（`_publishGate` 锁内，原子）：① 停机后到达的非终态发布**整体丢弃**——终态必须是最后对外可见的状态（并发 Reconnect/连接成功与 Dispose 竞速时不被翻转）；② 过期纪元的故障发布在任何对外可见效果之前丢弃；③ 会话编号分配/归零与纪元递增先于 `State` 写入（线性化点：`ConnectionChanged` 回调与 `WaitForConnectedAsync` 完成时刻读取 `CurrentSessionId` 必得一致值）。通知段（锁外）：用户回调（异常隔离）、等待器完成、会话创建/拆除派发——回调可能重入，不得持有发布锁。
-**会话创建的纪元现行性门控**：`StartSession` 只在"发布帧的纪元仍是最新、编号非零"时创建会话。Connected 回调内同步 `Reconnect()` 或发布窗口内的现行纪元故障会让位于取代者（其循环重建连接），迟到帧不再"兜底重建"——那曾以过期编号复用、拆除新会话刚建立的 socket（撕裂会话）。被取代帧的 socket 在新连接覆盖 `_socket` 时当场释放。
+**会话创建的纪元现行性门控**：`StartSession` 只在"发布帧的纪元仍是最新、编号非零"时装配会话对象。Connected 回调内同步 `Reconnect()` 或发布窗口内的现行纪元故障会让位于取代者（其循环重建连接），迟到帧不再"兜底重建"——那曾以过期编号复用、拆掉新会话刚建立的 socket（撕裂会话）。被取代帧的 socket 在新连接覆盖 `_socket` 时当场释放。
+
+**拆除分段（PR-2）**：会话拆除 = 锁内快速中止（`DetachSession`：结算指标 → 取消令牌 → 唤醒解码读取 → 关闭本会话 socket——netfx 上取消令牌无法中止已提交的接收，只有关 socket 能立即解锁阻塞 I/O，即"先中止 I/O 再等待"）+ 锁外收尾（`WaitAndDisposeSession`：有界等待三任务 ≤2s、释放 CTS）。会话锁内无任何等待：并发的会话建立/拆除互不排队；停机与重连的总阻塞时长与旧实现一致（≤2s）。
**资源所有权**：
| 资源 | 创建者 | 释放者 | 备注 |
|---|---|---|---|
-| `_socket`（会话 socket） | 连接循环（connect/accept） | 会话拆除（`StopSessionCore`）；新连接覆盖时；`Shutdown` 末尾兜底 | 赋值与关闭互斥于 `_sessionGate`；停机守门失败的连接当场释放 |
+| 会话对象（内部 `Session`：socket + Pipe + CTS + 接收/解码/发送三任务 + 编号/纪元/建立时刻） | `StartSession`（纪元现行性守门后装配） | `DetachSession`（锁内快速中止：结算时长指标 → 取消令牌 → 唤醒解码读取 → **先关本会话 socket 解锁 I/O**）+ `WaitAndDisposeSession`（锁外有界等待 ≤2s、释放 CTS） | 不可变对象：任务只操作所属会话的资源，直连迁移/拆除竞速中新旧会话互不干扰；发布到 `_session` 前（`_sessionGate` 内）完成装配 |
+| `_socket`（当前连接 socket） | 连接循环（connect/accept，停机守门 + 与关闭块互斥） | 会话拆除时若指向该会话 socket 则置空；新连接覆盖时释放旧值；`Shutdown` 末尾兜底 | "当前值"镜像：诊断（`RemoteIpAddress`）与建立交接用；I/O 一律走会话自己的 socket |
| `_server`（监听 socket） | `InitServer`（Bind/Listen 全部成功才发布） | 单客户端接受后关闭；`Shutdown` 释放；bind 失败自清理 | 归当代接受循环所有（代次门控，#47） |
-| 会话五元组（`_sessionCts`/`_pipe`/接收、解码、发送三任务） | `StartSession` | `StopSessionCore`（cancel → 唤醒 → 置空字段 → 有界等待 ≤2s → 关 socket → 释放 cts） | 终态后不再创建（`StartSession` 停机守门） |
| `_messageRelay`（消息通道） | 构造 | 仅 `Shutdown`（连接级、跨会话存活） | 会话重建不完成它（1.2.0 假活修复） |
-| `_sendQueue` / 挂起的会话绑定条目 | 业务侧入队 | `Shutdown` 完成通道；拆除时 `FaultPendingSessionSends` | 条目三态 CAS：Queued/Claimed/Cancelled/Expired |
+| `_sendQueue` / 挂起的会话绑定发送条目 | 业务侧入队 | `Shutdown` 完成通道；拆除时 `FaultPendingSessionSends` | 条目三态 CAS：Queued/Claimed/Cancelled/Expired |
-**锁与同步原语清单**：`_publishGate`（状态发布段，锁内无 await/回调）；`_sessionGate`（会话创建/拆除、`_socket` 赋值互斥；拆除等待最长 2s——已知同步等待，见"已知限制"）；`_listenerGate`（监听器发布与停机释放互斥）；`_acceptLock`（接受尝试串行化 + 代次门控落点）；`_sendLock`（socket 写串行化）；CAS 字段：`_retryInProgress`（重拆防重入）、`_acceptLoopId`（接受循环代次）、`_disposeStage`/`_started`、发送条目状态字。
+**锁与同步原语清单**：`_publishGate`（状态发布段，锁内无 await/回调）；`_sessionGate`（会话装配/拆除、`_socket` 赋值互斥；**锁内无等待**——任务收尾在锁外，拆除先关 socket 再有界等待 ≤2s）；`_listenerGate`（监听器发布与停机释放互斥）；`_acceptLock`（接受尝试串行化 + 代次门控落点）；`_sendLock`（socket 写串行化，连接级）；CAS 字段：`_retryInProgress`（重拆防重入）、`_acceptLoopId`（接受循环代次）、`_disposeStage`/`_started`、发送条目状态字。
**允许的重入点**（用户回调可调用任意公共 API，异常一律隔离）：
@@ -179,7 +181,7 @@ socket --ReceiveAsync--> pipe.Writer --[Pipe]--> FrameDecoder.RunAsync
- 停机后的任何调用：`Reconnect()`/`SendInSessionAsync` 等抛 `ObjectDisposedException` 或以会话失效收尾；迟到的内部状态发布被发布段丢弃。
- 迟到的旧会话任务/故障：入口过期纪元检查丢弃，不误杀新会话（1.2.0/2.3.1 既有保证，重入路径由纪元门控补齐）。
-**已知限制（后续 PR 方向）**：① `StopSessionCore` 持 `_sessionGate` 同步等待会话任务最长 2 秒——线程池饥饿时放大停机/重连延迟（高并发压力下可观测），#68 后续以"先中止 I/O、异步等待任务"改造；② `_socket`/会话五元组是"当前值"字段语义而非会话对象所有——并发双循环的直连迁移路径靠覆盖释放与拆除收敛，集中所有权（内部 Session 对象）是重构目标。
+**已知限制（后续 PR 方向）**：① 停机与重连仍**同步**等待会话任务退出（锁外、有界 ≤2s）——公共 API 的同步/异步行为按 #68 约束保持不变；若未来允许 `DisposeAsync` 异步化，可把该等待移入后台收尾。② 接受循环被取代的"僵尸"帧在已取得 `_acceptLock` 且 accept 已完成时不再复查代次（ tolerated path：其发布经纪元机制收敛，socket 覆盖时释放）——统一状态转换入口（每次发布前重验代次/停机）是剩余的结构性收敛项。
### netstandard2.0 兼容层（2.1.0）
diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs
index fa782f6..e7a2a5b 100644
--- a/src/StreamFrame/Connection/StreamConnection.cs
+++ b/src/StreamFrame/Connection/StreamConnection.cs
@@ -22,9 +22,10 @@ namespace StreamFrame;
/// 发送：业务 SendAsync/SendInSessionAsync 入队 --有界Channel--> 发送worker → codec.Encode → framing.EncodeFrame → _sendLock → socket
///
///
-/// 会话模型：一次 TCP 连接 = 一个会话（Pipe + 三个会话任务 + 公共会话编号）；断线或会话故障时
-/// 整个会话作废重建，而 _messageRelay 消息通道是连接级的、跨会话存活——业务侧的
-/// 枚举在重连前后是同一条稳定流，仅在 Dispose 时正常结束。
+/// 会话模型：一次 TCP 连接 = 一个会话（内部 Session 对象集中持有 socket + Pipe + 三个
+/// 会话任务 + 取消源）；断线或会话故障时整个会话作废重建，而 _messageRelay
+/// 消息通道是连接级的、跨会话存活——业务侧的 枚举在重连前后
+/// 是同一条稳定流，仅在 Dispose 时正常结束。
/// 会话感知收发（）在此之上提供
/// 绑定会话的发送与带会话编号的接收视图。
///
@@ -69,6 +70,9 @@ public string? RemoteIpAddress
private const int DisposalComplete = 1;
private int _disposeStage;
+ /// 当前连接 socket：建立路径向会话交接用，兼作诊断（RemoteIpAddress）。
+ /// I/O 一律走所属会话的 ；本字段是"当前值"镜像，
+ /// 会话拆除时若仍指向该会话的 socket 则同步置空。
private Socket? _socket;
private Socket? _server;
@@ -85,7 +89,6 @@ public string? RemoteIpAddress
private readonly object _sessionGate = new();
private readonly object _publishGate = new();
#endif
- private Pipe? _pipe;
///
/// 接受循环代次（#47）：每次 StartCore 递增。用户重连与自动重连竞速会产生多个并发的
@@ -95,10 +98,13 @@ public string? RemoteIpAddress
/// _acceptLock 后静默退出：监听器的创建/关闭/accept 只归属当代循环。
///
private int _acceptLoopId;
- private CancellationTokenSource? _sessionCts;
- private Task? _receiveTask;
- private Task? _decodeTask;
- private Task? _sendWorkerTask;
+
+ /// 当前会话（#68）：一次 TCP 连接的全部会话资源——socket、Pipe、取消源、
+ /// 接收/解码/发送三个任务与编号/纪元/建立时刻——集中在一个不可变对象上，任务只操作
+ /// 所属会话的资源，新旧会话互不干扰（直连迁移不再关错 socket、发错连接）。
+ /// null = 无会话；读写均持 。连接级状态（编号计数器、
+ /// 当前编号、纪元）留在连接上——它们跨会话单调。
+ private Session? _session;
private int _sessionEpoch;
///
@@ -107,9 +113,6 @@ public string? RemoteIpAddress
/// 公共会话编号的分配计数器：仅在真实 TCP 会话建立时递增（有间隔、单调、不复用）；checked 防静默回绕。
private long _sessionCounter;
- /// 当前会话的建立时刻（Stopwatch 时戳），拆除时用于会话时长指标。
- private long _sessionStartedAt;
-
/// 当前会话编号：0 = 无会话。分配/归零都必须先于对应状态对外发布（线性化点，见 CommunicationStateChanging）。
private long _currentSessionId;
@@ -584,6 +587,7 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE
try
{
+ Session? session = null;
lock (_sessionGate)
{
// 取锁等待期间停机可能已完成：不拆除、不派发新循环（_lifetimeCts
@@ -596,9 +600,12 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE
_sessionEpoch++; // 同会话的其它迟到故障立即过期，防止二次拆建
FaultPendingSessionSends(); // 拆除时立即失败挂起的会话绑定发送（先于任务拆除）
- StopSessionCore();
+ session = DetachSession();
}
+ // 锁外收尾被拆除会话（有界 ≤2s），再派发新循环——顺序与旧实现一致
+ WaitAndDisposeSession(session);
+
// 取锁后停机仍可能已完成（_lifetimeCts 被 Shutdown 回收）：放弃派发，
// 静默让位终态；令牌源在取消与回收之间的窗口取 Token 可能抛 ODE，一并容忍。
var lifetime = _lifetimeCts;
@@ -623,6 +630,7 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE
private void StartSession(long sessionId, int epoch)
{
+ Session? superseded;
lock (_sessionGate)
{
// 停机守门（#68）：本方法唯一的迟到路径是"Connected 回调内同步 Dispose"——
@@ -634,24 +642,24 @@ private void StartSession(long sessionId, int epoch)
// 纪元现行性守门（#68）：本发布帧的纪元已被更新的重连取代——典型是
// Connected 回调内同步 Reconnect()（回调里 Retry 已递增纪元并派发新循环），
// 或发布窗口内现行纪元的故障重连。迟到帧不得以过期编号/纪元重建会话：
- // 编号会被复用、StopSessionCore 会拆掉新会话刚建立的 socket（旧"兜底重建"
- // 曾在该路径撕裂会话）。放弃创建，连接由取代者的循环完成；被取代帧的
- // socket 在新循环覆盖 _socket 时释放。sessionId==0 同理：编号已被归零
- // 说明拆除正在并发进行，等待取代者重建。
+ // 编号会被复用、会拆掉新会话刚建立的资源（旧"兜底重建"曾在该路径撕裂会话）。
+ // 放弃创建，连接由取代者的循环完成；被取代帧的 socket 在新循环覆盖 _socket
+ // 时释放。sessionId==0 同理：编号已被归零说明拆除正在并发进行。
if (Volatile.Read(ref _sessionEpoch) != epoch || sessionId == 0)
return;
- StopSessionCore();
+ var socket = _socket;
+ if (socket is null)
+ return; // 防御：无交接 socket（正常路径不会发生——发布前必有赋值）
+
+ superseded = DetachSession();
var cts = new CancellationTokenSource();
- var token = cts.Token;
// 段尺寸对齐接收缓冲：大报文（64KB 级）在默认 4KB 段下跨 16 段重组，
// 对齐后整帧常驻单段（64KB 基准实测见 bench/README）
var pipe = new Pipe(new PipeOptions(minimumSegmentSize: _options.SocketReceiveBufferSize));
- _sessionCts = cts;
- _pipe = pipe;
- _sessionStartedAt = Stopwatch.GetTimestamp();
+ var session = new Session(sessionId, epoch, socket, pipe, cts, Stopwatch.GetTimestamp());
var maxIncompleteFrameBytes = _options.MaxIncompleteFrameBufferBytes > 0
? _options.MaxIncompleteFrameBufferBytes
@@ -662,10 +670,16 @@ private void StartSession(long sessionId, int epoch)
maxIncompleteFrameBytes, _options.IncompleteFrameTimeoutMs,
_options.DecodeErrorPolicy, RaiseFrameError);
- _receiveTask = WatchSessionFaultsAsync(ReceiveLoopAsync(pipe.Writer, token, epoch), cts, epoch);
- _decodeTask = WatchSessionFaultsAsync(decoder.RunAsync(token), cts, epoch);
- _sendWorkerTask = WatchSessionFaultsAsync(SendWorkerAsync(token, sessionId), cts, epoch);
+ session.SetTasks(
+ WatchSessionFaultsAsync(ReceiveLoopAsync(session), session, epoch),
+ WatchSessionFaultsAsync(decoder.RunAsync(cts.Token), session, epoch),
+ WatchSessionFaultsAsync(SendWorkerAsync(session), session, epoch));
+ _session = session;
}
+
+ // 锁外收尾被取代会话（有界等待 ≤2s + 释放 CTS）：拆除等待不再阻塞 _sessionGate，
+ // 并发的新会话建立/拆除不必排队（#68 已知限制①收口）。
+ WaitAndDisposeSession(superseded);
}
///
@@ -673,13 +687,13 @@ private void StartSession(long sessionId, int epoch)
/// 不完整帧超限、未完成帧超时、发送失败）统一转为断线重连（按
/// 防止过期故障误杀新会话）。
///
- private async Task WatchSessionFaultsAsync(Task sessionTask, CancellationTokenSource sessionCts, int epoch)
+ private async Task WatchSessionFaultsAsync(Task sessionTask, Session session, int epoch)
{
try
{
await sessionTask.ConfigureAwait(false);
}
- catch (OperationCanceledException) when (sessionCts.IsCancellationRequested)
+ catch (OperationCanceledException) when (session.Cts.IsCancellationRequested)
{
}
catch (Exception ex)
@@ -706,82 +720,89 @@ private void FaultPendingSessionSends()
private void StopSession()
{
+ Session? session;
lock (_sessionGate)
{
- StopSessionCore();
+ session = DetachSession();
}
+
+ // 锁外收尾：停机的总阻塞时长与旧实现一致（有界 ≤2s），但不再占住会话锁
+ WaitAndDisposeSession(session);
}
- private void StopSessionCore()
+ ///
+ /// 取走并快速中止当前会话（须持 ）：结算时长指标、取消会话
+ /// 令牌、唤醒解码读取、关闭会话 socket。关 socket 在等待任务之前——
+ /// netfx 上取消令牌无法中止已提交的接收，只有关 socket 能立即解锁阻塞 I/O（先中止
+ /// I/O 再等待任务，#68）。锁内无任何等待；任务收尾由
+ /// 在锁外完成。
+ ///
+ private Session? DetachSession()
{
- if (_sessionCts is not { } cts)
- return;
+ var session = _session;
+ _session = null;
+ if (session is null)
+ return null;
// 会话时长指标：从建立到拆除的真实存活时间
- var startedAt = Interlocked.Exchange(ref _sessionStartedAt, 0);
- if (startedAt != 0)
- _metrics.SessionEnded((Stopwatch.GetTimestamp() - startedAt) / (double)Stopwatch.Frequency);
+ _metrics.SessionEnded((Stopwatch.GetTimestamp() - session.StartedAt) / (double)Stopwatch.Frequency);
- var pipe = _pipe;
-
- if (!cts.IsCancellationRequested)
- cts.Cancel();
+ if (!session.Cts.IsCancellationRequested)
+ session.Cts.Cancel();
// 唤醒可能阻塞在 ReadAsync 的解码循环（writer.CompleteAsync 由接收循环的
// finally 负责；这里是兜底，二者都会让解码循环自然收尾退出）
- pipe?.Reader.CancelPendingRead();
-
- var tasks = new[] { _receiveTask, _decodeTask, _sendWorkerTask };
+ session.Pipe.Reader.CancelPendingRead();
- _sessionCts = null;
- _receiveTask = null;
- _decodeTask = null;
- _sendWorkerTask = null;
- _pipe = null;
-
- var pending = tasks.Where(static task => task is not null).Cast().ToArray();
- if (pending.Length > 0)
+ // 只关闭本会话的 socket（不关闭则每次重连泄漏一个）：直连迁移中 _socket 字段
+ // 可能已指向新会话的连接，字段与旧 socket 相同时才同步置空。
+ try
{
- try
- {
- Task.WhenAll(pending).Wait(TimeSpan.FromSeconds(2));
- }
- catch (AggregateException)
- {
- // 会话任务在取消/断连时可能以异常结束，此处仅等待其退出。
- }
+ if (session.Socket.Connected)
+ session.Socket.Shutdown(SocketShutdown.Both);
}
-
- // 会话已死，关闭旧连接的 socket（不关闭则每次重连泄漏一个，等终结器收场）
- if (_socket is { } socket)
+ catch (Exception ex)
{
- try
- {
- if (socket.Connected)
- socket.Shutdown(SocketShutdown.Both);
- }
- catch (Exception ex)
- {
- _logger.LogDebug(ex, "Socket shutdown on session stop failed.");
- }
+ _logger.LogDebug(ex, "Socket shutdown on session stop failed.");
+ }
- socket.Dispose();
+ session.Socket.Dispose();
+ if (ReferenceEquals(_socket, session.Socket))
_socket = null;
+
+ return session;
+ }
+
+ /// 锁外有界等待会话任务退出并释放取消源（≤2 秒；任务异常仅观察，不外抛）。
+ private void WaitAndDisposeSession(Session? session)
+ {
+ if (session is null)
+ return;
+
+ try
+ {
+ Task.WhenAll(session.Tasks).Wait(TimeSpan.FromSeconds(2));
+ }
+ catch (AggregateException)
+ {
+ // 会话任务在取消/断连时可能以异常结束，此处仅等待其退出。
}
- cts.Dispose();
+ session.Cts.Dispose();
}
- private async Task ReceiveLoopAsync(PipeWriter writer, CancellationToken ct, int epoch)
+ /// 接收循环：只操作所属会话的资源（socket/Pipe/令牌）——直连迁移中字段被
+ /// 新会话接管也不受影响，退出仅由本会话的取消或 socket 关闭驱动。
+ private async Task ReceiveLoopAsync(Session session)
{
+ var socket = session.Socket;
+ var writer = session.Pipe.Writer;
+ var ct = session.Cts.Token;
+ var epoch = session.Epoch;
try
{
while (true)
{
- var socket = _socket;
- if (socket is null)
- break;
-
var memory = writer.GetMemory(_options.SocketReceiveBufferSize);
var count = await ReceiveWithIdleTimeoutAsync(socket, memory, ct).ConfigureAwait(false);
writer.Advance(count);
@@ -833,10 +854,15 @@ private async Task ReceiveWithIdleTimeoutAsync(Socket socket, Memory
}
}
- private async Task SendWorkerAsync(CancellationToken ct, long sessionId)
+ /// 发送 worker：只向所属会话的 socket 写出（编号校验防跨会话错发之外，
+ /// 连接目标也随会话固定——直连迁移不会把旧条目发到新连接上）。
+ private async Task SendWorkerAsync(Session session)
{
// 发送失败（socket 故障/会话失效）不在此吞掉：上抛给会话守护 → 断线重连，
// 未发送的普通消息留在连接级队列，由重连后的新 worker 继续发送。
+ var socket = session.Socket;
+ var sessionId = session.Id;
+ var ct = session.Cts.Token;
var reader = _sendQueue.Reader;
while (await reader.WaitToReadAsync(ct).ConfigureAwait(false))
{
@@ -850,13 +876,13 @@ private async Task SendWorkerAsync(CancellationToken ct, long sessionId)
if (!entry.IsSessionBound)
{
- await SendFramedAsync(entry.Message, ct).ConfigureAwait(false);
+ await SendFramedAsync(entry.Message, ct, socket).ConfigureAwait(false);
_metrics.FrameSent();
continue;
}
// 编号校验（防跨会话错发）：绑定旧会话的残留条目绝不能发到当前会话的 socket 上。
- // 覆盖 Connected→Connected 直连拆除（StartSession 内的 StopSessionCore 刻意不
+ // 覆盖 Connected→Connected 直连拆除（StartSession 内的会话清理刻意不
// 清扫，保护"刚注册的新条目"）等一切条目跨会话存活的路径——
// "会话绑定消息绝不转移到新会话重放"的保证在此闭合
if (entry.SessionId != sessionId)
@@ -872,7 +898,7 @@ private async Task SendWorkerAsync(CancellationToken ct, long sessionId)
try
{
- await SendFramedAsync(entry.Message, ct).ConfigureAwait(false);
+ await SendFramedAsync(entry.Message, ct, socket).ConfigureAwait(false);
entry.Complete(); // 整帧已写入本机 socket，任务成功完成
_metrics.FrameSent();
}
@@ -920,7 +946,7 @@ private int AdaptiveEncodeBufferSize
/// 自适应缓冲封顶：超过此值回退到配置初始值（防御异常尺寸的偶发大帧抬高常态水位）。
private const int MaxAdaptiveEncodeBufferBytes = 1024 * 1024;
- private async Task SendFramedAsync(TMessage message, CancellationToken ct)
+ private async Task SendFramedAsync(TMessage message, CancellationToken ct, Socket socket)
{
// 流式编码：单缓冲、零 memcpy（BeginFrame 占位 → codec 直接写 → EndFrame 回填/收尾）
if (_options.UseStreamingEncode && _framing is IStreamingFramer streaming)
@@ -930,7 +956,7 @@ private async Task SendFramedAsync(TMessage message, CancellationToken ct)
_codec.Encode(message, frame, ct);
streaming.EndFrame(frame);
RememberEncodeHighWater(frame.WrittenCount);
- await SendRawAsync(frame.WrittenMemory, ct).ConfigureAwait(false);
+ await SendRawAsync(frame.WrittenMemory, ct, socket).ConfigureAwait(false);
return;
}
@@ -940,7 +966,7 @@ private async Task SendFramedAsync(TMessage message, CancellationToken ct)
using var frame2 = new PooledBufferWriter(payload.WrittenCount + 16);
_framing.EncodeFrame(payload.WrittenSpan, frame2);
RememberEncodeHighWater(frame2.WrittenCount);
- await SendRawAsync(frame2.WrittenMemory, ct).ConfigureAwait(false);
+ await SendRawAsync(frame2.WrittenMemory, ct, socket).ConfigureAwait(false);
}
private void RememberEncodeHighWater(int frameBytes)
@@ -949,17 +975,15 @@ private void RememberEncodeHighWater(int frameBytes)
_encodeBufferHighWater = frameBytes;
}
- private async Task SendRawAsync(ReadOnlyMemory buffer, CancellationToken ct)
+ /// 向指定会话的 socket 逐段写出；socket 归会话所有（拆除时关闭），
+ /// 写出途中被关闭以 ObjectDisposedException 收尾——由 worker 收敛为会话失效。
+ private async Task SendRawAsync(ReadOnlyMemory buffer, CancellationToken ct, Socket socket)
{
await _sendLock.WaitAsync(ct).ConfigureAwait(false);
try
{
while (!buffer.IsEmpty)
{
- var socket = _socket;
- if (socket is null)
- throw new SessionFaultException("会话已结束，无可用连接。");
-
var length = await socket.SendAsync(buffer, SocketFlags.None, ct).ConfigureAwait(false);
// 按实际写出的分片回调：部分发送失败时，已上线字节也可见
@@ -1082,6 +1106,41 @@ private static SessionExpiredException NewSessionExpired(long sessionId)
private static void ObserveLater(Task task)
=> _ = task.ContinueWith(static completed => _ = completed.Exception, TaskScheduler.Default);
+ ///
+ /// 一次会话的资源集合（#68）：一次 TCP 连接对应一个不可变的会话对象，集中持有该会话的
+ /// socket、Pipe、取消源、接收/解码/发送三个任务与编号/纪元/建立时刻。会话任务只操作
+ /// 所属会话的资源——直连迁移、拆除竞速中新旧会话互不干扰；连接级状态（编号计数器、
+ /// 当前编号、纪元）不在此处（跨会话单调，归连接所有）。
+ ///
+ private sealed class Session
+ {
+ public Session(long id, int epoch, Socket socket, Pipe pipe, CancellationTokenSource cts, long startedAt)
+ {
+ Id = id;
+ Epoch = epoch;
+ Socket = socket;
+ Pipe = pipe;
+ Cts = cts;
+ StartedAt = startedAt;
+ }
+
+ public long Id { get; }
+ public int Epoch { get; }
+
+ /// 本会话的连接 socket：I/O 唯一入口，拆除时由 关闭。
+ public Socket Socket { get; }
+
+ public Pipe Pipe { get; }
+ public CancellationTokenSource Cts { get; }
+ public long StartedAt { get; }
+
+ /// 三个会话任务（含守护包装）。在 _sessionGate 内、会话发布到 _session 之前装配。
+ public Task[] Tasks { get; private set; } = Array.Empty();
+
+ public void SetTasks(Task receive, Task decode, Task sendWorker)
+ => Tasks = new[] { receive, decode, sendWorker };
+ }
+
///
/// 发送队列条目。普通条目（）无完成源：入队即完成调用方任务，
/// 未发送的消息跨会话续发（既有语义）。会话绑定条目（）
diff --git a/test/StreamFrame.Tests/SendQueueContinuationTests.cs b/test/StreamFrame.Tests/SendQueueContinuationTests.cs
index 3c9dc00..0c1f841 100644
--- a/test/StreamFrame.Tests/SendQueueContinuationTests.cs
+++ b/test/StreamFrame.Tests/SendQueueContinuationTests.cs
@@ -32,11 +32,23 @@ public async Task CancelledWorker_AfterSuccessfulWrite_LeavesUnclaimedPlainEntry
await SoakRun.BoundedAsync(peer1.ConnectAsync(IPAddress.Loopback, port), "first regression peer connect");
await UntilAsync(() => server.CurrentSessionId != 0);
var oldSession = server.CurrentSessionId;
- var workerField = typeof(StreamConnection).GetField("_sendWorkerTask",
+ // 会话资源集中在内部 Session 对象上（#68）：经 _session 取发送 worker 任务。
+ // Connected 发布先于会话装配完成，轮询等待 worker 就位。
+ // 嵌套类型反射须从开放泛型定义构造后再闭合，否则属性带未绑定泛型参数无法取值。
+ var sessionField = typeof(StreamConnection).GetField("_session",
System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!;
- // Connected is published before StartSession finishes; wait for the worker to be installed.
- await UntilAsync(() => workerField.GetValue(server) is Task);
- var oldWorker = (Task)workerField.GetValue(server)!;
+ var tasksProperty = typeof(StreamConnection<>)
+ .GetNestedType("Session", System.Reflection.BindingFlags.NonPublic)!
+ .MakeGenericType(typeof(string))
+ .GetProperty("Tasks",
+ System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance
+ | System.Reflection.BindingFlags.Public)!;
+ Task[]? TasksOf(StreamConnection c)
+ => sessionField.GetValue(c) is { } s && tasksProperty.GetValue(s) is Task[] tasks
+ ? tasks
+ : null;
+ await UntilAsync(() => TasksOf(server) is { Length: 3 });
+ var oldWorker = TasksOf(server)![2];
try
{
await server.SendAsync("already-written");
diff --git a/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs b/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs
index 0699684..6d0d841 100644
--- a/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs
+++ b/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs
@@ -68,12 +68,9 @@ private static void AssertTerminalInvariants(StreamConnection connection
Assert.True(connection.IsDisposed);
Assert.Equal(ConnectionState.Disconnected, connection.State);
Assert.Equal(0, connection.CurrentSessionId);
- Assert.Null(Field(connection, "_server")); // 监听器已释放
- Assert.Null(Field(connection, "_socket")); // 连接 socket 已释放
- Assert.Null(Field(connection, "_sessionCts")); // 无孤儿会话
- Assert.Null(Field(connection, "_receiveTask"));
- Assert.Null(Field(connection, "_decodeTask"));
- Assert.Null(Field(connection, "_sendWorkerTask"));
+ Assert.Null(Field(connection, "_server")); // 监听器已释放
+ Assert.Null(Field(connection, "_socket")); // 连接 socket 已释放
+ Assert.Null(Field(connection, "_session")); // 无孤儿会话（会话对象集中持有全部会话资源）
// 监听端口可立即重新占用（未泄漏仍在监听的 socket）
var rebinding = new TcpListener(IPAddress.Loopback, port);
@@ -105,9 +102,8 @@ await Task.WhenAll(server.WaitForConnectedAsync(), client.WaitForConnectedAsync(
PublishState(server, ConnectionState.Connected);
Assert.Equal(ConnectionState.Disconnected, server.State);
Assert.Equal(0, server.CurrentSessionId);
- Assert.Null(Field(server, "_sessionCts"));
- Assert.Null(Field(server, "_receiveTask"));
- Assert.Null(Field(server, "_whenConnected")); // 等待器不被迟到的 Connected 唤醒
+ Assert.Null(Field(server, "_session")); // 迟到的 Connected 不得创建会话
+ Assert.Null(Field(server, "_whenConnected")); // 等待器不被迟到的 Connected 唤醒
AssertTerminalInvariants(server, port);
}