From bd06e9190f3e8f863fb715fda41da57e2d99dab6 Mon Sep 17 00:00:00 2001 From: Marci Date: Thu, 15 Oct 2026 17:01:29 +0800 Subject: [PATCH 1/3] =?UTF-8?q?fix:=20=E7=8A=B6=E6=80=81=E6=9C=BA=E5=81=9C?= =?UTF-8?q?=E6=9C=BA=E7=AB=9E=E9=80=9F=E5=AE=88=E9=97=A8=E2=80=94=E2=80=94?= =?UTF-8?q?=E5=8F=91=E5=B8=83=E6=AE=B5=E5=8E=9F=E5=AD=90=E5=8C=96=E3=80=81?= =?UTF-8?q?=E8=BF=9F=E5=88=B0=E7=BB=88=E6=80=81=E5=8F=91=E5=B8=83=E4=B8=A2?= =?UTF-8?q?=E5=BC=83=E3=80=81=E7=BA=AA=E5=85=83=E7=8E=B0=E8=A1=8C=E6=80=A7?= =?UTF-8?q?=E9=97=A8=E6=8E=A7=E6=9B=BF=E4=BB=A3=E5=85=9C=E5=BA=95=E9=87=8D?= =?UTF-8?q?=E5=BB=BA=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 | 155 ++++++++++++------ 1 file changed, 109 insertions(+), 46 deletions(-) diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs index 0216d22..fa782f6 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -79,9 +79,11 @@ public string? RemoteIpAddress #if NET9_0_OR_GREATER private readonly Lock _listenerGate = new(); // 监听器发布与停机释放互斥，不跨 await 持有。 private readonly Lock _sessionGate = new(); // System.Threading.Lock（net9+，比 monitor 锁更轻量） + private readonly Lock _publishGate = new(); // 状态发布段互斥（见 CommunicationStateChanging） #else private readonly object _listenerGate = new(); private readonly object _sessionGate = new(); + private readonly object _publishGate = new(); #endif private Pipe? _pipe; @@ -238,10 +240,29 @@ private async Task StartAsync(CancellationToken ct, int loopId) if (!IsActive && Volatile.Read(ref _acceptLoopId) != loopId) return; // 已被更新的接受循环取代：静默退出，不再竞争 accept - _socket = IsActive ? await ConnectAsync(ct).ConfigureAwait(false) : await AcceptAsync(ct, loopId).ConfigureAwait(false); - if (_socket is null) + var socket = IsActive ? await ConnectAsync(ct).ConfigureAwait(false) : await AcceptAsync(ct, loopId).ConfigureAwait(false); + if (socket is null) return; // 排队等到锁时已被取代（AcceptAsync 内部的代次检查） + // 停机与"连接已建立"的竞速守门（#68）：Shutdown 可能在传输建立完成后、 + // 发布前的窗口内完成——终态已广播、会话已拆除。此时发布 Connected 会翻转 + // 终态并重建会话；把 socket 留进 _socket 则无人负责关闭（Shutdown 已走过 + // 关闭点），泄漏到终结器。判定与赋值同锁于 Shutdown 末尾的关闭块：窗口闭合。 + lock (_sessionGate) + { + if (ct.IsCancellationRequested || IsDisposed) + { + socket.Dispose(); + return; + } + + // 被取代的旧 socket（回调内 Reconnect 重入、并发双循环的直连迁移）： + // 新连接接管字段时旧值已无人认领（其会话从未建立或已被拆除）， + // 当场释放不泄漏。正常路径此处恒为 null（会话拆除时已清理）。 + _socket?.Dispose(); + _socket = socket; + } + _retryScheduler.Reset(); CommunicationStateChanging(ConnectionState.Connected); connected = true; @@ -491,32 +512,47 @@ private void ScheduleReconnect(int epoch) private void CommunicationStateChanging(ConnectionState newState, int? observedEpoch = null) { - // 迟到的过期故障（会话已被替换，epoch 不匹配）：在任何对外可见的发布之前丢弃。 - // 若放到 Retry 分支的 gate 内才检查（旧实现），State=Retry 的发布与 CurrentSessionId - // 归零已经污染了完全存活的会话——WaitForConnectedAsync 悬挂、活会话编号被误判失效。 - // gate 内的权威复查仍然保留，防检查与发布之间 epoch 变化的窄竞态。 - if (newState == ConnectionState.Retry && observedEpoch is { } staleEpoch - && Volatile.Read(ref _sessionEpoch) != staleEpoch) - return; - - // 公共会话编号的线性化点：Connected 对外发布（事件回调/等待器完成）之前完成分配； - // 会话纪元（epoch）同样在发布之前递增——发布后到达的旧纪元幽灵故障在函数入口 - // 即被判为过期丢弃，不会卷进"已发布、任务未创建"的新生会话（评审附带发现收口）。 - if (newState == ConnectionState.Connected) + // 发布段（_publishGate 锁内）：迟到/过期判定 + 编号纪元簿记 + State 写入。 + // 判定与写入必须原子：否则"已通过停机检查、尚未写入 State"的发布线程被抢占， + // 停机线程在此间隙写入 Disconnected 后，迟到的 Retry/Connected 会把终态翻转 + // （并发 Reconnect 与 Dispose 竞速实测可命中）。锁内无 await、无用户回调。 + lock (_publishGate) { - var id = checked(Interlocked.Increment(ref _sessionCounter)); - Volatile.Write(ref _currentSessionId, id); - Interlocked.Increment(ref _sessionEpoch); - } - else if (newState is ConnectionState.Retry or ConnectionState.Disconnected) - { - Volatile.Write(ref _currentSessionId, 0); - if (newState == ConnectionState.Retry) - _metrics.Reconnect(); // 过期幽灵已在函数入口被丢弃，不会计入 - } + // 停机后到达的迟到发布（竞速中的连接成功、用户 Reconnect、会话故障重连）整体 + // 丢弃：其后任何非终态发布都会把对外 State 翻成 Connected/Retry、重分配/归零 + // 会话编号——终态必须是最后对外可见的状态。Shutdown 自身的 Disconnected 发布 + // 不受影响（停机标记先于本调用置位，且停机是 Disconnected 的唯一来源）。 + if (IsDisposed && newState != ConnectionState.Disconnected) + return; + + // 迟到的过期故障（会话已被替换，epoch 不匹配）：在任何对外可见的发布之前丢弃。 + // 若放到 Retry 分支的 gate 内才检查（旧实现），State=Retry 的发布与 CurrentSessionId + // 归零已经污染了完全存活的会话——WaitForConnectedAsync 悬挂、活会话编号被误判失效。 + // gate 内的权威复查仍然保留，防检查与发布之间 epoch 变化的窄竞态。 + if (newState == ConnectionState.Retry && observedEpoch is { } staleEpoch + && Volatile.Read(ref _sessionEpoch) != staleEpoch) + return; - State = newState; + // 公共会话编号的线性化点：Connected 对外发布（事件回调/等待器完成）之前完成分配； + // 会话纪元（epoch）同样在发布之前递增——发布后到达的旧纪元幽灵故障在函数入口 + // 即被判为过期丢弃，不会卷进"已发布、任务未创建"的新生会话（评审附带发现收口）。 + if (newState == ConnectionState.Connected) + { + var id = checked(Interlocked.Increment(ref _sessionCounter)); + Volatile.Write(ref _currentSessionId, id); + Interlocked.Increment(ref _sessionEpoch); + } + else if (newState is ConnectionState.Retry or ConnectionState.Disconnected) + { + Volatile.Write(ref _currentSessionId, 0); + if (newState == ConnectionState.Retry) + _metrics.Reconnect(); // 过期幽灵已在锁内被丢弃，不会计入 + } + State = newState; + } + + // 通知段（锁外）：用户回调可能重入（回调内 Reconnect/Dispose），不得持有发布锁。 try { ConnectionChanged?.Invoke(this, newState); @@ -550,6 +586,11 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE { lock (_sessionGate) { + // 取锁等待期间停机可能已完成：不拆除、不派发新循环（_lifetimeCts + // 已被 Shutdown 置空，继续执行会对 null 解引用） + if (IsDisposed) + return; + if (observedEpoch is { } epoch && Volatile.Read(ref _sessionEpoch) != epoch) return; // 故障会话已被替换，避免误杀新会话 @@ -558,7 +599,18 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE StopSessionCore(); } - StartCore(_lifetimeCts!.Token); + // 取锁后停机仍可能已完成（_lifetimeCts 被 Shutdown 回收）：放弃派发， + // 静默让位终态；令牌源在取消与回收之间的窗口取 Token 可能抛 ODE，一并容忍。 + var lifetime = _lifetimeCts; + if (lifetime is null) + return; + try + { + StartCore(lifetime.Token); + } + catch (ObjectDisposedException) + { + } } finally { @@ -573,17 +625,23 @@ private void StartSession(long sessionId, int epoch) { lock (_sessionGate) { - StopSessionCore(); + // 停机守门（#68）：本方法唯一的迟到路径是"Connected 回调内同步 Dispose"—— + // 发布已过守门、停机随后完成，外层才轮到创建会话。终态后不再创建会话任务 + //（此时 _sendLock/_acceptLock 已释放，迟到的 worker 只会在已释放的锁上故障）。 + if (IsDisposed) + return; - // 病态重入兜底：Connected 回调里同步触发 Reconnect() 会让编号/纪元在 StartSession - // 执行前被归零/递增——此时重新分配两者，保证会话任务拿到的编号恒非零、 - // 纪元为最新（旧纪元的任务故障会被入口的过期检查丢弃）。 - if (sessionId == 0) - { - sessionId = checked(Interlocked.Increment(ref _sessionCounter)); - Volatile.Write(ref _currentSessionId, sessionId); - epoch = Interlocked.Increment(ref _sessionEpoch); - } + // 纪元现行性守门（#68）：本发布帧的纪元已被更新的重连取代——典型是 + // Connected 回调内同步 Reconnect()（回调里 Retry 已递增纪元并派发新循环）， + // 或发布窗口内现行纪元的故障重连。迟到帧不得以过期编号/纪元重建会话： + // 编号会被复用、StopSessionCore 会拆掉新会话刚建立的 socket（旧"兜底重建" + // 曾在该路径撕裂会话）。放弃创建，连接由取代者的循环完成；被取代帧的 + // socket 在新循环覆盖 _socket 时释放。sessionId==0 同理：编号已被归零 + // 说明拆除正在并发进行，等待取代者重建。 + if (Volatile.Read(ref _sessionEpoch) != epoch || sessionId == 0) + return; + + StopSessionCore(); var cts = new CancellationTokenSource(); var token = cts.Token; @@ -1182,19 +1240,24 @@ private void Shutdown() // 消息通道归连接所有：停机是它唯一的完成点（正常完成，枚举端自然结束） _messageRelay.Writer.TryComplete(); - if (_socket is { } socket) + // 与 StartAsync 的"停机守门 + 赋值"共用 _sessionGate：连接建立与停机互斥， + // 刚建立的 socket 不会漏进已完成的停机（#68）。 + lock (_sessionGate) { - try + if (_socket is { } socket) { - if (socket.Connected) - socket.Shutdown(SocketShutdown.Both); - } - catch (Exception ex) - { - _logger.LogDebug(ex, "Socket shutdown on disconnect failed."); + try + { + if (socket.Connected) + socket.Shutdown(SocketShutdown.Both); + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Socket shutdown on disconnect failed."); + } + socket.Dispose(); + _socket = null; } - socket.Dispose(); - _socket = null; } lock (_listenerGate) From a1e0d54018026db0de085cdff47a96296c54b25e Mon Sep 17 00:00:00 2001 From: Marci Date: Thu, 15 Oct 2026 17:01:29 +0800 Subject: [PATCH 2/3] =?UTF-8?q?test:=20=E7=8A=B6=E6=80=81=E6=9C=BA?= =?UTF-8?q?=E8=A1=8C=E4=B8=BA=E5=9F=BA=E7=BA=BF=E2=80=94=E2=80=94=E8=BF=9F?= =?UTF-8?q?=E5=88=B0=E5=8F=91=E5=B8=83=E9=97=A8=E6=8E=A7/=E5=9B=9E?= =?UTF-8?q?=E8=B0=83=E9=87=8D=E5=85=A5/=E5=81=9C=E6=9C=BA=E7=AB=9E?= =?UTF-8?q?=E9=80=9F=E4=B8=8D=E5=8F=98=E9=87=8F=EF=BC=88#68=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../StateOwnershipBaselineTests.cs | 306 ++++++++++++++++++ 1 file changed, 306 insertions(+) create mode 100644 test/StreamFrame.Tests/StateOwnershipBaselineTests.cs diff --git a/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs b/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs new file mode 100644 index 0000000..0699684 --- /dev/null +++ b/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs @@ -0,0 +1,306 @@ +using System.Collections.Concurrent; +using System.Net; +using System.Net.Sockets; +using System.Reflection; +using System.Threading.Channels; + +namespace StreamFrame.Tests; + +/// +/// 状态机与会话所有权行为基线（#68 PR-1）：以确定性或不变量测试固定当前已承诺的语义—— +/// 终态后迟到发布的丢弃、用户回调重入（Connected 回调内 Reconnect/Dispose）、 +/// 连接成功与停机竞速、并发关闭/重连下的终态不变量。为后续会话所有权重构提供行为底线。 +/// +public class StateOwnershipBaselineTests +{ + private static readonly TimeSpan Budget = TimeSpan.FromSeconds(10); + + private static int GetFreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + private static StreamConnection Create(int port, bool active, int retryDelayMs = 25) + => new( + new LengthPrefixFramer(), + StringCodec.Instance, + IPAddress.Loopback, + port, + isActive: active, + new StreamConnectionOptions { ConnectRetryDelayMs = retryDelayMs, AcceptRetryDelayMs = retryDelayMs }); + + private static object? Field(StreamConnection connection, string name) + => typeof(StreamConnection).GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(connection); + + /// 白盒直发一次内部状态发布（迟到发布的确定性门控测试用）。 + private static void PublishState(StreamConnection connection, ConnectionState state) + => typeof(StreamConnection) + .GetMethod("CommunicationStateChanging", BindingFlags.Instance | BindingFlags.NonPublic)! + .Invoke(connection, new object?[] { state, null }); + + private static async Task ReceiveOne(StreamConnection connection, int budgetMs = 5000) + { + using var timeout = new CancellationTokenSource(budgetMs); + await foreach (var message in connection.GetMessages(timeout.Token)) + return message; + throw new InvalidOperationException("消息流提前结束"); + } + + private static async Task WaitForAsync(Func condition, string what, int budgetMs = 8000) + { + var deadline = TestClock.TickCount64 + budgetMs; + while (TestClock.TickCount64 < deadline) + { + if (condition()) + return; + await Task.Delay(20); + } + throw new TimeoutException($"等待 {what} 超时"); + } + + /// 终态不变量：停机完成后 State/会话编号/会话资源必须收敛到唯一合法形态。 + private static void AssertTerminalInvariants(StreamConnection connection, int port) + { + 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")); + + // 监听端口可立即重新占用（未泄漏仍在监听的 socket） + var rebinding = new TcpListener(IPAddress.Loopback, port); + rebinding.Start(); + rebinding.Stop(); + } + + [Fact] + public async Task DisposedState_LatePublications_AreDroppedEntirely() + { + var port = GetFreePort(); + await using var server = Create(port, false); + await using var client = Create(port, true); + server.Start(CancellationToken.None); + client.Start(CancellationToken.None); + await Task.WhenAll(server.WaitForConnectedAsync(), client.WaitForConnectedAsync()) + .WaitAsync(Budget); + var liveSessionId = server.CurrentSessionId; + Assert.NotEqual(0, liveSessionId); + + await server.DisposeAsync(); + + // 停机后到达的迟到发布（用户 Reconnect / 迟到的会话故障）与竞速中的连接成功： + // 整体丢弃——终态不被翻转、会话编号不被归零后重分配、不创建会话任务。 + PublishState(server, ConnectionState.Retry); + Assert.Equal(ConnectionState.Disconnected, server.State); + Assert.Equal(0, server.CurrentSessionId); + + 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 唤醒 + + AssertTerminalInvariants(server, port); + } + + [Fact] + public async Task ConnectedCallback_ReconnectReentry_ConvergesToUsableSession() + { + var port = GetFreePort(); + await using var server = Create(port, false); + await using var firstClient = Create(port, true); + + var states = new ConcurrentQueue(); + var connectedSessionIds = new ConcurrentQueue(); + var reentered = 0; + server.ConnectionChanged += (_, state) => + { + states.Enqueue(state); + if (state != ConnectionState.Connected) + return; + + // 线性化契约：Connected 事件可见时编号必已分配（分配先于发布） + connectedSessionIds.Enqueue(server.CurrentSessionId); + + // 病态重入：仅首个 Connected 回调里同步触发 Reconnect() + if (Interlocked.CompareExchange(ref reentered, 1, 0) == 0) + server.Reconnect(); + }; + + server.Start(CancellationToken.None); + firstClient.Start(CancellationToken.None); + // 服务端就绪不走 WaitForConnectedAsync：回调内重连会消费掉 Connected#1 的等待器， + // 迟到的等待将挂到 Connected#2（第二个客户端之后才接入）。以客户端就绪 + 回调 + // 执行标志为准。 + await firstClient.WaitForConnectedAsync().WaitAsync(Budget); + await WaitForAsync(() => Volatile.Read(ref reentered) != 0, "首个 Connected 回调执行重入 Reconnect"); + var firstSessionId = connectedSessionIds.Single(); + + // 重连后的新会话由第二个客户端接入 + await using var secondClient = Create(port, true); + secondClient.Start(CancellationToken.None); + await secondClient.WaitForConnectedAsync().WaitAsync(Budget); + await WaitForAsync(() => server.State == ConnectionState.Connected && server.CurrentSessionId != firstSessionId, + "重入重连后的新会话发布"); + + // 收敛后的会话双向可用；编号在 Connected 可见期间恒非零且不被复用 + await secondClient.SendAsync("after-reentry-reconnect"); + Assert.Equal("after-reentry-reconnect", await ReceiveOne(server)); + await server.SendAsync("reply"); + Assert.Equal("reply", await ReceiveOne(secondClient)); + + var observedIds = connectedSessionIds.ToArray(); + var observedStates = states.ToArray(); + Assert.All(observedIds, id => Assert.NotEqual(0, id)); + Assert.Equal(2, observedIds.Length); + Assert.NotEqual(observedIds[0], observedIds[1]); + Assert.Equal(server.CurrentSessionId, observedIds[1]); + Assert.Equal(ConnectionState.Connected, server.State); + + // 观测到的状态序列只含合法转移，且终止于 Connected + Assert.Equal(ConnectionState.Connected, observedStates[^1]); + for (var i = 1; i < observedStates.Length; i++) + { + var legal = observedStates[i - 1] switch + { + ConnectionState.Connecting => observedStates[i] is ConnectionState.Connecting or ConnectionState.Connected or ConnectionState.Retry or ConnectionState.Disconnected, + ConnectionState.Connected => observedStates[i] is ConnectionState.Retry or ConnectionState.Connected or ConnectionState.Disconnected, + ConnectionState.Retry => observedStates[i] is ConnectionState.Retry or ConnectionState.Connecting or ConnectionState.Disconnected, + _ => false, // Disconnected 是终态，之后不得再有可观测转移 + }; + Assert.True(legal, $"非法状态转移：{observedStates[i - 1]} → {observedStates[i]}"); + } + } + + [Fact] + public async Task ConnectedCallback_DisposeReentry_TerminatesCleanly() + { + var port = GetFreePort(); + var server = Create(port, false); + await using var client = Create(port, true); + var disposedInCallback = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var disposeCompletedSynchronously = false; + + server.ConnectionChanged += (_, state) => + { + if (state == ConnectionState.Connected) + { + // 若停机挂起，回调无法返回：上方 WaitForConnectedAsync（完成点在回调之后）将超时暴露 + disposeCompletedSynchronously = server.DisposeAsync().IsCompletedSuccessfully; + disposedInCallback.TrySetResult(true); + } + }; + + server.Start(CancellationToken.None); + // 服务端等待器先挂起：Connected 发布（回调之后）的完成点在停机取消之后—— + // 回调内 Dispose 的合法结果就是等待器以取消收尾（#65 契约在重入场景同样成立） + var serverReady = server.WaitForConnectedAsync().WaitAsync(Budget); + client.Start(CancellationToken.None); + await client.WaitForConnectedAsync().WaitAsync(Budget); + await disposedInCallback.Task.WaitAsync(Budget); + Assert.True(disposeCompletedSynchronously); + await Assert.ThrowsAnyAsync(() => serverReady); + + AssertTerminalInvariants(server, port); + Assert.Equal(0, server.CurrentSessionId); + + // 停机后的公共契约：消息流自然结束、发送通道关闭 + using var messageBudget = new CancellationTokenSource(Budget); + await using (var enumerator = server.GetMessages(messageBudget.Token).GetAsyncEnumerator()) + Assert.False(await enumerator.MoveNextAsync()); // 通道已完成：立即 false；若悬挂则由预算令牌超时暴露 + await Assert.ThrowsAnyAsync(() => server.SendAsync("late")); + await client.DisposeAsync(); + } + + [Theory] + [InlineData(0)] + [InlineData(1)] + [InlineData(2)] + public async Task ConnectSuccess_RacingDispose_TerminalInvariantsHold(int jitterMode) + { + // 连接建立与停机的竞速不变量：无论 accept 成功与 Dispose 的相对次序如何， + // 终态收敛且不泄漏（连接 socket / 监听器 / 孤儿会话任务）。 + // 迭代量刻意保守：这是行为基线测试，不是 soak；过高并发会放大其它测试的 + // 慢收敛敏感（StopSessionCore 同步等待），干扰并行套件。 + const int iterations = 10; + for (var i = 0; i < iterations; i++) + { + var port = GetFreePort(); + await using var server = Create(port, false); + server.Start(CancellationToken.None); + + using var client = new TcpClient(); + var connect = Task.Run(async () => + { + try + { + if (jitterMode == 1) + await Task.Delay(1); + await client.ConnectAsync(IPAddress.Loopback, port).WaitAsync(Budget); + } + catch (SocketException) + { + // 服务端先停机：连接被拒，属预期分支 + } + catch (OperationCanceledException) + { + } + }); + var dispose = Task.Run(async () => + { + if (jitterMode == 2) + await Task.Delay(1); + await server.DisposeAsync(); + }); + + await Task.WhenAll(connect, dispose).WaitAsync(Budget); + client.Dispose(); + AssertTerminalInvariants(server, port); + } + } + + [Fact] + public async Task ConcurrentDisposeAndReconnect_TerminalInvariantsHold() + { + // 故障重连进行中并发执行用户 Reconnect 与 Dispose：全部参与者收敛后终态不变量成立。 + // 迭代量刻意保守（同上：行为基线而非 soak，避免放大并行套件的慢收敛敏感）。 + const int iterations = 8; + for (var i = 0; i < iterations; i++) + { + var port = GetFreePort(); + var server = Create(port, false); + await using var client = Create(port, true); + server.Start(CancellationToken.None); + client.Start(CancellationToken.None); + await Task.WhenAll(server.WaitForConnectedAsync(), client.WaitForConnectedAsync()) + .WaitAsync(Budget); + + // 对端拆除 → 服务端进入故障重连（重新监听），与并发 Reconnect/Dispose 竞速 + await client.DisposeAsync(); + + var actors = Enumerable.Range(0, 4).Select(n => Task.Run(() => + { + try + { + server.Reconnect(); + } + catch (ObjectDisposedException) + { + // 已停机：合法竞速结果 + } + })).Append(Task.Run(() => server.DisposeAsync().AsTask())); + + await Task.WhenAll(actors).WaitAsync(Budget); + AssertTerminalInvariants(server, port); + } + } +} From 9ea471bf76d09dcc273937980c08538cddcc53a8 Mon Sep 17 00:00:00 2001 From: Marci Date: Thu, 15 Oct 2026 17:02:17 +0800 Subject: [PATCH 3/3] =?UTF-8?q?docs:=20DESIGN=20=E5=A2=9E=E8=A1=A5?= =?UTF-8?q?=E7=8A=B6=E6=80=81=E6=9C=BA=E4=B8=8E=E6=89=80=E6=9C=89=E6=9D=83?= =?UTF-8?q?=E8=A7=84=E8=8C=83=EF=BC=9BCHANGELOG=20Unreleased=20=E8=AE=B0?= =?UTF-8?q?=E5=BD=95=E7=BB=88=E6=80=81=E7=AB=9E=E9=80=9F=E5=AE=88=E9=97=A8?= =?UTF-8?q?=E4=B8=8E=E5=9B=9E=E8=B0=83=E9=87=8D=E5=85=A5=E4=BF=AE=E5=A4=8D?= =?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 | 3 +++ docs/DESIGN.md | 39 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a63ecd5..0f01d31 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ ## [Unreleased] ### 修复 +- **终态竞速守门与发布段原子化（[#68](https://github.com/CSJ608/StreamFrame/issues/68)）**：并发 Dispose 与 Reconnect/连接成功竞速时，已通过停机检查但尚未写入 State 的发布被抢占，停机写入 Disconnected 后迟到的 Retry/Connected 会把终态翻转（State 停在 Retry/Connected）；Retry 派发还会对停机回收的 `_lifetimeCts` 空引用。状态发布收敛为发布段（`_publishGate` 锁内原子：停机/过期纪元判定 + 编号纪元簿记 + State 写入）与通知段（锁外用户回调/等待器/派发，回调可安全重入）；停机后到达的非终态发布整体丢弃；传输建立与停机竞速时新 socket 在守门处当场释放（此前泄漏到终结器）。新增迟到发布确定性门控、连接成功与停机竞速、并发关闭/重连终态不变量回归。 +- **回调重入会话撕裂（[#68](https://github.com/CSJ608/StreamFrame/issues/68)）**：Connected 回调内同步 `Reconnect()` 时，迟到发布帧的 `StartSession` 以过期编号/纪元"兜底重建"僵尸会话，其拆除会关掉新会话刚建立的 socket（编号复用 + 会话失活，实测可复现）。改为纪元现行性门控：过期帧放弃创建，连接由取代者循环完成；`_socket` 被新连接覆盖时旧值当场释放（直连迁移路径不再泄漏）。新增回调重入（Reconnect/DisposeAsync）收敛性、合法状态转移序列与编号线性化回归。 - **基准工作量对齐（[#70](https://github.com/CSJ608/StreamFrame/issues/70)）**：单向不再回显；直接 TCP 与框架共享负载、定界、Codec、物化校验及完成点，覆盖双 Framer、64B/1KB/64KB 和字节/两种字符串编码，往返逐条等待回显。增加全矩阵连续批次验证；原始结果与环境随仓库保存，撤回不同工作量相减所得的框架税结论，修正指标单位和筛选命令，不修改框架实现。 - **NativeAOT 运行验收（[#71](https://github.com/CSJ608/StreamFrame/issues/71)）**：CI 发布后执行 linux-x64 原生产物；有界等待连接与三条消息，校验会话绑定/普通发送及内容，观察消费任务并释放资源。动态选择端口并仅在连接阶段有限重选；连接失败、少收、消费异常和内容错误以非零退出，通过受控故障反证检查有效。保留警告为错误和非必需 AOT 检查。 - **Soak 队列续发与收尾（[#67](https://github.com/CSJ608/StreamFrame/issues/67)）**：发送 worker 在每次出队前检查会话取消，避免上一帧写完后拆除的旧 worker 取走并丢弃待续发普通条目；新增确定性门控回归。长流测试精确排空后关闭读 socket 并观察任务，修复 net48 仅取消令牌时读任务不收敛。 @@ -16,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-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 粒度和入队非确认。示例纳入三目标编译并以主动覆写缓冲验证异步消费；不改变默认值或实现公共契约。 - **Soak 可观测协议（[#67](https://github.com/CSJ608/StreamFrame/issues/67)）**：保持 SendAsync 仅入队契约，记录可重放 seed、动作、尝试 ID、会话及发送/读取阶段；业务测试使用 ACK、重试、去重并精确校验，重试前独立检查队列续发与会话边界。所有阶段有界等待，未启用长测报告 Skip；完整 600 秒五目标连续三次验收另行跟踪，见 docs/SOAK.md。 diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 449f6da..3a7daa1 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -142,6 +142,45 @@ socket --ReceiveAsync--> pipe.Writer --[Pipe]--> FrameDecoder.RunAsync **原始字节与诊断（1.2.0）**：`RawBytesReceived/Sent` 全量输出（含被定界器丢弃的噪声字节），发送侧按实际写出分片回调；两者与 `ConnectionChanged`、`FrameError` 的用户回调异常均被隔离，不反噬会话。`IFrameDiscardReporting`（可选接口，内置两个 codec 实现）让定界器精确上报重同步丢弃的字节，经 `FrameError(DiscardedByResync)` 交上层调试。 +### 状态机与所有权规范（#68 行为基线） + +> 本节是 `StreamConnection` 状态转换、资源所有权、锁范围与允许重入点的权威清单， +> 与 `StateOwnershipBaselineTests` 共同构成后续会话所有权重构的行为底线。 + +**状态转换全表**（`CommunicationStateChanging` 是唯一发布入口；`State` 只在发布段写入）： + +| 目标状态 | 触发源 | 发布点 | +|---|---|---| +| `Connecting` | `Start`/重连的 `StartAsync` 循环头（每轮尝试前） | 连接循环 | +| `Connected` | 传输建立成功（connect/accept 返回后、过停机守门） | 连接循环（唯一发布点） | +| `Retry` | 用户 `Reconnect()`；会话任务故障经 `ScheduleReconnect`（携带 epoch，过期即丢） | 用户线程 / 线程池 | +| `Disconnected` | `Shutdown`（Dispose 或生命周期令牌取消）唯一来源 | 停机路径 | + +发布段规则（`_publishGate` 锁内，原子）：① 停机后到达的非终态发布**整体丢弃**——终态必须是最后对外可见的状态（并发 Reconnect/连接成功与 Dispose 竞速时不被翻转）；② 过期纪元的故障发布在任何对外可见效果之前丢弃；③ 会话编号分配/归零与纪元递增先于 `State` 写入（线性化点：`ConnectionChanged` 回调与 `WaitForConnectedAsync` 完成时刻读取 `CurrentSessionId` 必得一致值）。通知段（锁外）：用户回调（异常隔离）、等待器完成、会话创建/拆除派发——回调可能重入，不得持有发布锁。 + +**会话创建的纪元现行性门控**：`StartSession` 只在"发布帧的纪元仍是最新、编号非零"时创建会话。Connected 回调内同步 `Reconnect()` 或发布窗口内的现行纪元故障会让位于取代者（其循环重建连接），迟到帧不再"兜底重建"——那曾以过期编号复用、拆除新会话刚建立的 socket（撕裂会话）。被取代帧的 socket 在新连接覆盖 `_socket` 时当场释放。 + +**资源所有权**： + +| 资源 | 创建者 | 释放者 | 备注 | +|---|---|---|---| +| `_socket`（会话 socket） | 连接循环（connect/accept） | 会话拆除（`StopSessionCore`）；新连接覆盖时；`Shutdown` 末尾兜底 | 赋值与关闭互斥于 `_sessionGate`；停机守门失败的连接当场释放 | +| `_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 | + +**锁与同步原语清单**：`_publishGate`（状态发布段，锁内无 await/回调）；`_sessionGate`（会话创建/拆除、`_socket` 赋值互斥；拆除等待最长 2s——已知同步等待，见"已知限制"）；`_listenerGate`（监听器发布与停机释放互斥）；`_acceptLock`（接受尝试串行化 + 代次门控落点）；`_sendLock`（socket 写串行化）；CAS 字段：`_retryInProgress`（重拆防重入）、`_acceptLoopId`（接受循环代次）、`_disposeStage`/`_started`、发送条目状态字。 + +**允许的重入点**（用户回调可调用任意公共 API，异常一律隔离）： + +- `Connected` 回调内 `Reconnect()`：合法（病态但支持）。回调内 Retry 先行（纪元递增 + 派发新循环），原发布帧的会话创建被纪元门控放弃；新连接由新循环完成，编号恒非零且不复用。 +- `Connected` 回调内 `DisposeAsync()`：合法。停机同步完成（会话尚未创建则无拆除等待）；原发布帧随后的会话创建被停机守门放弃；挂起等待器以取消收尾。 +- 停机后的任何调用：`Reconnect()`/`SendInSessionAsync` 等抛 `ObjectDisposedException` 或以会话失效收尾；迟到的内部状态发布被发布段丢弃。 +- 迟到的旧会话任务/故障：入口过期纪元检查丢弃，不误杀新会话（1.2.0/2.3.1 既有保证，重入路径由纪元门控补齐）。 + +**已知限制（后续 PR 方向）**：① `StopSessionCore` 持 `_sessionGate` 同步等待会话任务最长 2 秒——线程池饥饿时放大停机/重连延迟（高并发压力下可观测），#68 后续以"先中止 I/O、异步等待任务"改造；② `_socket`/会话五元组是"当前值"字段语义而非会话对象所有——并发双循环的直连迁移路径靠覆盖释放与拆除收敛，集中所有权（内部 Session 对象）是重构目标。 + ### netstandard2.0 兼容层（2.1.0） 多目标 `netstandard2.0;net8.0;net10.0`：