From ebeb2798c7ae1b821b2dae72fd5e68867f9f8918 Mon Sep 17 00:00:00 2001 From: Marci Date: Thu, 15 Oct 2026 18:19:52 +0800 Subject: [PATCH 1/3] =?UTF-8?q?refactor:=20=E5=86=85=E9=83=A8=20Session=20?= =?UTF-8?q?=E5=AF=B9=E8=B1=A1=E9=9B=86=E4=B8=AD=E4=BC=9A=E8=AF=9D=E8=B5=84?= =?UTF-8?q?=E6=BA=90=EF=BC=8C=E4=BB=BB=E5=8A=A1=E5=8F=AA=E6=93=8D=E4=BD=9C?= =?UTF-8?q?=E6=89=80=E5=B1=9E=E4=BC=9A=E8=AF=9D=EF=BC=9B=E6=8B=86=E9=99=A4?= =?UTF-8?q?=E5=88=86=E6=AE=B5=EF=BC=88=E9=94=81=E5=86=85=E4=B8=AD=E6=AD=A2?= =?UTF-8?q?+=E9=94=81=E5=A4=96=E6=9C=89=E7=95=8C=E7=AD=89=E5=BE=85?= =?UTF-8?q?=EF=BC=89=EF=BC=88#68=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Connection/StreamConnection.cs | 233 +++++++++++------- 1 file changed, 146 insertions(+), 87 deletions(-) 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 }; + } + /// /// 发送队列条目。普通条目（）无完成源：入队即完成调用方任务， /// 未发送的消息跨会话续发（既有语义）。会话绑定条目（） From 3fd262a34c1d31c51128d835b756e64fab13bf7e Mon Sep 17 00:00:00 2001 From: Marci Date: Thu, 15 Oct 2026 18:19:53 +0800 Subject: [PATCH 2/3] =?UTF-8?q?test:=20=E9=80=82=E9=85=8D=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E5=AF=B9=E8=B1=A1=E5=8C=96=E7=9A=84=E5=86=85=E9=83=A8?= =?UTF-8?q?=E7=BB=93=E6=9E=84=E5=8F=8D=E5=B0=84=EF=BC=88#68=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../SendQueueContinuationTests.cs | 20 +++++++++++++++---- .../StateOwnershipBaselineTests.cs | 14 +++++-------- 2 files changed, 21 insertions(+), 13 deletions(-) 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); } From 16fa3b252118ed5b68d9bcbe043115b6e25c04e7 Mon Sep 17 00:00:00 2001 From: Marci Date: Thu, 15 Oct 2026 18:19:53 +0800 Subject: [PATCH 3/3] =?UTF-8?q?docs:=20DESIGN=20=E8=B5=84=E6=BA=90?= =?UTF-8?q?=E6=89=80=E6=9C=89=E6=9D=83/=E6=8B=86=E9=99=A4=E5=88=86?= =?UTF-8?q?=E6=AE=B5/=E5=B7=B2=E7=9F=A5=E9=99=90=E5=88=B6=E6=9B=B4?= =?UTF-8?q?=E6=96=B0=EF=BC=9BCHANGELOG=20Unreleased=20=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=EF=BC=88#68=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 1 + docs/DESIGN.md | 16 +++++++++------- 2 files changed, 10 insertions(+), 7 deletions(-) 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）