diff --git a/AGENTS.md b/AGENTS.md index 048bdda..4d630c0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -115,6 +115,7 @@ StdUnit.Tags.Core 硬件无关的核心抽象 + Schemas/tagsproje 7. **解析大小写敏感度**:`EndianKinds` / `TagAccessMode`(`LittleEndian`、`R1W` …)**大小写敏感**;`isEntry` / `isEnabled` **不敏感**。测试常覆盖这些边界。 8. **`IntentCapacity` 必须 > 0**,且只能在 `RunAsync` 之前设置。 9. **`isEnabled` 只作用于自动轮询**:`TagGrpRunner` 以 `TraversalMode.RespectEnabled` 遍历(自顶向下短路——父节点未使能,其整棵子树既不读也不写,脏标记保留);手动/外部意图里的显式 `ReadAsync`/`WriteAsync` 走 `TraversalMode.IgnoreEnabled`,**不检查使能**("平时禁用 + 需要时手动点动"是刻意支持的用法)。不要把公开的读写路径也改成检查使能,也不要让驱动自行判 `IsEnabled`。`ITagGrp` 的契约里只有 `ReadAsync(TraversalMode, ct)` / `WriteAsync(TraversalMode, ct)` 这一对「真身」,1 参版本(`ReadAsync(ct)`)是 `ITagGrpExtensions` 上的扩展方法、等价于传 `IgnoreEnabled`(`TagUnion` 上同理,两个重载都在 `TagUnionExtensions`)。**实现 `ITagGrp` 时必须把 mode 透传给子节点**,否则门控会在该层断掉。`ITagCbnt` / `ITag` **不加** mode / `IsEnabled`:门控的边界是 Cbnt(其下无递归层,mode 传下去没人用),且同一个 `Tag.IsEnabled` 在「塞进 Cbnt」(整块缓存 IO,拦不住)与「直挂 `ITagGrp`」(自己做 IO)两种挂法下行为不同,语义不自洽——代价是 `DirectTag` 想在自动轮询时单独停采只能套一层 `isEnabled="false"` 的 `TagGrp`。「不可写」仍用 `access="RO"` 表达。 +10. **启动是两阶段的(先算后启,事务性)**:`RunAsync` 先对全部入口算完逻辑组件匹配(`ILogicet.MatchEntry` / `Order`,纯计算),再准备 runner(建意图通道 / 订阅事件)、最后统一启动。因此:匹配/准备阶段失败 → **一个轮次都没开始**;启动阶段失败(含第三方 `StartAsync` 同步抛出)→ **取消并等待已启动的入口退出**再抛原异常。两条路径都**不能**留下 `Task` 无人等待、令牌永不取消的孤儿 runner。代价:各入口拿到的生命周期令牌是内部 linked 令牌(同一次运行内各回调仍是同一个实例,但**不保证** `==` 传给 `RunAsync` 的那一个),所以**不要用引用相等做判断**,取消时机才是契约。`MatchEntry` / `Order` 抛异常是**确定性的配置错误**:报 `TagsProjectConfigurationException`(`Location` = 入口路径,消息带逻辑组件类型),**不得**送入 `RunnerCrashed` 的重试/断开重连路径,也不得静默跳过该逻辑组件。`Enabled` 仍是每轮读取,运行期抛异常按普通轮询故障处理。 ## 约定 diff --git a/CHANGELOG.md b/CHANGELOG.md index b5b670f..aa80d78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,53 @@ 本文件记录**用户可见**的变化。版本号遵循[语义化版本](https://semver.org/lang/zh-CN/) `..`。 -## 2.0.0-beta6 (尚未发布) +## 2.0.0-beta7(尚未发布) + +> 本批仍是 `2.0.0` 的预发布——`2.0.0` 正式版保持未发布。 + +StdUnit.Tags / StdUnit.Tags.Core: **`ITagGrpRunner` 的三个回调由 `event` 改为单槽属性** + - `RunnerStarted` / `RunnerCrashed` / `TurnProcess` 由 `event` 改为 `{ get; set; }` 属性。 + runner 实例从不对外暴露(由 `TagsProject` 通过 `ITagGrpRunnerFactory` 创建并持有),唯一的使用方就是 `ITagsProject`, + 需要多播时由上层组合出**一个**委托再赋值。原来的 `event` 形态是**假承诺**:这些委托返回 `Task`, + 而 C# 的多播只返回最后一个处理器的 `Task`,照 `await TurnProcess(...)` 写就是"只等待最后一个、其余无人观察"。 + - 自定义 `ITagGrpRunner` 实现:把 `event` 声明改成属性;调用点改成"先取本地再判空" + (`var h = TurnProcess; if (h is not null) await h(grp, channel, ct);`)。订阅方写 `+=` 仍然可用(等价于读取 + 赋值)。 + - 虽然对外部开发者没有影响,不过这仍属于是**公开接口形态变更**,用旧版本编译的自定义 runner 实现需要重编。 + +StdUnit.Tags: **`ITagsProject.RunnerStarted` / `RunnerCrashed` 修复为真正的多播(形态仍是 `event`,无 API 变更)** + - 此前它们是"转发给 runner 时抓一次委托快照",`RunAsync` 之后订阅的观察者**永远收不到**通知;且当有 2 个订阅者时 runner 只 `await` 最后一个,其余 `Task` 无人观察;运行期 `-=` 也不生效。现在转发改为"**每次通知时读当前订阅者快照**",按注册顺序串行 await 全部订阅者。 + - 异常语义:任一订阅者抛异常即中断本次通知(排在它后面的订阅者不再被调用),异常原样抛出 + (保留类型与堆栈,不包 `AggregateException`)。不做"收集全部异常再抛第一个"——那样等于把其余静默丢弃; + `Core` 没有 logger 可以留痕,而"谁抛谁负责、后续不再执行"更符合直觉、也更确定。 + 抛出的异常仍等价于该入口本轮失败(`RunnerCrashed` 抛出 = 该入口停机),所以观察者请自己 `try/catch`。 + - `RunnerStarted` 的文档澄清为"**按入口**触发:每个入口各一次,崩溃恢复重连后再次触发";项目级启停仍看 + `ITagsProjectCtrl.StartedOrStopped`。回调仍在轮询线程上执行,多个订阅者的耗时**相加**。 + +StdUnit.Tags: 在beta6.0中短暂引入了**`ITagsProject.SetTurnProcessHandler`**,它是一个单槽钩子。现在为了支持多播场景,改为订阅式 **`SubscribeTurnProcess`** + - 新签名 `IDisposable SubscribeTurnProcess(string entry, TurnProcess handler)`:每个入口可有**任意多个**订阅者, + 按注册顺序**串行** await;任一订阅者抛异常即**中断**本次通知并原样抛出(= 该入口本轮失败,进入 `RunnerCrashed` + 的恢复路径),与 `RunnerStarted` / `RunnerCrashed` 的语义一致。 + - 返回的句柄就是退订凭据:`Dispose` 即退订,**幂等**;项目已释放后再 `Dispose` 是**无操作**(不抛)。 + - 旧 `SetTurnProcessHandler(entry, handler)`(替换语义、传 `null` 清除)**已删除**。迁移: + `proj.SetTurnProcessHandler("g1", h)` → `var sub = proj.SubscribeTurnProcess("g1", h)`, + 停止观测时 `sub.Dispose()`。`null` 回调现在抛 `ArgumentNullException`(退订请用句柄)。 + - 顺带修复:`StdUnit.Tags.WebUi` 的 hub 不再独占该槽位——改为订阅,并在项目停止/切换、hub 自己停止时退订, + 使用方自己的轮次观察者不会再被它静默覆盖。 + +StdUnit.Tags: **启动改为两阶段(事务性)** + - `RunAsync` 先对全部入口算完逻辑组件匹配(`ILogicet.MatchEntry` / `Order`),再建 runner 并启动轮询。启动窗口内任何失败 + (`MatchEntry` / `Order` 抛异常、`IntentCapacity` 非法、runner 工厂抛异常)都不会再留下**已经跑起来但无人等待**的入口—— + 旧实现里这些入口的 `StartAsync` 任务会被丢弃、生命周期令牌永不取消,在后台一直轮询。 + - `MatchEntry` / `Order` 抛出的异常现在报成 `TagsProjectConfigurationException`(`Location` = 入口路径,`InnerException` = 原异常, + 消息带逻辑组件类型),而不是从 `RunAsync` 同步抛出的 `AggregateException`;这是确定性的配置错误,**不会**重试,也不会触发 + `RunnerCrashed`。逻辑组件必须把它们实现成无副作用的纯判定:无法判断时返回 `false`。 + - 启动中途失败(含第三方 `ITagGrpRunner.StartAsync` 在返回 `Task` 之前直接抛出)现在会**取消并等待已启动的入口退出**, + 再把原异常抛出——不会再留下"孤儿轮询"(此前调用方连那些 `Task` 都拿不到)。 + 代价:入口各回调拿到的生命周期令牌不再与传给 `RunAsync` 的是同一个实例(同一次运行内仍彼此一致), + 请按"取消时机"而不是引用相等来判断。 + + +## 2.0.0-beta6 > 本批仍是 `2.0.0` 的预发布——`2.0.0` 正式版保持未发布。 @@ -11,6 +57,7 @@ StdUnit.Tags.WebUi: 改动点 按 `MapTagsStream()` 映射过端点的宿主无需额外代码。 + ## 2.0.0-beta5 > 本批仍是 `2.0.0` 的预发布——`2.0.0` 正式版保持未发布。 diff --git "a/docs/\345\274\202\345\270\270\345\244\204\347\220\206.md" "b/docs/\345\274\202\345\270\270\345\244\204\347\220\206.md" index b9303d9..cded0e1 100644 --- "a/docs/\345\274\202\345\270\270\345\244\204\347\220\206.md" +++ "b/docs/\345\274\202\345\270\270\345\244\204\347\220\206.md" @@ -53,6 +53,20 @@ catch (TagsProjectLoadException ex) 继承 `TagsProjectXmlException` / `TagsProjectAddressException` / `TagsProjectConfigurationException`, 或直接继承 `TagsProjectLoadException`——它们都能被 `catch (TagsProjectLoadException)` 接住,不会破坏使用方的统一处理。 +#### 启动期(`RunAsync`)的"入口 × 逻辑组件"匹配也算配置判定 + +`ILogicet.MatchEntry` / `ILogicet.Order` 是在项目**启动时**、对每个"入口 × 逻辑组件"组合求值的**纯判定**(不是轮询工作)。 +它们抛出异常意味着"这条配置无法判定",是**确定性**错误:重试、换设备、重启都不会变好。因此: + +- 报 `TagsProjectConfigurationException`(`Location` = 入口路径,如 `TagGrp(g2)`,`InnerException` = 原异常,消息带逻辑组件类型名); +- **不**走 `RunnerCrashed` 的重试/退避/断开重连恢复路径,也**不**"跳过这个逻辑组件"(静默失效); +- `RunAsync` 的启动是**两阶段**的(先算完全部匹配,再建 runner、启动轮询),所以匹配失败时**没有任何入口开始轮询**, + 也就不会留下"Task 无人等待、生命周期令牌永不取消"的孤儿 runner。 + +同样的口径适用于启动阶段的其它准备动作(建意图通道、runner 工厂):它们同样发生在任何 `StartAsync` 之前。 +`Enabled` 是**每轮**读取的,它在运行期抛异常仍按普通轮询故障处理(走 `RunnerCrashed`)。 +启动阶段(`ITagGrpRunner.StartAsync`)失败则**取消并等待已启动的入口退出**再抛原异常——不留孤儿轮询。 + ### **运行期**异常 在**运行期**,任何异常都需要被`ITagGrpRUnner`捕获进行重试——不论这是什么异常,也不论这个异常来自哪里。运行期发生的异常,是在本类库层面、还是在更底层的BCL层面,被单独捕获并识别带来的收益不大。 diff --git "a/docs/\350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214/0. \350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214.md" "b/docs/\350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214/0. \350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214.md" index 4b4bec8..33ea3f3 100644 --- "a/docs/\350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214/0. \350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214.md" +++ "b/docs/\350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214/0. \350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214.md" @@ -40,15 +40,18 @@ API 细节见教程站点()。 - 值的**受支持的观测方式**有两种: 1. **按测点**:`ITag.OnTagRead` / `ITag.OnTagWritten`,或 Rx / R3 的 `Watch()` / `ObserveOnRead()`; - 2. **按入口、按轮次**:`ITagsProject.SetTurnProcessHandler(entry, handler)`——某入口一轮结束(读取输入 + 逻辑处理之后、刷写输出之前)触发一次,适合"一次拿到该入口当前值"的观察者(如界面推送流)。回调签名是 `(entry, channel, ct)`,`ct` 是项目生命周期令牌,回调里的异步操作应透传它。 -- `SetTurnProcessHandler` **每个入口最多一个回调,重复设置是替换**(传 `null` 清除)。要多个观察者就自己组装成一个 `TurnProcess`——顺序在组合处显式可见,比"按注册顺序"更可控: + 2. **按入口、按轮次**:`ITagsProject.SubscribeTurnProcess(entry, handler)`(返回退订句柄)——某入口一轮结束(读取输入 + 逻辑处理之后、刷写输出之前)触发一次,适合"一次拿到该入口当前值"的观察者(如界面推送流)。回调签名是 `(entry, channel, ct)`,`ct` 是项目生命周期令牌,回调里的异步操作应透传它。 +- `SubscribeTurnProcess` **每个入口可以有任意多个订阅者**:按**注册顺序串行** await(总耗时相加),任一订阅者抛异常即**中断**本次通知并**原样抛出**(= 该入口本轮失败,进入 `RunnerCrashed` 的恢复路径),排在后面的订阅者本次不再被调用。所以观察者请自己 `try/catch`,也不要指望别人替你兜住异常。 ```csharp - proj.SetTurnProcessHandler("g1", async (entry, channel, ct) => { await 遥测(...); await 推送(...); }); + using var 遥测 = proj.SubscribeTurnProcess("g1", 遥测回调); + using var 推送 = proj.SubscribeTurnProcess("g1", 推送回调); ``` -- `SetTurnProcessHandler` **运行前、运行中都可以设置**:入口列表在 `Initialize` 之后固定,设置只是替换该入口的回调;运行期设置/清除**从下一轮开始生效**(不中断正在执行的那一轮),所以"观测端比轮询晚启动"(如 WebUi 的 hub 后启动)也能补挂上。设置无需清除:回调的作用域就是项目实例,项目停止即随之释放(项目已释放后调用抛 `ObjectDisposedException`)。但**每个入口只有一个回调**——要多个观察者就自己组装成一个 `TurnProcess`(见上),不要用反复设置/清除做"动态开关",那会与其它观察者互相覆盖;要动态开关就在自己的回调内部判断(一次集合查找)。 - - 历史:该 API 早期要求"只能在运行前设置"(与 `IntentCapacity` 同构)。那条约束只服务于"回调表运行期只读、轮询线程零同步"的实现便利,代价却是**晚挂观测静默降级**(结构树在、值流不在,只有一条 Warning 且不自愈),而顺序由调用方决定、库无法保证,因此改为写时复制的回调表(读路径仍是普通字典读、零锁)。 -- 启停用 `ObserveStartedOrStopped()` / `RunnerStarted`;注意 `SetTurnProcessHandler` **只覆盖"产生了轮次"**的情形,入口被禁用、项目未启动/已停止时不产生轮次,也就不会触发。 +- `SubscribeTurnProcess` **运行前、运行中都可以订阅**:入口列表在 `Initialize` 之后固定;运行期订阅/退订**从下一轮开始生效**(正在跑的那一轮用的是它开始时读到的快照,"刚退订"的那一次可能仍被调用一次),所以"观测端比轮询晚启动"(如 WebUi 的 hub 后启动)也能补挂上。**退订就是 Dispose 返回的句柄**(幂等;项目已释放后 Dispose 是无操作);项目已释放后**订阅**抛 `ObjectDisposedException`。"订阅/退订"做动态开关是受支持的用法——各订阅者互不覆盖。 + - 历史:该 API 早期是"每个入口一个回调、重复设置即替换",并要求"只能在运行前设置"。前者让库自带的观测端与使用方**互相静默覆盖**,后者带来**晚挂观测静默降级**(结构树在、值流不在,只有一条 Warning 且不自愈);两处现由"写时复制的订阅者数组 + 退订句柄"取代(读路径仍是普通字典读、零锁)。 +- 启停用 `ObserveStartedOrStopped()` / `RunnerStarted`;注意 `SubscribeTurnProcess` **只覆盖"产生了轮次"**的情形,入口被禁用、项目未启动/已停止时不产生轮次,也就不会触发。 +- `RunnerStarted` / `RunnerCrashed` 是**多播事件**:按注册顺序**串行** await 全部订阅者(总耗时相加),**运行期订阅/退订从下一次通知开始生效**;任一订阅者抛异常即**中断**本次通知并**原样抛出**(后面的订阅者本次不再被调用)——不做"收集全部再只抛第一个",那等于把其余异常静默丢弃。抛出的异常等价于该入口本轮失败(`RunnerCrashed` 抛出 = 该入口停机),所以观察者请自己 `try/catch`。`RunnerStarted` 是**按入口**触发的(每次建连、以及崩溃重连后各一次),不是"项目启动一次"——项目级启停看 `StartedOrStopped`。 - **取消不是崩溃**:`StopAsync` 导致的取消不会触发 `RunnerCrashed`,也不计入连续失败、不按重试策略退避(仍会断开该入口的全部通道)。因此进到 `RunnerCrashed` 基本等价于"真故障",可放心在那里做告警/计数。 +- **启动是两阶段的(事务性)**:先为全部入口算完逻辑组件匹配(`ILogicet.MatchEntry` / `Order`,纯计算),再准备 runner、订阅事件并启动轮询。所以启动失败时不会有"跑了一半"的项目,也不会留下"Task 无人等待、令牌永不取消"的孤儿 runner——启动阶段失败会**取消并等待已启动的入口退出**再抛出(取消靠内部 linked 令牌,因此各回调的令牌实例一致,但**不保证** `==` 传给 `RunAsync` 的那个)。匹配/排序抛异常按**配置错误**处理(`TagsProjectConfigurationException`,不会重试),而不是走 `RunnerCrashed` 的恢复路径;`Enabled` 是每轮读取的,它在运行期抛异常仍算普通轮询故障。 - 事件/回调都在**轮询线程**上同步触发,拿到的已经是本轮最终结果(值 + 时间戳),不需要、也不应该再去读字段做同步。 - **不同入口的回调并行**(各入口由各自的运行器驱动);回调必须尽快返回(要干重活就只投递到自己的队列,由独立循环消费)。回调抛出的异常不会被吞掉——等价于该入口本轮失败,会进入 `RunnerCrashed` 的恢复路径(例外:停机取消导致的 `OperationCanceledException` 不算失败),观察者请自行 try/catch。 - 直接读 `ITag.Value` 只适用于"同一轮询线程内"或"接受最终一致性"的场景(例如 `Watch(startWithCurrent: true)` 拿首帧)。**界面不要轮询字段**——那是事件该干的事。 diff --git a/src/StdUnit.Tags.Core/Exceptions/TagsProjectConfigurationException.cs b/src/StdUnit.Tags.Core/Exceptions/TagsProjectConfigurationException.cs index 8a4a8fc..cfa7a12 100644 --- a/src/StdUnit.Tags.Core/Exceptions/TagsProjectConfigurationException.cs +++ b/src/StdUnit.Tags.Core/Exceptions/TagsProjectConfigurationException.cs @@ -8,7 +8,12 @@ /// 统一成本类型后,用户不必再面对 NullReferenceException 或字典的 /// "An item with the same key has already been added"。
/// 与 的分工:前者是"构建过程中立刻发现",后者是 -/// "构建之前由校验器批量扫描后一次性报告"。 +/// "构建之前由校验器批量扫描后一次性报告"。
+///
+/// 启动期也用它: 为入口匹配逻辑组件时, +/// / 抛出异常意味着"这条配置无法判定"—— +/// 这是确定性的配置错误,重试不会变好,因此不走 的恢复路径, +/// 而是拒绝启动(此时没有任何入口开始轮询)。 /// public class TagsProjectConfigurationException : TagsProjectLoadException { diff --git a/src/StdUnit.Tags.Core/Exceptions/TagsProjectLoadException.cs b/src/StdUnit.Tags.Core/Exceptions/TagsProjectLoadException.cs index ab87600..05548dc 100644 --- a/src/StdUnit.Tags.Core/Exceptions/TagsProjectLoadException.cs +++ b/src/StdUnit.Tags.Core/Exceptions/TagsProjectLoadException.cs @@ -27,6 +27,12 @@ /// 一个核心的决策依据是,不管异常来自哪里,以及是什么原因导致,runner 都应该会统一 catch (Exception) 交给重试策略与 RunnerCrashed—— /// 所以不存在一个"能兜住本库全部异常"的根类型,请按加载期/运行期分别处理。 /// +/// +/// 启动期归属于本族: 在真正轮询之前为入口匹配逻辑组件 +/// ( / ),这也是配置判定而不是轮询工作—— +/// 它发生在任何 IO 之前、对它重试也不会变好,所以同样报 , +/// 而不是交给 RunnerCrashed 的重试/重连路径。 +/// /// public abstract class TagsProjectLoadException : Exception { diff --git a/src/StdUnit.Tags.Core/Plugins/ILogicet.cs b/src/StdUnit.Tags.Core/Plugins/ILogicet.cs index eca11f1..8491589 100644 --- a/src/StdUnit.Tags.Core/Plugins/ILogicet.cs +++ b/src/StdUnit.Tags.Core/Plugins/ILogicet.cs @@ -13,7 +13,9 @@ public interface ILogicet ITagsProject Project { get; } /// - /// 运行顺序 + /// 运行顺序(较小的先执行)。启动期读取一次( 为入口匹配逻辑组件时), + /// 必须无副作用、不阻塞、不得抛异常:抛出会被报成 (配置错误), + /// 而不是按轮询故障重试——重试对确定性的配置错误没有意义。 /// int Order { get; } @@ -23,7 +25,15 @@ public interface ILogicet bool Enabled { get; } /// - /// 是否能匹配入口?返回true表示应该处理当前entry,否则应该跳过处理 + /// 是否能匹配入口?返回true表示应该处理当前entry,否则应该跳过处理。
+ ///
+ /// 调用时机: 启动时,对每个"入口 × 逻辑组件"组合调用一次(重新启动会重新求值)。
+ /// 契约:必须是纯函数——只读已加载的配置、幂等、不阻塞、不做 IO/等待,不得抛异常。
+ /// - 判定不出来就返回 false,不要把"要不要匹配"变成一次可能失败的操作;
+ /// - 真正的配置错误请在构造函数 / 加载期报告();
+ /// - 一旦抛异常,框架会把它报成 并拒绝启动项目—— + /// 这是确定性的配置错误,既不会重试,也不会走 的断开重连恢复路径; + /// 而且此时没有任何入口开始轮询(框架先算完所有入口的匹配,再启动)。 ///
/// /// @@ -36,7 +46,9 @@ public interface ILogicet /// 入口的主通道( 的解析结果,可能为 null) /// /// 项目的生命周期令牌:项目停止()时取消。
- /// 它不是"单轮超时"令牌——框架不会为某一轮单独取消,同一入口每一轮拿到的都是同一个令牌。
+ /// 它不是"单轮超时"令牌——框架不会为某一轮单独取消,同一入口每一轮拿到的都是同一个实例 + /// (同一次运行内所有回调共享它;但它不必与传给 的令牌是同一个实例, + /// 也不要拿它们做 == 比较,"取消时机"才是契约)。
/// 建议用 if (ct.IsCancellationRequested) return; 提前结束本轮;不要在这里抛 /// ,令牌没取消时抛它会被运行器当成一次失败(触发 并按策略断开重连)。
/// 不要缓存它、也不要每轮 Register;它也不能让本方法变成"可以慢慢跑"——本方法仍必须尽快返回。 diff --git a/src/StdUnit.Tags.Core/Projects/ITagsProject.cs b/src/StdUnit.Tags.Core/Projects/ITagsProject.cs index 454a32b..8878c74 100644 --- a/src/StdUnit.Tags.Core/Projects/ITagsProject.cs +++ b/src/StdUnit.Tags.Core/Projects/ITagsProject.cs @@ -55,69 +55,73 @@ public interface ITagsProject : IDisposable XElement? GetRootElement(); /// - /// 轮询开始,会透传给。
+ /// 某个入口的运行器开始工作(含崩溃恢复后的重连):每个入口各触发一次,不是"项目启动一次", + /// 它的运行器会再次进入轮询循环时也会再触发。要观测"项目整体启停"请用 + /// 。
+ /// 多播:按注册顺序串行 await 全部订阅者(总耗时相加);运行期订阅/退订从下一次通知开始生效。 + /// 任一订阅者抛异常即中断(后续订阅者本次不再被调用),异常原样抛出——等价于该入口本轮失败, + /// 进入 的恢复路径。
+ /// 回调运行在轮询线程上,必须尽快返回(重活请投递到自己的队列)。观察者若比项目活得短,需要自己 -=。
/// 回调拿到的是项目的生命周期取消令牌( 时取消)。 ///
event RunnerStarted? RunnerStarted; /// - /// 轮询崩溃,会透传给。
+ /// 某个入口的轮询崩溃,透传自该入口运行器的 。
/// 取消不是崩溃:项目停止导致的取消不会触发本事件。
/// 通常来说,在该委托正常结束后,项目会自行重启轮询循环。
/// 但需要注意的是,这个自行重启是指事件委托没有再向外抛出异常的情况:
- /// 在中外抛异常会被视作对应入口需要停机! + /// 在中外抛异常会被视作对应入口需要停机!
+ /// 多播:按注册顺序串行 await 全部订阅者;任一订阅者抛异常即中断(排在它后面的订阅者本次收不到通知), + /// 异常原样抛出并按"该入口停机"处理。所以告警/计数类观察者请自己 try/catch——不要把"自己不出错"的责任交给别人。
+ /// 运行期订阅/退订从下一次通知开始生效;回调运行在轮询线程上,必须尽快返回。 ///
event RunnerCrashed? RunnerCrashed; /// - /// 为指定入口设置"轮次处理完成"回调(读取输入 + 逻辑处理 之后、刷写输出 之前触发), - /// 等价于该入口的 。
- /// 每个入口最多一个回调;重复设置是替换(不是追加)。需要多个观察者时,请自行组合成一个 - /// 再设置——这样调用顺序由你在组合处显式决定,比"按注册顺序"更可控:
- /// - /// proj.SetTurnProcessHandler("g1", async (entry, channel, ct) => - /// { - /// await 遥测(entry, channel, ct); - /// await 推送(entry, channel, ct); - /// }); - /// + /// 为指定入口订阅"轮次处理完成"回调(读取输入 + 逻辑处理 之后、刷写输出 之前触发一次), + /// 即该入口运行器 的多播入口。
///
- /// 运行前、运行中都可以设置:入口列表在 之后固定,本方法只替换该入口的回调; - /// 运行期设置/清除从下一轮开始生效(不中断正在执行的那一轮)。常见设置点是 + /// 多播:每个入口可以有任意多个订阅者,按注册顺序串行 await(总耗时相加); + /// 任一订阅者抛异常即中断本次通知并原样抛出(等价于该入口本轮失败,进入 + /// 的恢复路径)——排在它后面的订阅者本次不会被调用, + /// 所以观察者请自己 try/catch,不要让观测逻辑影响轮询,也别指望别人替你兜住异常。
+ ///
+ /// 返回的句柄就是退订凭据: 即退订,幂等; + /// 项目已释放之后再 Dispose 是无操作(不抛)。同一个委托实例重复订阅是多个独立订阅,各自 Dispose。
+ ///
+ /// 运行前、运行中都可以订阅:入口列表在 之后固定。 + /// 运行期订阅/退订从下一轮开始生效:正在执行的那一轮用的是它开始时读到的快照, + /// 因此"刚退订"的那一次仍可能被调用一次(不会并发,也不会打乱顺序)。常见订阅点是 /// /// 的 hook 或 的"已启动"事件(两者都在轮询开始前同步执行), /// 但晚挂同样有效——例如观测端(界面推送流)比轮询晚启动时,可以直接在项目运行中补挂。
- /// 不要用"反复设置/清除"来做动态开关:每个入口只有一个回调,那样会与其它观察者互相覆盖; - /// 要动态开关就在自己的回调内部判断(例如先检查订阅者集合是否为空再直接返回)。
///
- /// 回调运行在轮询线程上,必须尽快返回:不要阻塞,也不要在回调里做 IO / 序列化。 + /// 回调运行在轮询线程上,必须尽快返回:不要阻塞,也不要在回调里做 IO / 序列化; /// 需要较重的处理时,请只把必要信息投递到自己的队列,由独立的后台循环消费。
- ///
- /// 不同入口的回调是并行的(各入口由各自的运行器并行驱动)。同一个回调实例若被设置到多个入口, + /// 不同入口的回调是并行的(各入口由各自的运行器并行驱动):同一个委托实例若订阅到多个入口, /// 必须自行保证线程安全。
- ///
- /// 回调签名是 (entry, channel, ct):ct 是项目的生命周期取消令牌。 + /// 回调签名是 (entry, channel, ct):ct 是项目的生命周期取消令牌, /// 回调里发起的异步操作请把它透传下去,否则"停止项目"要等到 的等待超时才收场。
///
- /// 回调抛出的异常不会被吞掉:它等价于该入口本轮失败(与 抛异常同义), - /// 会进入 的恢复路径。观察者请自行 try/catch,不要让观测逻辑影响轮询。
- ///
/// 注意: /// /// 必须在 之后调用(运行期也可以:从下一轮开始生效); /// 入口名不存在时抛 (与 WriteIntent 的校验一致); - /// 不需要清除:回调的作用域就是项目实例本身,项目停止后实例被释放,回调随之消失; - /// 只覆盖"产生了轮次"的情形:入口被禁用、项目未启动或已停止时不会产生轮次,也就不会调用; - /// 生命周期请用 与 。 + /// 只在"产生了轮次"时调用:入口被禁用、项目未启动或已停止时不会产生轮次,也就不会调用; + /// 生命周期请用 与 ; + /// 订阅的作用域是项目实例:项目停止/释放后订阅随之失效,但持有句柄的观察者仍应尽快 Dispose, + /// 以免让已释放的项目因为委托引用而多活一段时间。 /// ///
/// 入口测点组的名称(即 entry.TagName()) - /// 回调;传 null 表示清除该入口已设置的回调 - /// 为 null + /// 回调;不允许为 null(要退订请 Dispose 返回的句柄) + /// 退订句柄( 即退订,幂等、不抛) + /// 或 为 null /// 未找到指定的入口测点组 - /// 项目已释放(回调的作用域就是项目实例) - void SetTurnProcessHandler(string entry, TurnProcess? handler); + /// 项目已释放(订阅的作用域就是项目实例) + IDisposable SubscribeTurnProcess(string entry, TurnProcess handler); /// @@ -130,9 +134,19 @@ public interface ITagsProject : IDisposable /// /// 运行本项目下所有的入口测点组。
/// 这个方法在所有入口组都运行结束之前,不会返回! + /// + /// 启动是两阶段的:先对全部入口算出逻辑组件匹配(纯计算),再准备 runner 并启动轮询。 + /// 匹配/准备阶段失败时不会有入口处于"已经跑起来"的状态;启动阶段失败则会取消并等待已启动的入口退出后 + /// 再把原异常抛出,不会留下"Task 无人等待、令牌永不取消"的孤儿轮询。 + /// ///
/// /// + /// 项目未初始化(通道/测点/逻辑组件为空,或没有任何入口测点组) + /// + /// 为入口匹配逻辑组件时 / 抛出了异常。 + /// 这是确定性的配置错误,不走 的重试路径(也不会启动任何入口)。 + /// Task RunAsync(CancellationToken ct); diff --git a/src/StdUnit.Tags.Core/Projects/TagsProject.cs b/src/StdUnit.Tags.Core/Projects/TagsProject.cs index 4dcf2b6..e64e6f5 100644 --- a/src/StdUnit.Tags.Core/Projects/TagsProject.cs +++ b/src/StdUnit.Tags.Core/Projects/TagsProject.cs @@ -204,6 +204,15 @@ public IList GetEntries() /// /// 项目未初始化(通道/测点/逻辑组件为空,或没有任何入口测点组) + /// + /// 启动前为入口匹配逻辑组件( / )时抛出了异常。 + /// 这是确定性的配置错误(重试不会变好),因此不走 的重试/重连恢复路径; + /// 抛出时没有任何入口开始轮询(启动分两阶段:先算完匹配、再准备并启动 runner)。 + /// + /// + /// 启动失败(含 抛出的异常)会先取消并等待已启动的入口退出, + /// 再把原异常抛出——不会留下无人等待的孤儿轮询。 + /// public virtual Task RunAsync(CancellationToken ct) { if (this.Channels == null || this.Channels.Count == 0) @@ -225,17 +234,81 @@ public virtual Task RunAsync(CancellationToken ct) throw new InvalidOperationException("未配置入口测点组"); } - var tasks = new ConcurrentBag(); - Parallel.ForEach(entries, entry => + // 阶段 1:纯计算(匹配 + 排序)。失败时"什么都还没发生"——没有 runner、没有轮询。 + var plans = new List<(ITagGrp Entry, List Logicets)>(entries.Count); + foreach (var entry in entries) { - var writeIntentChannel = this._entryWriteIntentChannels.GetOrAdd(entry.TagName(), _ => this.CreateIntentChannel()); - var logicets = this.Logicets - .Where(l => l.MatchEntry(entry)) - .OrderBy(l => l.Order) - .ToList(); + plans.Add((entry, this.SelectLogicets(entry))); + } + + // 阶段 2a:准备(建意图通道 / 建 runner / 订阅事件)。仍可能有同步抛出(IntentCapacity 非法、第三方工厂), + // 但都发生在任何轮询启动之前,所以失败时项目"一个轮次都没开始"。 + var prepared = this.PrepareEntries(plans); + + // 阶段 2b:启动 + 失败回滚(见 StartPreparedEntriesAsync)。 + return this.StartPreparedEntriesAsync(prepared, ct); + } + + /// + /// 阶段 1:为入口算出要执行哪些逻辑组件( 匹配 + 按 排序)。
+ /// 纯计算:不建通道、不建 runner、不改任何状态,所以可以对全部入口先算完再启动。
+ /// 逻辑组件是扩展点(用户代码):匹配/排序属于配置判定而不是轮询工作,抛异常意味着"这条配置无法判定", + /// 重试不会变好,因此报成配置错误,而不是走 的恢复路径。 + ///
+ private List SelectLogicets(ITagGrp entry) + { + var matched = new List<(ILogicet Logicet, int Order)>(); + foreach (var logicet in this.Logicets) + { + if (this.InvokeLogicet(entry, logicet, nameof(ILogicet.MatchEntry), () => logicet.MatchEntry(entry))) + { + matched.Add((logicet, this.InvokeLogicet(entry, logicet, nameof(ILogicet.Order), () => logicet.Order))); + } + } + + // 排序只操作已经取到的 Order,不再调用用户代码。 + return matched + .OrderBy(x => x.Order) + .Select(x => x.Logicet) + .ToList(); + } + + /// + /// 调用逻辑组件的启动期成员,把抛出的异常转成带定位上下文的 。 + /// + private T InvokeLogicet(ITagGrp entry, ILogicet logicet, string operation, Func func) + { + try + { + return func(); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + var location = entry.GetLocationPath(); + throw new TagsProjectConfigurationException( + $"逻辑组件 {logicet.GetType().FullName} 在入口 {location} 上执行 {operation} 发生异常({ex.GetType().Name}: {ex.Message})。" + + $"{operation} 属于启动期的配置判定:必须无副作用、不阻塞、不得抛异常", + location, + ex); + } + } + + /// + /// 阶段 2a:准备全部入口(建意图通道、建 runner、订阅事件),但不启动轮询。
+ /// 这一步仍可能同步抛出( 非法、第三方 runner 工厂失败), + /// 但它发生在任何 之前,所以失败时一个轮次都没开始。 + ///
+ private List<(ITagGrp Entry, ITagGrpRunner Runner)> PrepareEntries(List<(ITagGrp Entry, List Logicets)> plans) + { + var prepared = new List<(ITagGrp Entry, ITagGrpRunner Runner)>(plans.Count); + foreach (var (entry, logicets) in plans) + { + _ = this._entryWriteIntentChannels.GetOrAdd(entry.TagName(), _ => this.CreateIntentChannel()); + var runner = this._tagGrpRunnerFactory.Create(this); - runner.RunnerStarted += RunnerStarted; - runner.TurnProcess += async (entry, ch, turnCt) => + // runner 是单槽回调:把"多播 + 调用时取快照"收口在本类(项目层才是多个独立观察者的地方)。 + runner.RunnerStarted = (grp, ch, startedCt) => this.InvokeRunnerStartedAsync(grp, ch, startedCt); + runner.TurnProcess = async (grp, ch, turnCt) => { foreach (var l in logicets) { @@ -243,23 +316,60 @@ public virtual Task RunAsync(CancellationToken ct) { continue; } - await l.ProcessAsync(entry, ch, turnCt); + await l.ProcessAsync(grp, ch, turnCt); } - // 公共回调排在 Logicets 之后:订阅者看到的是逻辑处理完成后的值。 - // 有意不 catch:回调抛异常等价于本轮失败(与 Logicet 抛异常同义),见 ITagsProject.SetTurnProcessHandler 的文档。 - // 回调表是写时复制的:这里读到的那份快照本轮内不会被改(运行期设置/清除从下一轮生效),所以不需要同步。 - if (this._entryTurnHandlers.TryGetValue(entry.TagName(), out var handler)) + // 公共订阅排在 Logicets 之后:订阅者看到的是逻辑处理完成后的值。 + // 有意不 catch:回调抛异常等价于本轮失败(与 Logicet 抛异常同义),并且按注册顺序 + // "首个异常即中断后续订阅者"(与 RunnerStarted / RunnerCrashed 的语义一致)。 + // 订阅表是写时复制的:这里读到的那份快照本轮内不会被改(运行期变更从下一轮生效)。 + if (this._entryTurnHandlers.TryGetValue(grp.TagName(), out var handlers)) { - await handler(entry, ch, turnCt); + foreach (var handler in handlers) + { + await handler(grp, ch, turnCt); + } } }; - runner.RunnerCrashed += RunnerCrashed; - Task task = runner.StartAsync(entry, ct); - tasks.Add(task); - }); + runner.RunnerCrashed = (grp, ch, ex, crashedCt) => this.InvokeRunnerCrashedAsync(grp, ch, ex, crashedCt); + prepared.Add((entry, runner)); + } - return Task.WhenAll(tasks); + return prepared; + } + + /// + /// 阶段 2b:启动全部入口,并在任一入口启动失败时回滚——取消并等待此前已启动的入口退出。
+ /// 回滚靠内部的 linked 令牌实现:同一次运行里各回调拿到的仍然是同一个令牌实例, + /// 但不保证 == 传给 的那个(否则无法回滚已经启动的入口)。 + ///
+ private async Task StartPreparedEntriesAsync(List<(ITagGrp Entry, ITagGrpRunner Runner)> prepared, CancellationToken ct) + { + using var abort = CancellationTokenSource.CreateLinkedTokenSource(ct); + var tasks = new List(prepared.Count); + try + { + foreach (var (entry, runner) in prepared) + { + tasks.Add(runner.StartAsync(entry, abort.Token)); + } + + await Task.WhenAll(tasks).ConfigureAwait(false); + } + catch + { + // 回滚:同步抛出的第三方实现、外部停机、运行器任务故障,都在这里收口。 + // 清理路径的既有约定:回滚期间的异常不覆盖原始异常。 + abort.Cancel(); + try + { + await Task.WhenAll(tasks).ConfigureAwait(false); + } + catch + { + } + throw; + } } #region Intent Mgmt @@ -342,58 +452,210 @@ protected virtual void CompleteIntentChannels(string disposeMsg) } #endregion + #region Runner events(多播) + + /// + /// 订阅者列表(低频事件:每次建连/崩溃才触发),写侧用 + /// 保护"读-改-写",读侧用 + /// 读取一次快照。
+ /// 这里刻意不做轮询热路径式的无锁设计(那是 的诉求): + /// 转发给 runner 的委托每次调用时才读快照,所以运行期订阅/退订从下一次通知开始生效。 + ///
+ private RunnerStarted? _runnerStartedHandlers; + + /// 见 。 + private RunnerCrashed? _runnerCrashedHandlers; + + /// 保护上面两个委托字段的合并/移除。 + private readonly object _runnerEventsGate = new(); + /// - public event RunnerCrashed? RunnerCrashed; + public event RunnerStarted? RunnerStarted + { + add + { + // 不需要判 null:Delegate.Combine(list, null) 就是 list 本身(空操作)。 + lock (this._runnerEventsGate) + { + this._runnerStartedHandlers += value; + } + } + remove + { + // 不需要判 null:Delegate.Remove(list, null) 原样返回 list(空操作)。 + lock (this._runnerEventsGate) + { + this._runnerStartedHandlers -= value; + } + } + } + /// - public event RunnerStarted? RunnerStarted; + public event RunnerCrashed? RunnerCrashed + { + add + { + lock (this._runnerEventsGate) + { + this._runnerCrashedHandlers += value; + } + } + remove + { + lock (this._runnerEventsGate) + { + this._runnerCrashedHandlers -= value; + } + } + } + + /// + /// 按注册顺序串行调用订阅者;遇到第一个异常立即中断并原样抛出(保留类型与堆栈)。
+ /// 刻意不做"收集全部异常再抛":收集了却只抛一个等于把其余静默丢弃,而 Core 没有 logger 可以留痕; + /// 也不包装成 ——"谁抛谁负责、后续不再执行"更符合直觉,异常类型也可预测。 + ///
+ private async Task InvokeRunnerStartedAsync(ITagGrp grp, ITagChannel? channel, CancellationToken ct) + { + var handlers = Volatile.Read(ref this._runnerStartedHandlers); + if (handlers is null) + { + return; + } + + foreach (var d in handlers.GetInvocationList()) + { + await ((RunnerStarted)d)(grp, channel, ct); + } + } + + /// 见 。 + private async Task InvokeRunnerCrashedAsync(ITagGrp grp, ITagChannel? channel, Exception ex, CancellationToken ct) + { + var handlers = Volatile.Read(ref this._runnerCrashedHandlers); + if (handlers is null) return; + + foreach (var d in handlers.GetInvocationList()) + { + await ((RunnerCrashed)d)(grp, channel, ex, ct); + } + } + #endregion - #region TurnProcess handlers + #region TurnProcess subscriptions /// - /// 入口名 → 该入口的回调(每个入口最多一个)。写时复制:设置/清除时在 - /// 下复制一份新表再整体替换,已发布的表实例永不再改, - /// 所以轮询线程可以直接读、不需要任何同步,运行期设置/清除自然就是"从下一轮开始生效"。 + /// 入口名 → 该入口的订阅者数组(每个入口 0..N 个)。写时复制:订阅/退订时在 + /// 下复制一份新表(和新的数组)再整体替换,已发布的实例永不再改, + /// 所以轮询线程可以直接读、不需要任何同步,运行期订阅/退订自然就是"从下一轮开始生效"。 /// - private volatile Dictionary _entryTurnHandlers = new(StringComparer.Ordinal); + private volatile Dictionary _entryTurnHandlers = new(StringComparer.Ordinal); - /// 保护 的复制作业(只在设置/清除时短暂持有)。 + /// 保护 的复制作业(只在订阅/退订时短暂持有)。 private readonly object _entryTurnHandlersGate = new(); /// - public void SetTurnProcessHandler(string entry, TurnProcess? handler) + public IDisposable SubscribeTurnProcess(string entry, TurnProcess handler) { if (entry is null) { throw new ArgumentNullException(nameof(entry)); } + if (handler is null) + { + throw new ArgumentNullException( + nameof(handler), + "轮次回调不能为 null;要退订请 Dispose 订阅时返回的句柄。"); + } if (this._disposed) { throw new ObjectDisposedException( this.GetType().Name, - $"项目已释放,不能再设置 TurnProcess 回调(入口={entry}):回调的作用域就是项目实例。"); + $"项目已释放,不能再订阅 TurnProcess 回调(入口={entry}):订阅的作用域就是项目实例。"); } - // 与 WriteIntent 的入口校验保持一致:只允许向合法的入口设置 + // 与 WriteIntent 的入口校验保持一致:只允许向合法的入口订阅 if (!this.GetEntries().Any(e => e.TagName() == entry)) { throw new KeyNotFoundException($"未找到指定的入口测点组: {entry}"); } - // 替换语义:需要组合多个观察者时,由调用方自己组装成一个 TurnProcess。 - // 运行期也允许设置/清除(写时复制,从下一轮开始生效)。 + var subscription = new TurnProcessSubscription(this, entry, handler); lock (this._entryTurnHandlersGate) { - var updated = new Dictionary(this._entryTurnHandlers, StringComparer.Ordinal); - if (handler is null) + var current = this._entryTurnHandlers; + var updated = new Dictionary(current, StringComparer.Ordinal); + if (current.TryGetValue(entry, out var handlers) && handlers.Length > 0) { - updated.Remove(entry); + var appended = new TurnProcess[handlers.Length + 1]; + handlers.CopyTo(appended, 0); + appended[appended.Length - 1] = handler; + updated[entry] = appended; } else { - updated[entry] = handler; + updated[entry] = new[] { handler }; } this._entryTurnHandlers = updated; } + + return subscription; + } + + /// + /// 退订(由 调用)。只摘掉那一个订阅实例: + /// 同一个委托实例订阅多次时,每次 Dispose 摘掉其中一次。 + /// + private void UnsubscribeTurnProcess(TurnProcessSubscription subscription) + { + lock (this._entryTurnHandlersGate) + { + // 不变量:订阅在 SubscribeTurnProcess 里写入该入口的数组,而 Dispose 由 + // TurnProcessSubscription 用 Interlocked 保证只生效一次 ⇒ 条目必然存在、且必然能定位到这一个委托实例。 + // 退订只缩数组(退到最后一个时留下空数组),不删键——空数组是合法状态,之后还能再订阅。 + var current = this._entryTurnHandlers; + var handlers = current[subscription.Entry]; + var index = Array.FindIndex(handlers, h => ReferenceEquals(h, subscription.Handler)); + + var updated = new Dictionary(current, StringComparer.Ordinal); + if (handlers.Length == 1) + { + updated[subscription.Entry] = Array.Empty(); + } + else + { + var removed = new TurnProcess[handlers.Length - 1]; + Array.Copy(handlers, 0, removed, 0, index); + Array.Copy(handlers, index + 1, removed, index, handlers.Length - index - 1); + updated[subscription.Entry] = removed; + } + + this._entryTurnHandlers = updated; + } + } + + /// + /// 订阅句柄。幂等:Dispose 只生效一次;项目已释放之后再 Dispose 是无操作(不抛)。 + /// + private sealed class TurnProcessSubscription : IDisposable + { + private TagsProject? _owner; + + internal TurnProcessSubscription(TagsProject owner, string entry, TurnProcess handler) + { + this._owner = owner; + this.Entry = entry; + this.Handler = handler; + } + + internal string Entry { get; } + + internal TurnProcess Handler { get; } + + public void Dispose() + { + var owner = Interlocked.Exchange(ref this._owner, null); + owner?.UnsubscribeTurnProcess(this); + } } #endregion diff --git a/src/StdUnit.Tags.Core/TagGrpRunner/ITagGrpRunner.cs b/src/StdUnit.Tags.Core/TagGrpRunner/ITagGrpRunner.cs index 5a244c1..93662d2 100644 --- a/src/StdUnit.Tags.Core/TagGrpRunner/ITagGrpRunner.cs +++ b/src/StdUnit.Tags.Core/TagGrpRunner/ITagGrpRunner.cs @@ -67,8 +67,8 @@ namespace StdUnit.Tags; /// 多通道:一个入口下可以挂多个通道。每轮在读写之前,运行器会先为入口子树中 /// 所有会被用到的通道(见 )逐一执行 /// ,这样不会出现“读了一半才发现辅通道未连接、整轮作废”的 -/// 情况。入口自身解析出的通道称为主通道,它是各事件委托中 channel 参数的语义; -/// 辅通道不进事件,需要时请从入口上自行解析。 +/// 情况。入口自身解析出的通道称为主通道,它是各回调中 channel 参数的语义; +/// 辅通道不进回调,需要时请从入口上自行解析。 /// /// /// 通道独占约束:崩溃/取消后的清理路径会断开入口相关的全部通道, @@ -81,27 +81,41 @@ public interface ITagGrpRunner { /// - /// 启动事件 + /// 启动回调(单槽,可为 null)。
+ /// 运行器实例不对外暴露,唯一的使用方是 ——需要多播时由上层组合出一个委托再赋值, + /// 运行器不负责合并多个订阅者( 之类的委托返回 , + /// 用 event 多播会退化成"只 await 最后一个处理器")。
+ /// 必须在 之前赋值。 ///
- event RunnerStarted? RunnerStarted; + RunnerStarted? RunnerStarted { get; set; } /// - /// 崩溃事件。
- /// 取消不是崩溃:项目停止导致的取消不会触发本事件。
- /// 注意这里不要再有异常发生,否则会视作对应的入口轮询需要停机 + /// 崩溃回调(单槽,可为 null)。
+ /// 取消不是崩溃:项目停止导致的取消不会触发本回调。
+ /// 注意这里不要再有异常发生,否则会视作对应的入口轮询需要停机。 ///
- event RunnerCrashed? RunnerCrashed; + RunnerCrashed? RunnerCrashed { get; set; } /// - /// 每一轮处理 + /// 每一轮处理回调(单槽,可为 null),在"读取输入 + 逻辑处理"之后、"刷写输出"之前调用。 /// - event TurnProcess? TurnProcess; + TurnProcess? TurnProcess { get; set; } /// /// 启动对群组的监控: loop(意图执行-> 读取输入 -> 逻辑处理 -> 刷写输出)
/// 一旦成功启动轮询,除非主动取消,否则不会抛出异常。
/// 取消不是崩溃: 取消时本方法以 结束 /// (入口子树里的通道照常断开),但不会触发 。 + /// + /// 失败形态:启动失败推荐交给返回的 ——用 async 方法实现即可(异常会自动装进 Task)。 + /// 若实现选择在返回前直接抛出(如形参校验), 也会取消并等待此前已启动的入口退出后 + /// 再抛出,因此同样不会留下无人等待的轮询。两种形态都可以,只是前者能让故障更早地以 的形式被观察到。 + /// + /// + /// 令牌:同一次运行里,本方法的 与 / + /// / 拿到的是同一个实例;但它不必等于传给 + /// 的令牌(内部用 linked 令牌以便启动失败时回滚)。 + /// ///
/// 入口;其子树中的通道由本运行器建立与断开连接 /// 项目生命周期取消令牌;取消即结束本方法 diff --git a/src/StdUnit.Tags.WebUi/Core/TagsStreamHub.cs b/src/StdUnit.Tags.WebUi/Core/TagsStreamHub.cs index 08a2f38..3e1a757 100644 --- a/src/StdUnit.Tags.WebUi/Core/TagsStreamHub.cs +++ b/src/StdUnit.Tags.WebUi/Core/TagsStreamHub.cs @@ -7,7 +7,7 @@ namespace StdUnit.Tags.WebUi; /// /// 只读观测流的核心实现(对外只暴露 ):
/// - 跟踪 的启停,维护每次启动的 epoch 与结构索引;
-/// - 为每个入口用 挂一个"轮次完成"回调,做值 diff;
+/// - 为每个入口用 订阅一个"轮次完成"回调,做值 diff;
/// - 把结果序列化成 SSE 帧,扇出给各订阅者(每帧只序列化一次)。
/// 同时是订阅者集合的持有者:订阅相关的成员只出现在内部的 /// 上,只服务同一程序集里的端点与 。 @@ -50,6 +50,13 @@ internal sealed class TagsStreamHub : ITagsStreamHub, IDisposable private readonly HashSet _pipers = new(); private readonly Channel _pending; + /// + /// 当前项目上的轮次订阅句柄(只在 下访问)。
+ /// 项目停止/切换、或 hub 自己停止时必须退订:否则已释放的项目会因为委托引用而多活, + /// 回调也会继续打到已经停掉的 hub 上。 + ///
+ private readonly List _turnSubscriptions = new(); + private volatile ProjectIndex? _index; private volatile ITagsProject? _project; private volatile StateDto _state = new(0, false, null); @@ -167,6 +174,12 @@ public void Stop() Interlocked.Exchange(ref this._pipersCount, 0); } + // 停止观测后不再接收轮次通知:退订并把句柄还给项目(避免已释放的项目被委托引用拖住)。 + lock (this._transitionGate) + { + this.UnsubscribeTurns(); + } + var cts = Interlocked.Exchange(ref this._cts, null); if (cts is null) { @@ -333,6 +346,9 @@ private void OnProjectStarted(ITagsProject project) return; } + // 换项目(或上一次"已停止"没派发到):先把旧项目上的订阅摘干净,再挂新的。 + this.UnsubscribeTurns(); + var epoch = Interlocked.Increment(ref this._epoch); ProjectIndex index; try @@ -357,33 +373,52 @@ private void OnProjectStarted(ITagsProject project) } } - // 运行期也能挂(SetTurnProcessHandler 从下一轮生效):hub 晚于轮询启动时,这里就是在补挂回调。 + // 运行期也能订阅(从下一轮生效):hub 晚于轮询启动时,这里就是在补挂回调。 foreach (var entry in index.Entries) { + if (Volatile.Read(ref this._stopped) != 0) + { + // 已经停机:不再补挂,否则会留下指向已停止 hub 的回调。 + break; + } + var entryKey = entry.Key; try { - project.SetTurnProcessHandler(entryKey, (_, _, _) => + this._turnSubscriptions.Add(project.SubscribeTurnProcess(entryKey, (_, _, _) => { this.OnTurn(epoch, entryKey); return Task.CompletedTask; - }); + })); } catch (Exception ex) { this._logger?.LogWarning( ex, - "无法为入口 {Entry} 注册轮次回调(项目可能已释放):该入口只能观测结构,观测不到值", + "无法为入口 {Entry} 订阅轮次回调(项目可能已释放):该入口只能观测结构,观测不到值", entryKey); } } } } + /// 退订当前项目的全部轮次订阅(调用方必须持有 )。 + private void UnsubscribeTurns() + { + foreach (var subscription in this._turnSubscriptions) + { + subscription.Dispose(); + } + + this._turnSubscriptions.Clear(); + } + private void OnProjectStopped() { lock (this._transitionGate) { + this.UnsubscribeTurns(); + lock (this._gate) { if (this._project is null) diff --git a/src/StdUnit.Tags/TagGrpRunners/TagGrpRunner.cs b/src/StdUnit.Tags/TagGrpRunners/TagGrpRunner.cs index 2898da0..03e06a3 100644 --- a/src/StdUnit.Tags/TagGrpRunners/TagGrpRunner.cs +++ b/src/StdUnit.Tags/TagGrpRunners/TagGrpRunner.cs @@ -6,7 +6,7 @@ namespace StdUnit.Tags; /// /// 的默认实现。
/// 多通道:入口子树中所有会被用到的通道(见 ) -/// 会在每轮读写之前逐一建连;入口自身解析出的通道额外称为主通道,即各事件委托中的 channel 参数。
+/// 会在每轮读写之前逐一建连;入口自身解析出的通道额外称为主通道,即各回调中的 channel 参数。
/// 通道独占约束:清理路径会断开入口相关的全部通道,因此同一通道实例不得被多个入口 /// (含嵌套入口)共用,否则一方崩溃/取消会掐掉另一方正在使用的连接。 ///
@@ -37,20 +37,20 @@ public TagGrpRunner( } /// - public event RunnerStarted? RunnerStarted; + public RunnerStarted? RunnerStarted { get; set; } /// - public event TurnProcess? TurnProcess; + public TurnProcess? TurnProcess { get; set; } /// - public event RunnerCrashed? RunnerCrashed; + public RunnerCrashed? RunnerCrashed { get; set; } /// public virtual async Task StartAsync(ITagGrp entry, CancellationToken ct) { while (!ct.IsCancellationRequested) { - // 入口主通道。各事件委托中的 channel 参数恒为它。 + // 入口主通道。各回调中的 channel 参数恒为它。 ITagChannel? channel = null; // 本次启动需要确保连接的全部通道(含主通道,主通道排在最前)。 // 用空集合初始化,保证即使“解析通道”阶段抛异常,清理路径也不会空引用。 @@ -68,9 +68,11 @@ public virtual async Task StartAsync(ITagGrp entry, CancellationToken ct) channel = entry.SearchChannel(); channels = entry.CollectChannels(); - if (RunnerStarted is not null) + // 单槽回调:先取本地再判空,避免两次读取之间被改成 null。 + var onStarted = this.RunnerStarted; + if (onStarted is not null) { - await RunnerStarted(entry, channel, ct); + await onStarted(entry, channel, ct); } // 开始轮询 @@ -94,9 +96,10 @@ public virtual async Task StartAsync(ITagGrp entry, CancellationToken ct) await entry.WriteAsync(TraversalMode.RespectEnabled, ct); } await entry.ReadAsync(TraversalMode.RespectEnabled, ct); - if (TurnProcess is not null) + var onTurn = this.TurnProcess; + if (onTurn is not null) { - await TurnProcess(entry, channel, ct); + await onTurn(entry, channel, ct); } await entry.WriteAsync(TraversalMode.RespectEnabled, ct); @@ -126,11 +129,12 @@ public virtual async Task StartAsync(ITagGrp entry, CancellationToken ct) aborted = true; _consecutiveFailures++; - if (RunnerCrashed is not null) + var onCrashed = this.RunnerCrashed; + if (onCrashed is not null) { try { - await RunnerCrashed(entry, channel, ex, ct); + await onCrashed(entry, channel, ex, ct); } catch (Exception handlingError) { diff --git a/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetCancellationTests.cs b/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetCancellationTests.cs index a3a3cf4..b326ed7 100644 --- a/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetCancellationTests.cs +++ b/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetCancellationTests.cs @@ -90,6 +90,9 @@ public async Task RunAsync_PassesLifecycleTokenToLogicet() // Assert Assert.True(logicet.CapturedToken.CanBeCanceled, "ProcessAsync 应收到项目的生命周期令牌"); - Assert.Equal(cts.Token, logicet.CapturedToken); + // 不必与 RunAsync 的入参是同一个实例:框架用内部 linked 令牌启动各入口,以便启动失败时回滚 + // 已启动的入口。契约是"取消时机",不是引用相等。 + Assert.True(cts.IsCancellationRequested); + Assert.True(logicet.CapturedToken.IsCancellationRequested, "项目停止后回调拿到的令牌也应已取消"); } } diff --git a/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetLoaderAotTests.cs b/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetLoaderAotTests.cs index 789f540..d4f1c77 100644 --- a/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetLoaderAotTests.cs +++ b/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetLoaderAotTests.cs @@ -42,7 +42,7 @@ private sealed class DummyProject : ITagsProject public event RunnerStarted? RunnerStarted; public event RunnerCrashed? RunnerCrashed; - public void SetTurnProcessHandler(string entry, TurnProcess? handler) => throw new NotImplementedException(); + public IDisposable SubscribeTurnProcess(string entry, TurnProcess handler) => throw new NotImplementedException(); public void Dispose() { } diff --git a/tests/StdUnit.Tags.Tests/Core/Logicets/ProjectAddLogicetTests.cs b/tests/StdUnit.Tags.Tests/Core/Logicets/ProjectAddLogicetTests.cs index c56814e..8051a07 100644 --- a/tests/StdUnit.Tags.Tests/Core/Logicets/ProjectAddLogicetTests.cs +++ b/tests/StdUnit.Tags.Tests/Core/Logicets/ProjectAddLogicetTests.cs @@ -215,7 +215,7 @@ public void Initialize(string projRoot, XElement? root = null) { } public event RunnerStarted? RunnerStarted; public event RunnerCrashed? RunnerCrashed; - public void SetTurnProcessHandler(string entry, TurnProcess? handler) => throw new NotImplementedException(); + public IDisposable SubscribeTurnProcess(string entry, TurnProcess handler) => throw new NotImplementedException(); } /// diff --git a/tests/StdUnit.Tags.Tests/Core/Projects/RunAsyncStartupTests.cs b/tests/StdUnit.Tags.Tests/Core/Projects/RunAsyncStartupTests.cs new file mode 100644 index 0000000..6de0c62 --- /dev/null +++ b/tests/StdUnit.Tags.Tests/Core/Projects/RunAsyncStartupTests.cs @@ -0,0 +1,342 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using System.Xml.Linq; +using Microsoft.Extensions.DependencyInjection; +using StdUnit.Tags.Tests.Fakes; +using Xunit; + +namespace StdUnit.Tags.Tests.Core.Projects; + +/// +/// 验证 的启动是事务性的:
+/// - 为入口匹配逻辑组件( / )抛出的异常 +/// 报成 (带入口路径与逻辑组件类型),而不是变成 +/// AggregateException 或走 的重试路径;
+/// - 匹配失败时没有任何入口开始轮询;
+/// - 启动前准备工作(runner 工厂、意图通道)失败时同样一个 runner 都没启动,不留无人等待的孤儿 runner。 +///
+public class RunAsyncStartupTests +{ + private readonly string _projectRoot; + + public RunAsyncStartupTests() + { + _projectRoot = TestPaths.TempPath("RunAsyncStartupTests"); + System.IO.Directory.CreateDirectory(_projectRoot); + } + + #region fakes + + private abstract class ProbeLogicetBase : ILogicet + { + protected ProbeLogicetBase(ITagsProject project) => Project = project; + + public ITagsProject Project { get; } + + public virtual int Order => 0; + + public virtual bool Enabled => true; + + public abstract bool MatchEntry(ITagGrp entry); + + public Task ProcessAsync(ITagGrp entry, ITagChannel? thisChannel, CancellationToken ct) => Task.CompletedTask; + } + + /// 对指定入口抛异常的 。 + private sealed class MatchThrowingLogicet : ProbeLogicetBase + { + private readonly string _throwingEntry; + + public MatchThrowingLogicet(ITagsProject project, string throwingEntry) : base(project) + => _throwingEntry = throwingEntry; + + public override bool MatchEntry(ITagGrp entry) + => entry.TagName() == _throwingEntry + ? throw new InvalidOperationException("模拟的 MatchEntry 失败") + : true; + } + + /// 读取 就抛异常的逻辑组件。 + private sealed class OrderThrowingLogicet : ProbeLogicetBase + { + public OrderThrowingLogicet(ITagsProject project) : base(project) + { + } + + public override int Order => throw new InvalidOperationException("模拟的 Order 失败"); + + public override bool MatchEntry(ITagGrp entry) => true; + } + + /// 统计 Create / StartAsync 次数(可指定第几次 Create 抛异常)的 runner 工厂。 + private sealed class CountingRunnerFactory : ITagGrpRunnerFactory + { + private int _createCount; + private int _startCount; + + public int CreateCount => Volatile.Read(ref _createCount); + + public int StartCount => Volatile.Read(ref _startCount); + + /// 第几次 抛异常(0 表示从不抛)。 + public int ThrowOnCreate { get; set; } + + public ITagGrpRunner Create(ITagsProject project) + { + var n = Interlocked.Increment(ref _createCount); + if (n == ThrowOnCreate) + { + throw new InvalidOperationException("模拟的 runner 工厂失败"); + } + return new CountingRunner(this); + } + + private sealed class CountingRunner : ITagGrpRunner + { + private readonly CountingRunnerFactory _owner; + + public CountingRunner(CountingRunnerFactory owner) => _owner = owner; + + public RunnerStarted? RunnerStarted { get; set; } + + public RunnerCrashed? RunnerCrashed { get; set; } + + public TurnProcess? TurnProcess { get; set; } + + public Task StartAsync(ITagGrp entry, CancellationToken ct) + { + Interlocked.Increment(ref _owner._startCount); + return Task.CompletedTask; + } + } + } + + /// 第 N 个 runner 的 会同步抛出;其余会一直挂到令牌取消。 + private sealed class SyncThrowingRunnerFactory : ITagGrpRunnerFactory + { + private int _createCount; + + public List Created { get; } = new(); + + /// 第几个 runner 同步抛出(默认永不)。 + public int ThrowOnCreate { get; set; } = int.MaxValue; + + public ITagGrpRunner Create(ITagsProject project) + { + var n = Interlocked.Increment(ref _createCount); + var runner = new HangingRunner(throwOnStart: n == ThrowOnCreate); + Created.Add(runner); + return runner; + } + + internal sealed class HangingRunner : ITagGrpRunner + { + private readonly bool _throwOnStart; + + public HangingRunner(bool throwOnStart) => _throwOnStart = throwOnStart; + + /// 启动返回的 Task(同步抛出时为 null)。 + public Task? StartTask { get; private set; } + + public RunnerStarted? RunnerStarted { get; set; } + + public RunnerCrashed? RunnerCrashed { get; set; } + + public TurnProcess? TurnProcess { get; set; } + + public Task StartAsync(ITagGrp entry, CancellationToken ct) + { + if (_throwOnStart) + { + throw new InvalidOperationException("模拟的同步抛出"); + } + + StartTask = Task.Delay(Timeout.Infinite, ct); + return StartTask; + } + } + } + + #endregion + + private static XElement Channel(string name) => + new("Channel", new XAttribute("name", name), new XAttribute("driver", "fake")); + + private static XElement Entry(string name, string channelName) => + new("TagGrp", + new XAttribute("name", name), + new XAttribute("isEntry", "true"), + new XAttribute("isEnabled", "true"), + new XAttribute("channel", channelName), + new XAttribute("scanInterval", 0), + new XElement("Tag", + new XAttribute("name", "v"), + new XAttribute("type", "INT16"), + new XAttribute("access", "RW"), + new XAttribute("address", "0"))); + + /// 每个入口用独立通道(入口通道独占校验要求),并覆盖默认的 runner 工厂。 + private ServiceProvider CreateServices(ITagGrpRunnerFactory factory) + { + var services = new ServiceCollection(); + services.AddLogging(); + services.AddTagsProjectServices(b => + { + b.AddFakedSupport(); + b.AddFakedTagSupport(); + }); + + // 后注册者胜出:覆盖默认的 TagGrpRunnerFactory + services.AddSingleton(factory); + return services.BuildServiceProvider(); + } + + private ITagsProject CreateProject(IServiceProvider sp, XElement root) => + sp.GetRequiredService().Create(_projectRoot, root); + + [Fact] + public void RunAsync_MatchEntryThrows_ReportsConfigurationError_AndStartsNoRunner() + { + var factory = new CountingRunnerFactory(); + using var root = CreateServices(factory); + using var scope = root.CreateScope(); + using var proj = CreateProject(scope.ServiceProvider, + new XElement("Project", + Channel("fake1"), Entry("g1", "fake1"), + Channel("fake2"), Entry("g2", "fake2"))); + proj.Logicets.Add(new MatchThrowingLogicet(proj, throwingEntry: "g2")); + + var ex = Assert.Throws( + () => { _ = proj.RunAsync(CancellationToken.None); }); + + Assert.Equal("TagGrp(g2)", ex.Location); + Assert.IsType(ex.InnerException); + Assert.Contains(nameof(MatchThrowingLogicet), ex.Message, StringComparison.Ordinal); + Assert.Contains("MatchEntry", ex.Message, StringComparison.Ordinal); + + // 阶段 1(匹配)就失败了:连 runner 都没建,更没有入口开始轮询。 + Assert.Equal(0, factory.CreateCount); + Assert.Equal(0, factory.StartCount); + } + + [Fact] + public void RunAsync_OrderThrows_ReportsConfigurationError_AndStartsNoRunner() + { + var factory = new CountingRunnerFactory(); + using var root = CreateServices(factory); + using var scope = root.CreateScope(); + using var proj = CreateProject(scope.ServiceProvider, + new XElement("Project", Channel("fake1"), Entry("g1", "fake1"))); + proj.Logicets.Add(new OrderThrowingLogicet(proj)); + + var ex = Assert.Throws( + () => { _ = proj.RunAsync(CancellationToken.None); }); + + Assert.Equal("TagGrp(g1)", ex.Location); + Assert.Contains(nameof(OrderThrowingLogicet), ex.Message, StringComparison.Ordinal); + Assert.Contains("Order", ex.Message, StringComparison.Ordinal); + + Assert.Equal(0, factory.CreateCount); + Assert.Equal(0, factory.StartCount); + } + + [Fact] + public void RunAsync_RunnerFactoryThrowsOnSecondEntry_StartsNoRunner() + { + var factory = new CountingRunnerFactory { ThrowOnCreate = 2 }; + using var root = CreateServices(factory); + using var scope = root.CreateScope(); + using var proj = CreateProject(scope.ServiceProvider, + new XElement("Project", + Channel("fake1"), Entry("g1", "fake1"), + Channel("fake2"), Entry("g2", "fake2"))); + + var ex = Assert.Throws( + () => { _ = proj.RunAsync(CancellationToken.None); }); + + Assert.Contains("模拟的 runner 工厂失败", ex.Message, StringComparison.Ordinal); + + // 关键:全部准备动作都在任何 StartAsync 之前,所以第二个入口准备失败时第一个入口也还没启动。 + // (若恢复成"每入口 准备+启动"的旧结构,这里会是 1,留下一个没人等待的孤儿 runner。) + Assert.Equal(2, factory.CreateCount); + Assert.Equal(0, factory.StartCount); + } + + [Fact] + public void RunAsync_InvalidIntentCapacity_StartsNoRunner() + { + var factory = new CountingRunnerFactory(); + using var root = CreateServices(factory); + using var scope = root.CreateScope(); + using var proj = CreateProject(scope.ServiceProvider, + new XElement("Project", Channel("fake1"), Entry("g1", "fake1"))); + proj.IntentCapacity = 0; + + Assert.Throws( + () => { _ = proj.RunAsync(CancellationToken.None); }); + + Assert.Equal(0, factory.CreateCount); + Assert.Equal(0, factory.StartCount); + } + + [Fact] + public async Task RunAsync_AllRunnersCompleteNormally_ReturnsNormally() + { + // runner 正常结束(不是被取消、也不是故障)时,RunAsync 必须正常返回。 + var factory = new CountingRunnerFactory(); + using var root = CreateServices(factory); + using var scope = root.CreateScope(); + using var proj = CreateProject(scope.ServiceProvider, + new XElement("Project", Channel("fake1"), Entry("g1", "fake1"))); + + await proj.RunAsync(CancellationToken.None); + + Assert.Equal(1, factory.CreateCount); + Assert.Equal(1, factory.StartCount); + } + + [Fact] + public async Task RunAsync_FirstStartAsyncThrowsSynchronously_RollsBackEmptyList() + { + // 第一个入口就同步抛出:任务列表为空,回滚只能取消并等待"空集合"。 + var factory = new SyncThrowingRunnerFactory { ThrowOnCreate = 1 }; + using var root = CreateServices(factory); + using var scope = root.CreateScope(); + using var proj = CreateProject(scope.ServiceProvider, + new XElement("Project", Channel("fake1"), Entry("g1", "fake1"))); + + var ex = await Assert.ThrowsAsync( + () => proj.RunAsync(CancellationToken.None)); + + Assert.Contains("模拟的同步抛出", ex.Message, StringComparison.Ordinal); + Assert.Single(factory.Created); + Assert.Null(factory.Created[0].StartTask); + } + + [Fact] + public async Task RunAsync_StartAsyncThrowsSynchronously_RollsBackAlreadyStartedRunners() + { + // 第三方 runner 可以在返回 Task 之前直接抛出(Task 返回值不保证"异常都在 Task 上")。 + // 此时框架必须取消并等待此前已启动的入口退出,再把原异常抛出——不能留下无人持有的轮询。 + var factory = new SyncThrowingRunnerFactory { ThrowOnCreate = 2 }; + using var root = CreateServices(factory); + using var scope = root.CreateScope(); + using var proj = CreateProject(scope.ServiceProvider, + new XElement("Project", + Channel("fake1"), Entry("g1", "fake1"), + Channel("fake2"), Entry("g2", "fake2"))); + + var ex = await Assert.ThrowsAsync( + () => proj.RunAsync(CancellationToken.None)); + + Assert.Contains("模拟的同步抛出", ex.Message, StringComparison.Ordinal); + Assert.Equal(2, factory.Created.Count); + + var first = factory.Created[0].StartTask; + Assert.NotNull(first); + Assert.True(first!.IsCompleted, "回滚必须等待已启动的入口退出,不能留下孤儿轮询"); + Assert.True(first.IsCanceled, "回滚的方式是取消(生命周期令牌被取消)"); + Assert.Null(factory.Created[1].StartTask); // 第二个入口同步抛出,从未产生 Task + } +} diff --git a/tests/StdUnit.Tags.Tests/Core/Projects/TagsProjectRunnerEventTests.cs b/tests/StdUnit.Tags.Tests/Core/Projects/TagsProjectRunnerEventTests.cs new file mode 100644 index 0000000..59bd353 --- /dev/null +++ b/tests/StdUnit.Tags.Tests/Core/Projects/TagsProjectRunnerEventTests.cs @@ -0,0 +1,402 @@ +using System; +using System.Collections.Concurrent; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using System.Xml.Linq; +using Microsoft.Extensions.DependencyInjection; +using StdUnit.Tags.Tests.Fakes; +using Xunit; + +namespace StdUnit.Tags.Tests.Core.Projects; + +/// +/// 验证 / 的多播语义:
+/// - 多个订阅者按注册顺序串行执行,且每个订阅者的 都被 await;
+/// - 任一订阅者抛异常即中断本次通知(排在它后面的订阅者不再被调用),异常原样抛出并走入口故障路径;
+/// - 运行期( 之后)订阅仍能收到下一次通知(不是"转发时抓一次快照");
+/// - 运行期退订从下一次通知起生效。 +///
+public class TagsProjectRunnerEventTests : IDisposable +{ + private const int TimeoutMs = 10_000; + private const int ScanIntervalMs = 10; + + private readonly ServiceProvider _root; + private readonly string _projectRoot; + + public TagsProjectRunnerEventTests() + { + var services = new ServiceCollection(); + services.AddLogging(); + services.AddTagsProjectServices(b => + { + b.AddFakedSupport(); + b.AddFakedTagSupport(); + }); + + _root = services.BuildServiceProvider(); + _projectRoot = TestPaths.TempPath("TagsProjectRunnerEventTests"); + Directory.CreateDirectory(_projectRoot); + } + + public void Dispose() + { + _root.Dispose(); + } + + private static XElement Channel(string name) => + new("Channel", new XAttribute("name", name), new XAttribute("driver", "fake")); + + private static XElement Entry(string name, string channelName) => + new("TagGrp", + new XAttribute("name", name), + new XAttribute("isEntry", "true"), + new XAttribute("isEnabled", "true"), + new XAttribute("channel", channelName), + new XAttribute("scanInterval", ScanIntervalMs), + new XElement("Tag", + new XAttribute("name", "v"), + new XAttribute("type", "INT16"), + new XAttribute("access", "RW"), + new XAttribute("address", "0"))); + + private ITagsProject CreateSingleEntryProject(IServiceScope scope) => + scope.ServiceProvider.GetRequiredService().Create( + _projectRoot, + new XElement("Project", Channel("fake"), Entry("g1", "fake"))); + + private static async Task WaitOrFailAsync(Task task, string what) + { + var completed = await Task.WhenAny(task, Task.Delay(TimeoutMs)).ConfigureAwait(false); + Assert.True(ReferenceEquals(completed, task), $"等待 {what} 超时({TimeoutMs}ms)"); + } + + private static async Task StopAsync(CancellationTokenSource cts, Task run) + { + cts.Cancel(); + + var finished = await Task.WhenAny(run, Task.Delay(TimeoutMs)).ConfigureAwait(false); + Assert.True(ReferenceEquals(finished, run), "取消后轮询循环仍未退出"); + + try + { + await run.ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // 取消时轮询循环以 OCE 退出,这是既有设计(见 LogicetCancellationTests)。 + } + } + + [Fact] + public async Task RunnerStarted_MultipleSubscribers_InvokedSequentiallyInRegistrationOrder() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var log = new ConcurrentQueue(); + var firstCompleted = false; + var secondSawFirstCompleted = false; + var secondCalled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + proj.RunnerStarted += async (grp, ch, ct) => + { + await Task.Delay(50, ct); + log.Enqueue("A:" + grp.TagName()); + Volatile.Write(ref firstCompleted, true); + }; + proj.RunnerStarted += (grp, ch, ct) => + { + Volatile.Write(ref secondSawFirstCompleted, Volatile.Read(ref firstCompleted)); + log.Enqueue("B:" + grp.TagName()); + secondCalled.TrySetResult(true); + return Task.CompletedTask; + }; + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(secondCalled.Task, "第二个 RunnerStarted 订阅者被调用"); + + Assert.True( + secondSawFirstCompleted, + "第二个订阅者必须在第一个订阅者的 Task 完成后才被调用:多播必须逐个 await,而不是只等最后一个"); + Assert.Equal(new[] { "A:g1", "B:g1" }, log.ToArray()); + + await StopAsync(cts, run); + } + + [Fact] + public async Task RunnerCrashed_FirstSubscriberThrows_LaterSubscribersNotInvoked() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + // 触发崩溃:轮次回调抛异常(与 Logicet 抛异常同义)。 + _ = proj.SubscribeTurnProcess("g1", (_, _, _) => throw new InvalidOperationException("observer boom")); + + var log = new ConcurrentQueue(); + var secondCalled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + proj.RunnerCrashed += (grp, ch, ex, ct) => + { + log.Enqueue("A"); + throw new InvalidOperationException("A boom"); + }; + proj.RunnerCrashed += (grp, ch, ex, ct) => + { + log.Enqueue("B"); + secondCalled.TrySetResult(true); + return Task.CompletedTask; + }; + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + // 异常原样按"该入口停机"处理:runner 记录 Critical 后重抛,RunAsync 因此结束。 + var thrown = await Assert.ThrowsAsync(() => run); + Assert.Equal("A boom", thrown.Message); + + // 第一个订阅者抛出后,本次通知就此中断:后续订阅者不会被调用。 + Assert.Equal(new[] { "A" }, log.ToArray()); + Assert.False(secondCalled.Task.IsCompleted); + } + + [Fact] + public async Task LateSubscribers_ReceiveNextCrashAndReconnectStarted() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var firstTurn = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var crashOn = false; + var turnCount = 0; + + _ = proj.SubscribeTurnProcess("g1", (_, _, _) => + { + if (Volatile.Read(ref crashOn)) + { + throw new InvalidOperationException("late boom"); + } + if (Interlocked.Increment(ref turnCount) >= 1) + { + firstTurn.TrySetResult(true); + } + return Task.CompletedTask; + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + // 首轮跑完 ⇒ PrepareEntries 早已完成(runner 回调那时就转发好了)。 + await WaitOrFailAsync(firstTurn.Task, "首轮完成"); + + // 运行期补挂:修复前这两个订阅永远收不到任何通知(转发抓的是 PrepareEntries 那一刻的快照)。 + var crashSeen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var restarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + proj.RunnerCrashed += (grp, ch, ex, ct) => + { + crashSeen.TrySetResult(ex); + return Task.CompletedTask; + }; + proj.RunnerStarted += (grp, ch, ct) => + { + restarted.TrySetResult(true); + return Task.CompletedTask; + }; + + Volatile.Write(ref crashOn, true); + + await WaitOrFailAsync(crashSeen.Task, "晚挂的 RunnerCrashed 收到崩溃"); + Assert.Equal("late boom", (await crashSeen.Task).Message); + + // 只崩一次,避免继续刷崩溃 + Volatile.Write(ref crashOn, false); + await WaitOrFailAsync(restarted.Task, "晚挂的 RunnerStarted 收到重连后的再次启动"); + + await StopAsync(cts, run); + } + + [Fact] + public async Task UnsubscribedWhileRunning_NotInvokedFromNextNotification() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var aCount = 0; + var firstStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + RunnerStarted handler = (grp, ch, ct) => + { + if (Interlocked.Increment(ref aCount) == 1) + { + firstStarted.TrySetResult(true); + } + return Task.CompletedTask; + }; + proj.RunnerStarted += handler; + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(firstStarted.Task, "首次 RunnerStarted"); + Assert.Equal(1, Volatile.Read(ref aCount)); + + proj.RunnerStarted -= handler; + + // 换一个新订阅者作为"重连确实发生了"的探针(晚挂也必须能收到)。 + var reconnect = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + proj.RunnerStarted += (grp, ch, ct) => + { + reconnect.TrySetResult(true); + return Task.CompletedTask; + }; + + // 触发一次崩溃,让运行器重连(重连时会再次触发 RunnerStarted)。 + var crashOnce = 1; + _ = proj.SubscribeTurnProcess("g1", (_, _, _) => + { + if (Interlocked.Exchange(ref crashOnce, 0) == 1) + { + throw new InvalidOperationException("once boom"); + } + return Task.CompletedTask; + }); + + await WaitOrFailAsync(reconnect.Task, "重连后的 RunnerStarted"); + Assert.Equal(1, Volatile.Read(ref aCount)); + + await StopAsync(cts, run); + } + + [Fact] + public async Task RunnerStarted_FirstSubscriberThrows_LaterSubscribersNotInvoked_AndEntryFails() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var secondCalled = false; + proj.RunnerStarted += (grp, ch, ct) => throw new InvalidOperationException("A boom"); + proj.RunnerStarted += (grp, ch, ct) => + { + Volatile.Write(ref secondCalled, true); + return Task.CompletedTask; + }; + + var crashed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + proj.RunnerCrashed += (grp, ch, ex, ct) => + { + crashed.TrySetResult(ex); + return Task.CompletedTask; + }; + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(crashed.Task, "订阅者的异常被视为该入口本轮失败"); + + var captured = await crashed.Task; + var ex = Assert.IsType(captured); + Assert.Equal("A boom", ex.Message); + + // 第一个订阅者抛出后,本次通知就此中断(不是"先收集、后只抛一个")。 + Assert.False(Volatile.Read(ref secondCalled)); + + await StopAsync(cts, run); + } + + [Fact] + public async Task RunnerCrashed_UnsubscribedAfterStart_StopsBeingNotified() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var crashOn = false; + var firstTurn = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _ = proj.SubscribeTurnProcess("g1", (_, _, _) => + { + if (Volatile.Read(ref crashOn)) + { + throw new InvalidOperationException("boom"); + } + + firstTurn.TrySetResult(true); + return Task.CompletedTask; + }); + + var removedCalls = 0; + RunnerCrashed removed = (_, _, _, _) => + { + Interlocked.Increment(ref removedCalls); + return Task.CompletedTask; + }; + proj.RunnerCrashed += removed; + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + await WaitOrFailAsync(firstTurn.Task, "首轮完成"); + + // 运行期退订(晚挂的反向操作)必须从下一次通知起生效;换一个新订阅者作为"崩溃确实发生了"的探针。 + proj.RunnerCrashed -= removed; + var crashed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + proj.RunnerCrashed += (_, _, _, _) => + { + crashed.TrySetResult(true); + return Task.CompletedTask; + }; + + Volatile.Write(ref crashOn, true); + + await WaitOrFailAsync(crashed.Task, "新订阅者收到崩溃"); + Assert.Equal(0, Volatile.Read(ref removedCalls)); + + Volatile.Write(ref crashOn, false); + await StopAsync(cts, run); + } + + [Fact] + public async Task RunnerCrashed_NoSubscribers_CrashStillRecovers() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + // 没有任何 RunnerCrashed 订阅者:崩溃通知是空操作,不能因此抛空引用、也不能让入口停摆。 + var crashOnce = 0; + var turns = 0; + var firstTurn = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var recovered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + _ = proj.SubscribeTurnProcess("g1", (_, _, _) => + { + if (Interlocked.Exchange(ref crashOnce, 0) == 1) + { + throw new InvalidOperationException("boom"); + } + + var n = Interlocked.Increment(ref turns); + if (n >= 1) + { + firstTurn.TrySetResult(true); + } + if (n >= 2) + { + recovered.TrySetResult(true); + } + + return Task.CompletedTask; + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(firstTurn.Task, "首轮完成"); + + Interlocked.Exchange(ref crashOnce, 1); + + await WaitOrFailAsync(recovered.Task, "崩溃后自动重连并继续轮询"); + Assert.False(run.IsFaulted, run.Exception?.ToString()); + + await StopAsync(cts, run); + } +} + diff --git a/tests/StdUnit.Tags.Tests/Core/Projects/TagsProjectTurnProcessTests.cs b/tests/StdUnit.Tags.Tests/Core/Projects/TagsProjectTurnProcessTests.cs index 9642b58..439ac04 100644 --- a/tests/StdUnit.Tags.Tests/Core/Projects/TagsProjectTurnProcessTests.cs +++ b/tests/StdUnit.Tags.Tests/Core/Projects/TagsProjectTurnProcessTests.cs @@ -12,15 +12,15 @@ namespace StdUnit.Tags.Tests.Core.Projects; /// -/// 验证 :
-/// - 按入口设置,某入口每一轮触发一次,且只触发该入口的回调;
-/// - 排在 Logicets 之后(回调看到的是逻辑处理完成后的值);
-/// - 每个入口最多一个回调:重复设置是替换,多个观察者由调用方自己组合(顺序显式);
-/// - 运行前/运行中都能设置:运行期设置与清除都从下一轮开始生效,传 null 表示清除;
-/// - 项目已释放后设置抛 ;
-/// - 回调抛出的异常不被吞掉(走 的恢复路径);
+/// 验证 :
+/// - 按入口订阅,某入口每一轮触发一次,且只触发该入口的订阅者;
+/// - 排在 Logicets 之后(订阅者看到的是逻辑处理完成后的值);
+/// - 每个入口可以有多个订阅者:按注册顺序串行 await,任一订阅者抛异常即中断后续(本轮失败);
+/// - 运行前/运行中都能订阅:运行期订阅与退订都从下一轮开始生效;
+/// - 句柄 Dispose 即退订(幂等),项目已释放后再 Dispose 是无操作;
+/// - 项目已释放后订阅抛 ;
/// - 入口被禁用时不会产生轮次,因此不会触发;
-/// - 未知入口 / null 入口按契约抛异常。 +/// - 未知入口 / null 入口 / null 回调按契约抛异常。 ///
public class TagsProjectTurnProcessTests : IDisposable { @@ -120,7 +120,7 @@ private static async Task StopAsync(CancellationTokenSource cts, Task run) } [Fact] - public async Task Handler_FiresEveryTurn_ForRegisteredEntry() + public async Task Subscription_FiresEveryTurn_ForRegisteredEntry() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); @@ -128,7 +128,7 @@ public async Task Handler_FiresEveryTurn_ForRegisteredEntry() var count = new int[1]; var threeTurns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - proj.SetTurnProcessHandler("g1", (entry, channel, _) => + proj.SubscribeTurnProcess("g1", (entry, channel, _) => { Assert.Equal("g1", entry.TagName()); if (Interlocked.Increment(ref count[0]) >= 3) @@ -148,7 +148,7 @@ public async Task Handler_FiresEveryTurn_ForRegisteredEntry() } [Fact] - public async Task Handler_IsScopedToItsEntry() + public async Task Subscription_IsScopedToItsEntry() { // 通道不能跨入口共用(一个通道 = 一个轮询回路),所以每个入口用自己的通道实例。 using var scope = _root.CreateScope(); @@ -163,7 +163,7 @@ public async Task Handler_IsScopedToItsEntry() var g1Turns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var g2Turns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - proj.SetTurnProcessHandler("g1", (entry, channel, _) => + proj.SubscribeTurnProcess("g1", (entry, channel, _) => { if (Interlocked.Increment(ref g1Calls[0]) >= 2) { @@ -171,7 +171,7 @@ public async Task Handler_IsScopedToItsEntry() } return Task.CompletedTask; }); - proj.SetTurnProcessHandler("g2", (entry, channel, _) => + proj.SubscribeTurnProcess("g2", (entry, channel, _) => { if (Interlocked.Increment(ref g2Calls[0]) >= 2) { @@ -193,7 +193,7 @@ public async Task Handler_IsScopedToItsEntry() } [Fact] - public async Task Handler_OnlyRegisteredEntryIsNotified() + public async Task Subscription_OnlyRegisteredEntryIsNotified() { using var scope = _root.CreateScope(); using var proj = CreateProject(scope.ServiceProvider, new XElement("Project", @@ -205,7 +205,7 @@ public async Task Handler_OnlyRegisteredEntryIsNotified() var entries = new ConcurrentDictionary(StringComparer.Ordinal); var g1Turns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - proj.SetTurnProcessHandler("g1", (entry, channel, _) => + proj.SubscribeTurnProcess("g1", (entry, channel, _) => { entries.AddOrUpdate(entry.TagName(), 1, (_, old) => old + 1); g1Turns.TrySetResult(true); @@ -218,7 +218,7 @@ public async Task Handler_OnlyRegisteredEntryIsNotified() await WaitOrFailAsync(g1Turns.Task, "g1 的回调"); await Task.Delay(ScanIntervalMs * 3); - // 只为 g1 注册:无论 g2 跑了多少轮,都不应收到 g2 的通知。 + // 只为 g1 订阅:无论 g2 跑了多少轮,都不应收到 g2 的通知。 Assert.True(entries.ContainsKey("g1")); Assert.DoesNotContain("g2", entries.Keys); @@ -226,7 +226,7 @@ public async Task Handler_OnlyRegisteredEntryIsNotified() } [Fact] - public async Task Handler_RunsAfterLogicets() + public async Task Subscription_RunsAfterLogicets() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); @@ -237,7 +237,7 @@ public async Task Handler_RunsAfterLogicets() proj.Logicets.Add(new SetValueLogicet(proj, tag, written)); var observed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - proj.SetTurnProcessHandler("g1", (entry, channel, _) => + proj.SubscribeTurnProcess("g1", (entry, channel, _) => { observed.TrySetResult(tag.Value); return Task.CompletedTask; @@ -254,7 +254,7 @@ public async Task Handler_RunsAfterLogicets() } [Fact] - public async Task Handler_NotRegistered_DoesNotThrow() + public async Task NoSubscription_DoesNotThrow() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); @@ -262,31 +262,34 @@ public async Task Handler_NotRegistered_DoesNotThrow() using var cts = new CancellationTokenSource(); var run = proj.RunAsync(cts.Token); - // 让它跑几轮:没有注册回调时,内部必须走"查不到就跳过"而不是空引用。 + // 让它跑几轮:没有订阅者时,内部必须走"查不到就跳过"而不是空引用。 await Task.Delay(ScanIntervalMs * 5); await StopAsync(cts, run); } [Fact] - public async Task Set_SecondCall_ReplacesPreviousHandler() + public async Task MultipleSubscriptions_InvokedInRegistrationOrder_AndEachTaskAwaited() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); - var first = new int[1]; - var second = new int[1]; + var log = new List(); + var firstCompleted = false; + var secondSawFirstCompleted = false; var secondCalled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - proj.SetTurnProcessHandler("g1", (entry, channel, _) => + // 第一个订阅者异步完成:第二个必须等它返回的 Task 完成之后才被调用。 + proj.SubscribeTurnProcess("g1", async (entry, channel, ct) => { - Interlocked.Increment(ref first[0]); - return Task.CompletedTask; + await Task.Delay(30); + lock (log) { log.Add("first"); } + Volatile.Write(ref firstCompleted, true); }); - // 替换语义:后设置的生效,前一个不再被调用。 - proj.SetTurnProcessHandler("g1", (entry, channel, _) => + proj.SubscribeTurnProcess("g1", (entry, channel, ct) => { - Interlocked.Increment(ref second[0]); + Volatile.Write(ref secondSawFirstCompleted, Volatile.Read(ref firstCompleted)); + lock (log) { log.Add("second"); } secondCalled.TrySetResult(true); return Task.CompletedTask; }); @@ -294,60 +297,60 @@ public async Task Set_SecondCall_ReplacesPreviousHandler() using var cts = new CancellationTokenSource(); var run = proj.RunAsync(cts.Token); - await WaitOrFailAsync(secondCalled.Task, "后设置的回调被调用"); + await WaitOrFailAsync(secondCalled.Task, "第二个订阅者被调用"); + Assert.True(secondSawFirstCompleted, "多个订阅者必须逐个 await,而不是只等最后一个"); - Assert.True(Volatile.Read(ref second[0]) >= 1); - Assert.Equal(0, Volatile.Read(ref first[0])); + string[] snapshot; + lock (log) { snapshot = log.ToArray(); } + Assert.Equal("first", snapshot[0]); + Assert.Equal("second", snapshot[1]); await StopAsync(cts, run); } [Fact] - public async Task Set_ComposedHandler_InvokesObserversInDeclaredOrder() + public async Task SubscribeAfterRunStarted_TakesEffectFromNextRound() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); - var order = new List(); - var composed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - - // 多个观察者由调用方自己组合:顺序在组合处显式可见,而不是依赖"注册顺序"。 - proj.SetTurnProcessHandler("g1", async (entry, channel, ct) => - { - await Task.CompletedTask; - lock (order) { order.Add("first"); } - - await Task.CompletedTask; - lock (order) { order.Add("second"); } - - composed.TrySetResult(true); - }); - using var cts = new CancellationTokenSource(); var run = proj.RunAsync(cts.Token); - await WaitOrFailAsync(composed.Task, "组合回调被执行"); + // 运行期订阅:下一轮开始生效(订阅表是写时复制的,轮内读取的是开始时拿到的快照)。 + var count = new int[1]; + var twoTurns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + proj.SubscribeTurnProcess("g1", (entry, channel, _) => + { + if (Interlocked.Increment(ref count[0]) >= 2) + { + twoTurns.TrySetResult(true); + } + return Task.CompletedTask; + }); - string[] snapshot; - lock (order) { snapshot = order.ToArray(); } - Assert.Equal(new[] { "first", "second" }, snapshot); + await WaitOrFailAsync(twoTurns.Task, "运行期订阅的回调从下一轮开始触发"); + Assert.True(Volatile.Read(ref count[0]) >= 2); await StopAsync(cts, run); } [Fact] - public async Task Set_Null_ClearsHandler() + public async Task DisposeHandle_StopsFiring() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); var count = new int[1]; - proj.SetTurnProcessHandler("g1", (entry, channel, _) => + var subscription = proj.SubscribeTurnProcess("g1", (entry, channel, _) => { Interlocked.Increment(ref count[0]); return Task.CompletedTask; }); - proj.SetTurnProcessHandler("g1", null); + + // 幂等:重复 Dispose 不应抛,也不应影响其它订阅者。 + subscription.Dispose(); + subscription.Dispose(); using var cts = new CancellationTokenSource(); var run = proj.RunAsync(cts.Token); @@ -359,41 +362,54 @@ public async Task Set_Null_ClearsHandler() } [Fact] - public async Task SetTurnProcessHandler_AfterRunStarted_TakesEffectFromNextRound() + public async Task DisposeHandle_RemovesOnlyThatSubscriber() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); - using var cts = new CancellationTokenSource(); - var run = proj.RunAsync(cts.Token); + var first = new int[1]; + var second = new int[1]; + var secondCalled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - // 运行期挂载:下一轮开始生效(回调表是写时复制的,选中快照不会被并发改坏)。 - var count = new int[1]; - var twoTurns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - proj.SetTurnProcessHandler("g1", (entry, channel, _) => + var firstSubscription = proj.SubscribeTurnProcess("g1", (entry, channel, _) => { - if (Interlocked.Increment(ref count[0]) >= 2) - { - twoTurns.TrySetResult(true); - } + Interlocked.Increment(ref first[0]); + return Task.CompletedTask; + }); + proj.SubscribeTurnProcess("g1", (entry, channel, _) => + { + Interlocked.Increment(ref second[0]); + secondCalled.TrySetResult(true); return Task.CompletedTask; }); - await WaitOrFailAsync(twoTurns.Task, "运行期设置的回调从下一轮开始触发"); - Assert.True(Volatile.Read(ref count[0]) >= 2); + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + await WaitOrFailAsync(secondCalled.Task, "两个订阅者都开始工作"); + + firstSubscription.Dispose(); + + // 等在途轮次用掉旧快照后取基线,之后的轮次不应再进第一个订阅者。 + await Task.Delay(ScanIntervalMs * 3); + var baselineFirst = Volatile.Read(ref first[0]); + var baselineSecond = Volatile.Read(ref second[0]); + + await Task.Delay(ScanIntervalMs * 10); + Assert.Equal(baselineFirst, Volatile.Read(ref first[0])); + Assert.True(Volatile.Read(ref second[0]) > baselineSecond); await StopAsync(cts, run); } [Fact] - public async Task SetTurnProcessHandler_ClearedWhileRunning_StopsFiringFromNextRound() + public async Task DisposeWhileRunning_StopsFiringFromNextRound() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); var count = new int[1]; var firstTurn = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - proj.SetTurnProcessHandler("g1", (entry, channel, _) => + var subscription = proj.SubscribeTurnProcess("g1", (entry, channel, _) => { Interlocked.Increment(ref count[0]); firstTurn.TrySetResult(true); @@ -404,8 +420,8 @@ public async Task SetTurnProcessHandler_ClearedWhileRunning_StopsFiringFromNextR var run = proj.RunAsync(cts.Token); await WaitOrFailAsync(firstTurn.Task, "回调先触发一次"); - // 运行期清除:等在途轮次用掉旧快照后取基线,之后的轮次都不应再进来。 - proj.SetTurnProcessHandler("g1", null); + // 运行期退订:等在途轮次用掉旧快照后取基线,之后的轮次都不应再进来。 + subscription.Dispose(); await Task.Delay(ScanIntervalMs * 3); var baseline = Volatile.Read(ref count[0]); @@ -416,10 +432,11 @@ public async Task SetTurnProcessHandler_ClearedWhileRunning_StopsFiringFromNextR } [Fact] - public async Task SetTurnProcessHandler_OnDisposedProject_Throws() + public async Task SubscribeOnDisposedProject_Throws_AndDisposeHandleIsNoOp() { using var scope = _root.CreateScope(); var proj = CreateSingleEntryProject(scope); + var subscription = proj.SubscribeTurnProcess("g1", (entry, channel, _) => Task.CompletedTask); using var cts = new CancellationTokenSource(); var run = proj.RunAsync(cts.Token); @@ -428,11 +445,15 @@ public async Task SetTurnProcessHandler_OnDisposedProject_Throws() proj.Dispose(); Assert.Throws( - () => proj.SetTurnProcessHandler("g1", (entry, channel, _) => Task.CompletedTask)); + () => proj.SubscribeTurnProcess("g1", (entry, channel, _) => Task.CompletedTask)); + + // 项目释放后,退订句柄仍是安全的无操作(幂等)。 + subscription.Dispose(); + subscription.Dispose(); } [Fact] - public async Task SetTurnProcessHandler_ConcurrentSetAndClearWhileRunning_DoesNotCorrupt() + public async Task ConcurrentSubscribeAndDisposeWhileRunning_DoesNotCorrupt() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); @@ -443,14 +464,10 @@ public async Task SetTurnProcessHandler_ConcurrentSetAndClearWhileRunning_DoesNo using var stopMutating = new CancellationTokenSource(); var mutator = Task.Run(() => { - var i = 0; while (!stopMutating.IsCancellationRequested) { - proj.SetTurnProcessHandler("g1", (entry, channel, _) => Task.CompletedTask); - if ((i++ % 2) == 0) - { - proj.SetTurnProcessHandler("g1", null); - } + var subscription = proj.SubscribeTurnProcess("g1", (entry, channel, _) => Task.CompletedTask); + subscription.Dispose(); } }); @@ -458,19 +475,19 @@ public async Task SetTurnProcessHandler_ConcurrentSetAndClearWhileRunning_DoesNo stopMutating.Cancel(); await mutator; - // 并发设置/清除不应让轮询崩掉(快照替换 + 轮询线程只读)。 + // 并发订阅/退订不应让轮询崩掉(快照替换 + 轮询线程只读)。 Assert.False(run.IsFaulted, run.Exception?.ToString()); await StopAsync(cts, run); } [Fact] - public async Task Handler_SubscriberThrows_EntersRunnerCrashed() + public async Task SubscriberThrows_EntersRunnerCrashed() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); - proj.SetTurnProcessHandler( + proj.SubscribeTurnProcess( "g1", (entry, channel, _) => throw new InvalidOperationException("observer boom")); @@ -493,7 +510,43 @@ public async Task Handler_SubscriberThrows_EntersRunnerCrashed() } [Fact] - public async Task Handler_DisabledEntry_NotFired() + public async Task FirstSubscriberThrows_LaterSubscribersNotInvoked() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var secondCalled = false; + proj.SubscribeTurnProcess("g1", (entry, channel, _) => throw new InvalidOperationException("first boom")); + proj.SubscribeTurnProcess("g1", (entry, channel, _) => + { + Volatile.Write(ref secondCalled, true); + return Task.CompletedTask; + }); + + var crashed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + proj.RunnerCrashed += (entry, channel, ex, ct) => + { + crashed.TrySetResult(ex); + return Task.CompletedTask; + }; + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(crashed.Task, "第一个订阅者的异常触发崩溃路径"); + + var captured = await crashed.Task; + var ex = Assert.IsType(captured); + Assert.Equal("first boom", ex.Message); + + // 首个异常即中断:排在后面的订阅者本次不会被调用。 + Assert.False(Volatile.Read(ref secondCalled)); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Subscription_DisabledEntry_NotFired() { using var scope = _root.CreateScope(); using var proj = CreateProject(scope.ServiceProvider, new XElement("Project", @@ -501,7 +554,7 @@ public async Task Handler_DisabledEntry_NotFired() Entry("g1", "fake", isEnabled: false))); var count = new int[1]; - proj.SetTurnProcessHandler("g1", (entry, channel, _) => + proj.SubscribeTurnProcess("g1", (entry, channel, _) => { Interlocked.Increment(ref count[0]); return Task.CompletedTask; @@ -518,7 +571,7 @@ public async Task Handler_DisabledEntry_NotFired() } [Fact] - public async Task Handler_ReceivesProjectCancellationToken() + public async Task Subscription_ReceivesProjectCancellationToken() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); @@ -526,7 +579,7 @@ public async Task Handler_ReceivesProjectCancellationToken() using var cts = new CancellationTokenSource(); var observed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - proj.SetTurnProcessHandler("g1", (entry, channel, ct) => + proj.SubscribeTurnProcess("g1", (entry, channel, ct) => { observed.TrySetResult(ct); return Task.CompletedTask; @@ -536,40 +589,124 @@ public async Task Handler_ReceivesProjectCancellationToken() await WaitOrFailAsync(observed.Task, "回调收到取消令牌"); var captured = await observed.Task; + + // 不必与 RunAsync 的入参是同一个实例:框架用内部 linked 令牌启动各入口(启动失败时要能回滚已启动的入口)。 + // 契约是"取消时机",不是引用相等。 Assert.True(captured.CanBeCanceled); - Assert.Equal(cts.Token, captured); + Assert.False(captured.IsCancellationRequested, "取消前回调拿到的令牌不应已取消"); await StopAsync(cts, run); + + Assert.True(captured.IsCancellationRequested, "项目停止后,回调拿到的令牌也应取消"); } [Fact] - public void SetTurnProcessHandler_UnknownEntry_Throws() + public void Subscribe_UnknownEntry_Throws() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); Assert.Throws( - () => proj.SetTurnProcessHandler("nope", (entry, channel, _) => Task.CompletedTask)); + () => proj.SubscribeTurnProcess("nope", (entry, channel, _) => Task.CompletedTask)); + } + + [Fact] + public void Subscribe_NullHandler_Throws() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + // 退订必须用 Dispose 句柄,而不是传 null。 + Assert.Throws(() => proj.SubscribeTurnProcess("g1", null!)); + } + + [Fact] + public async Task UnsubscribeAllThenResubscribe_Works() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + // 退到"该入口没有任何订阅者"(内部留下空数组)之后仍能重新订阅。 + var firstSubscription = proj.SubscribeTurnProcess("g1", (entry, channel, _) => Task.CompletedTask); + firstSubscription.Dispose(); + + var count = new int[1]; + var twoTurns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + proj.SubscribeTurnProcess("g1", (entry, channel, _) => + { + if (Interlocked.Increment(ref count[0]) >= 2) + { + twoTurns.TrySetResult(true); + } + return Task.CompletedTask; + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(twoTurns.Task, "退订后重新订阅仍能收到轮次"); + Assert.True(Volatile.Read(ref count[0]) >= 2); + + await StopAsync(cts, run); } [Fact] - public void SetTurnProcessHandler_NullHandler_IsAllowed() + public async Task DisposeMiddleSubscription_RemovesOnlyThatOne() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); - // null = 清除;在从未设置过的入口上清除也是合法的空操作。 - proj.SetTurnProcessHandler("g1", null); - proj.SetTurnProcessHandler("g1", (entry, channel, _) => Task.CompletedTask); - proj.SetTurnProcessHandler("g1", null); + var first = new int[1]; + var middle = new int[1]; + var last = new int[1]; + var threeTurns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + proj.SubscribeTurnProcess("g1", (entry, channel, _) => + { + Interlocked.Increment(ref first[0]); + return Task.CompletedTask; + }); + var middleSubscription = proj.SubscribeTurnProcess("g1", (entry, channel, _) => + { + Interlocked.Increment(ref middle[0]); + return Task.CompletedTask; + }); + proj.SubscribeTurnProcess("g1", (entry, channel, _) => + { + if (Interlocked.Increment(ref last[0]) >= 2) + { + threeTurns.TrySetResult(true); + } + return Task.CompletedTask; + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + await WaitOrFailAsync(threeTurns.Task, "最后一个订阅者连续触发"); + + // 摘掉中间那个(数组里 index > 0),其余两个不受影响。 + middleSubscription.Dispose(); + + await Task.Delay(ScanIntervalMs * 3); + var baselineFirst = Volatile.Read(ref first[0]); + var baselineMiddle = Volatile.Read(ref middle[0]); + var baselineLast = Volatile.Read(ref last[0]); + + await Task.Delay(ScanIntervalMs * 10); + Assert.True(Volatile.Read(ref first[0]) > baselineFirst); + Assert.Equal(baselineMiddle, Volatile.Read(ref middle[0])); + Assert.True(Volatile.Read(ref last[0]) > baselineLast); + + await StopAsync(cts, run); } [Fact] - public void SetTurnProcessHandler_NullEntry_Throws() + public void Subscribe_NullEntry_Throws() { using var scope = _root.CreateScope(); using var proj = CreateSingleEntryProject(scope); - Assert.Throws(() => proj.SetTurnProcessHandler(null!, (entry, channel, _) => Task.CompletedTask)); + Assert.Throws( + () => proj.SubscribeTurnProcess(null!, (entry, channel, _) => Task.CompletedTask)); } } diff --git a/tests/StdUnit.Tags.Tests/Core/TagGrpRunners/TagGrpRunnerTests.cs b/tests/StdUnit.Tags.Tests/Core/TagGrpRunners/TagGrpRunnerTests.cs index e45954f..06d833a 100644 --- a/tests/StdUnit.Tags.Tests/Core/TagGrpRunners/TagGrpRunnerTests.cs +++ b/tests/StdUnit.Tags.Tests/Core/TagGrpRunners/TagGrpRunnerTests.cs @@ -983,7 +983,7 @@ public void Dispose() { } public event RunnerStarted? RunnerStarted; public event RunnerCrashed? RunnerCrashed; - public void SetTurnProcessHandler(string entry, TurnProcess? handler) => throw new NotImplementedException(); + public IDisposable SubscribeTurnProcess(string entry, TurnProcess handler) => throw new NotImplementedException(); } /// diff --git a/tests/StdUnit.Tags.Tests/Core/TagsProjectCtrls/MockTagsProject.cs b/tests/StdUnit.Tags.Tests/Core/TagsProjectCtrls/MockTagsProject.cs index 638a2ef..284c0a1 100644 --- a/tests/StdUnit.Tags.Tests/Core/TagsProjectCtrls/MockTagsProject.cs +++ b/tests/StdUnit.Tags.Tests/Core/TagsProjectCtrls/MockTagsProject.cs @@ -48,7 +48,7 @@ internal class MockTagsProject : ITagsProject public event RunnerStarted? RunnerStarted; public event RunnerCrashed? RunnerCrashed; - public void SetTurnProcessHandler(string entry, TurnProcess? handler) => throw new NotImplementedException(); + public IDisposable SubscribeTurnProcess(string entry, TurnProcess handler) => throw new NotImplementedException(); public void Initialize(string projRoot, XElement? root = null) { InitializeCallCount++;