Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
## [Unreleased]

### 修复
- **被取代接受循环的状态杂音与迟到接入语义澄清([#68](https://github.com/CSJ608/StreamFrame/issues/68) PR-3,#47 世代语义边界定稿)**:用户重连与自动重连竞速产生的被取代"僵尸"接受循环——① 循环头代次复查前置:此前从重试退避中醒来的被取代循环会先发布 Connecting 再检查代次退出,状态序列混入替换者之外的杂音;② 明确并测试锁定迟到 accept 的收编语义:监听器是连接级共享资源,客户端向服务器端口的接入本身合法——被取代循环迟到完成的 accept **照常发布 Connected 并建立会话**(纪元机制与会话隔离收敛,取代者循环经代次检查静默退出)。曾评估"迟到连接当场释放"方案:会重置真实客户端、违背接入契约,已否决(实现一度引入后回归测试即复现客户端被吞)。新增确定性回归:accept 等待期间 Reconnect 取代循环,迟到客户端照常收编、状态序列恰为 Connecting→Retry→Connecting→Connected。
- **终态竞速守门与发布段原子化([#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 和字节/两种字符串编码,往返逐条等待回显。增加全矩阵连续批次验证;原始结果与环境随仓库保存,撤回不同工作量相减所得的框架税结论,修正指标单位和筛选命令,不修改框架实现。
Expand Down
2 changes: 1 addition & 1 deletion docs/DESIGN.md
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,7 @@ socket --ReceiveAsync--> pipe.Writer --[Pipe]--> FrameDecoder.RunAsync
- 停机后的任何调用:`Reconnect()`/`SendInSessionAsync` 等抛 `ObjectDisposedException` 或以会话失效收尾;迟到的内部状态发布被发布段丢弃。
- 迟到的旧会话任务/故障:入口过期纪元检查丢弃,不误杀新会话(1.2.0/2.3.1 既有保证,重入路径由纪元门控补齐)。

**已知限制(后续 PR 方向)**:① 停机与重连仍**同步**等待会话任务退出(锁外、有界 ≤2s)——公共 API 的同步/异步行为按 #68 约束保持不变;若未来允许 `DisposeAsync` 异步化,可把该等待移入后台收尾。② 接受循环被取代的"僵尸"帧在已取得 `_acceptLock` 且 accept 已完成时不再复查代次( tolerated path:其发布经纪元机制收敛,socket 覆盖时释放)——统一状态转换入口(每次发布前重验代次/停机)是剩余的结构性收敛项。
**已知限制/明确的竞速语义(#68 定稿)**:① 停机与重连仍**同步**等待会话任务退出(锁外、有界 ≤2s)——公共 API 的同步/异步行为按 #68 约束保持不变;若未来允许 `DisposeAsync` 异步化,可把该等待移入后台收尾。② 被取代循环建成的连接(被动端的迟到 accept、主动端的迟到 connect)**照常收编**:监听器是连接级共享资源,客户端向服务器端口发起的接入本身合法——被取代循环迟到完成的 accept 发布 Connected 并建立会话,由纪元机制与会话隔离收敛(取代者循环取锁后经代次检查静默退出)。曾评估"迟到连接当场释放"方案:会重置真实客户端、违背接入契约,已否决。代次门控的边界因此是**监听器生命周期(创建/关闭/接受串行化)与 Connecting 发布**(循环头复查前置——被取代循环不再发布 Connecting),而非客户端接入。

### netstandard2.0 兼容层(2.1.0)

Expand Down
10 changes: 7 additions & 3 deletions src/StreamFrame/Connection/StreamConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -237,12 +237,14 @@ private async Task StartAsync(CancellationToken ct, int loopId)
{
while (!connected && !ct.IsCancellationRequested && !IsDisposed)
{
// 代次复查前置(#68 统一发布入口):被取代的循环连 Connecting 也不发布,
// 状态序列不混入替换者之外的杂音
if (!IsActive && Volatile.Read(ref _acceptLoopId) != loopId)
return; // 已被更新的接受循环取代:静默退出,不再竞争 accept

CommunicationStateChanging(ConnectionState.Connecting);
try
{
if (!IsActive && Volatile.Read(ref _acceptLoopId) != loopId)
return; // 已被更新的接受循环取代:静默退出,不再竞争 accept

var socket = IsActive ? await ConnectAsync(ct).ConfigureAwait(false) : await AcceptAsync(ct, loopId).ConfigureAwait(false);
if (socket is null)
return; // 排队等到锁时已被取代(AcceptAsync 内部的代次检查)
Expand All @@ -251,6 +253,8 @@ private async Task StartAsync(CancellationToken ct, int loopId)
// 发布前的窗口内完成——终态已广播、会话已拆除。此时发布 Connected 会翻转
// 终态并重建会话;把 socket 留进 _socket 则无人负责关闭(Shutdown 已走过
// 关闭点),泄漏到终结器。判定与赋值同锁于 Shutdown 末尾的关闭块:窗口闭合。
// 注意:被取代循环的迟到 accept 不设代次门——监听器是连接级共享资源,
// 客户端向服务器端口的接入本身合法;收编后由纪元机制收敛(见 DESIGN)。
lock (_sessionGate)
{
if (ct.IsCancellationRequested || IsDisposed)
Expand Down
4 changes: 2 additions & 2 deletions test/StreamFrame.Tests/ReconnectWedgeTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -89,11 +89,11 @@ public async Task RapidServerInitiatedReconnects_PortStaysAcceptable()

for (var round = 0; round < rounds; round++)
{
// 对端接入形成会话
// 对端接入形成会话(预算与下方一致:探测永久不可用,并行负载下偶发慢收敛不应误报)
using (var client = new TcpClient())
{
await ConnectWithRetryAsync(client, port);
await WaitConnectedAsync(server, timeoutMs: 10_000);
await WaitConnectedAsync(server, timeoutMs: 30_000);
}

// 等自动重连完成、再次形成会话
Expand Down
33 changes: 33 additions & 0 deletions test/StreamFrame.Tests/StateOwnershipBaselineTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -217,6 +217,39 @@ public async Task ConnectedCallback_DisposeReentry_TerminatesCleanly()
await client.DisposeAsync();
}

[Fact]
public async Task ReconnectWhileAccepting_LateClientStillAdmittedBySupersededLoop()
{
var port = GetFreePort();
await using var server = Create(port, false);

var states = new ConcurrentQueue<ConnectionState>();
server.ConnectionChanged += (_, state) => states.Enqueue(state);

server.Start(CancellationToken.None);
await WaitForAsync(() => server.State == ConnectionState.Connecting, "接受循环进入监听等待");
// Start 同步推进到 accept 等待:循环 A 持有 _acceptLock、监听器已发布
Assert.IsType<Socket>(Field(server, "_server")!);

// 用户 Reconnect 在 accept 等待期间取代循环 A(Retry → StartCore 递增代次、派发循环 B;
// B 排队等锁)。Retry/Connecting 的发布在 Reconnect 返回前同步完成——被取代循环
// 不再发布 Connecting(代次复查前置),状态序列不混入杂音。
server.Reconnect();
Assert.Equal(
new[] { ConnectionState.Connecting, ConnectionState.Retry, ConnectionState.Connecting },
states.ToArray());

// 监听器是连接级共享资源:客户端向服务器端口的接入合法——迟到 accept 照常收编
// (发布 Connected、建立会话),由纪元机制收敛;取代者循环 B 取锁后经代次检查静默退出
using var client = new TcpClient();
await client.ConnectAsync(IPAddress.Loopback, port).WaitAsync(Budget);
await WaitForAsync(() => server.State == ConnectionState.Connected, "迟到客户端被收编并发布 Connected");
Assert.NotEqual(0, server.CurrentSessionId);
Assert.Equal(
new[] { ConnectionState.Connecting, ConnectionState.Retry, ConnectionState.Connecting, ConnectionState.Connected },
states.ToArray());
}

[Theory]
[InlineData(0)]
[InlineData(1)]
Expand Down
Loading