From 763c0151037c86b9fa8d6f12e0ebb01a56cecc11 Mon Sep 17 00:00:00 2001 From: Disy Date: Tue, 6 Oct 2026 17:36:00 +0800 Subject: [PATCH] feat: protect mod payloads with main-thread ownership in 1.1.17 --- README.md | 28 ++- build.gradle | 43 ++++ docs/payload-ownership-1.1.17-validation.md | 76 +++++++ .../plans/2026-10-06-payload-ownership.md | 133 ++++++++++++ .../2026-10-06-payload-ownership-design.md | 2 +- gradle.properties | 2 +- .../dev/modzuozhi/command/MasterCommand.java | 4 + .../compat/MainThreadPhaseCoordinator.java | 85 ++++++++ .../core/compat/PayloadDiagnostics.java | 76 +++++++ .../compat/PayloadMainThreadDispatch.java | 114 ++++++++++ .../core/compat/PayloadWorkQueue.java | 5 +- .../core/compat/SableAssemblyDispatch.java | 31 +-- .../core/compat/SableMainThreadBridge.java | 11 + .../compat/ThreadOwnershipDiagnostics.java | 101 +++++++++ .../core/dimthread/LevelTickLoop.java | 29 ++- .../core/dimthread/ServerManager.java | 56 ++++- .../ComputerCraftTickSchedulerMixin.java | 41 ++++ .../java/dev/modzuozhi/mixin/LevelMixin.java | 10 + .../mixin/MinecraftServerDimMixin.java | 8 +- .../mixin/MinecraftServerPayloadMixin.java | 16 ++ .../mixin/ServerPayloadContextMixin.java | 38 ++++ .../mixin/ThreadOwnershipReportMixin.java | 18 ++ .../resources/dimensionalripper.mixins.json | 4 + .../InteractionFlowRegression.java | 14 +- tests/ownership/OwnershipRegression.java | 200 ++++++++++++++++++ tests/payload/ModInteractionRegression.java | 78 +++++++ tests/payload/PayloadRegression.java | 200 ++++++++++++++++++ tests/payload/ReviewRegression.java | 81 +++++++ tests/payload/SimulatedHandleRegression.java | 101 +++++++++ 29 files changed, 1552 insertions(+), 53 deletions(-) create mode 100644 docs/payload-ownership-1.1.17-validation.md create mode 100644 docs/superpowers/plans/2026-10-06-payload-ownership.md create mode 100644 src/main/java/dev/modzuozhi/core/compat/MainThreadPhaseCoordinator.java create mode 100644 src/main/java/dev/modzuozhi/core/compat/PayloadDiagnostics.java create mode 100644 src/main/java/dev/modzuozhi/core/compat/PayloadMainThreadDispatch.java create mode 100644 src/main/java/dev/modzuozhi/core/compat/ThreadOwnershipDiagnostics.java create mode 100644 src/main/java/dev/modzuozhi/mixin/ComputerCraftTickSchedulerMixin.java create mode 100644 src/main/java/dev/modzuozhi/mixin/MinecraftServerPayloadMixin.java create mode 100644 src/main/java/dev/modzuozhi/mixin/ServerPayloadContextMixin.java create mode 100644 src/main/java/dev/modzuozhi/mixin/ThreadOwnershipReportMixin.java create mode 100644 tests/ownership/OwnershipRegression.java create mode 100644 tests/payload/ModInteractionRegression.java create mode 100644 tests/payload/PayloadRegression.java create mode 100644 tests/payload/ReviewRegression.java create mode 100644 tests/payload/SimulatedHandleRegression.java diff --git a/README.md b/README.md index 1691f1a..e78ff32 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,32 @@ Dimensional Ripper 是一个基于 **NeoForge 1.21.1** 的多线程维度并行优化模组。它通过独立维度 tick 循环与细粒度并行调度，充分利用多核 CPU 提升服务端性能，同时保持稳定的 20 TPS。 +## 1.1.17：公共收包线程归属候选修复 + +服务端游戏阶段经 NeoForge `ServerPayloadContext.enqueueWork` 明确提交到主线程的工作，现在统一进入有界队列，并在维度更新归还执行权后，由真正的服务器主线程调用原处理器。Create/航空学旧组装入口复用同一安全阶段；采用这条公共入口的模组不需要主动适配。登录、配置阶段、客户端及纯网络计算不改派，也不代表任意模组的任意线程访问已自动安全。 + +队列上限为服务器 4096 项、单连接 256 项，每批最多 64 项或 2 ms（不强行中断已经开始的处理器）。保留观察到的入队顺序；执行前取消的工作不再修改世界，换世界/换玩家/停服后的旧工作明确失效。超载通过原断线流程处理；业务处理器自己抛出的拒绝异常不会被误判为超载踢人。新循环继承菜单暂停，仍在退出的循环也必须归还执行权。等待期间继续服务 Sable 主线程交接，拒绝会产生锁反转的同步重入。 + +真实本地模组回归：Sable 物理子世界内创建了真实把手约束，复现原路径松手后约束仍有效；经新入口处理同一松手包后约束失效。Create 调速包及 Jade 查询通过，原距离拒绝仍有效。另有队列、取消、暂停、停止和真实 CC 回归。详细环境、日志标记及未覆盖项见 [1.1.17 验证报告](docs/payload-ownership-1.1.17-validation.md)。这些是隔离服务器测试，不等于用户完整整合包、客户端飞行或多人压力验收。 + +阶段之外维度并行仍保留；有交互时会等待最慢维度收口，不承诺零开销。没有待处理交互时不建立收包屏障。`/dimensionalripper status` 与归属崩溃快照包含队列积压、最老等待、最近批次等待/执行时间；仅凭 Carpet 的低 MSPT 不能判断所有维度正常。记录最多 128 个具名包 ID 加 other、最近 32 条结果，失败警告全局 30 秒限频，不记录包正文。 + +**仍为测试候选，不要直接替换正式服务器：ServerWarashi（坐服童子）启动分析崩溃、C2ME 冲突、任意后台世界访问及既有停服超时不在这次修复范围内。** 工作线程提交后同步等待公共队列 future 仍不受支持；不能把它宣传为所有模组免适配。 + +## 1.1.16：CC 调度线程归属候选修复 + +针对 CC:Tweaked 1.118.0 `TickScheduler` 主线程读取正在被维度线程更新的世界、导致显示器身份检查得到 `null` 的问题：有待处理 token 时，先暂停维度续排并等待在途更新归还所有权，在服务器主线程执行原调度检查，最后恢复原先运行的循环。等待期间继续处理主线程回调，允许 Sable 交接完成；不绕过身份检查，不吞掉真实异常，不强加载未加载区块。CC 的全局延迟队列表对应的区块回调也做串行保护。未安装 CC 时不启用此补丁。 + +这不是零开销修复：CC 有任务时，安全阶段会等待当前最慢的维度收口，可能暂时影响主线程 TPS；没有任务时不建立屏障，阶段结束后维度与细粒度并行继续运行。不能保证所有维度在这一兼容阶段完全互不影响。`/dimensionalripper status` 增加最近一次交接等待/执行耗时。 + +线程诊断记录维度、Level/ChunkSource 所属线程、调用线程、位置、阶段和 Sable 交接状态。每维度最多保留 16 条最近事件；入口归属不匹配警告每维度最多 30 秒一次，抑制期间累计计数，慢安全阶段全局最多 30 秒一次。入口快照不等于原子世界快照，也不据此单独判定死锁。崩溃报告新增 `DimensionalRipper Thread Ownership`。 + +隔离测试：`gradlew runServer -PownershipRegression -PccTestJar=<本机CC的jar绝对路径>`。使用 `runs/server` 下的专用测试世界（运行前确认 `level-name=interaction-regression`、`server-ip=127.0.0.1`、`server-port=25579`），不要指向正式存档。测试源码和 CC jar 不加入常规发布包。已在修复前用真实 CC 显示器复现 `Expected MonitorBlockEntity ... got null`；修复后运行任务会检查日志中的 `OWNERSHIP REGRESSION PASS`、`OWNERSHIP EDGE PASS`、`OWNERSHIP CALLBACK PASS` 和 `OWNERSHIP RESUME PASS`。Minecraft 崩溃有时仍以零退出码结束，不能只检查 Java 退出码。 + +覆盖：真实显示器身份检查、实际主线程交接桥的回调执行、嵌套/异常后的暂停恢复、拆除与真实身份不匹配、未加载 token、80 个延迟 token 的双维度并发 ticket/unload 回调，以及崩溃诊断。1001 次入口归属不匹配在测试日志中仅输出 1 条警告。另用 1 TPS 的空闲预算复现暂停后仍等待整秒的问题，修正为世界工作完成后立即退出已请求暂停的空闲轮询；保留在途工作的收尾。一次本机测试的 CC 等待/执行为约 7.4/0.12 ms，仅是隔离样本，不是整合包性能承诺。测试调用实际交接桥，但没有加载 Sable 原生物理引擎。 + +本修复限定于 CC 调度路径，不是对所有模组跨线程访问的通用放行。仍需在完整整合包的存档副本验证，尤其 CC 显示器随 Sable 结构移动的情况；不要据此认为所有航空学、区块发送、停服问题均已解决。 + ## 优化范围（均为服务端运算，不影响客户端画面/渲染） ### 🗺️ 维度 @@ -44,7 +70,7 @@ Dimensional Ripper 是一个基于 **NeoForge 1.21.1** 的多线程维度并行 ## ⚠️ 兼容性说明 -**已知未修复问题（1.1.15）**：与 C2ME 的 `c2me_fixes_worldgen_threading_issues` 同时加载时，双方对 `Level` 构造器内随机源初始化的 `@Redirect` 冲突，可能导致启动失败。日志已确认 C2ME 优先级 1100 的重定向覆盖本模组优先级 1000 的 `modzuozhi_threadSafeRandom`，继而触发注入检查失败。当前源码尚未包含此兼容修复，不能视为可直接部署的稳定版。 +**已知未修复问题（1.1.15 / 1.1.16）**：与 C2ME 的 `c2me_fixes_worldgen_threading_issues` 同时加载时，双方对 `Level` 构造器内随机源初始化的 `@Redirect` 冲突，可能导致启动失败。日志已确认 C2ME 优先级 1100 的重定向覆盖本模组优先级 1000 的 `modzuozhi_threadSafeRandom`，继而触发注入检查失败。当前源码尚未包含此兼容修复，不能视为可直接部署的稳定版。 1.1.15 针对 Sable 碰撞查询期间反复创建实心占位区块的问题： diff --git a/build.gradle b/build.gradle index 32efb72..07bd2cc 100644 --- a/build.gradle +++ b/build.gradle @@ -77,9 +77,52 @@ dependencies { } // Opt-in isolated runtime fixture; excluded from normal builds. +if (providers.gradleProperty('payloadRegression').isPresent()) { + if (providers.gradleProperty('reviewRegression').isPresent()) { + runs.configureEach { systemProperty 'dimensionalripper.reviewRegression', 'true' } + } + if (providers.gradleProperty('interactionRegression').isPresent() || providers.gradleProperty('ownershipRegression').isPresent()) { + throw new GradleException('Run payloadRegression separately.') + } + sourceSets.main.java.srcDir('tests/payload') + if (providers.gradleProperty('payloadModsDir').isPresent()) { + dependencies { + runtimeOnly fileTree(dir: providers.gradleProperty('payloadModsDir').get(), includes: [ + 'create-1.21.1-6.0.10.starlight.jar', 'create-aeronautics-bundled-1.21.1-1.3.0.starlight.jar', + 'sable-neoforge-1.21.1-2.0.5.starlight.jar', 'veil-neoforge-1.21.1-4.2.0.starlight.jar', + 'Jade-1.21.1-NeoForge-15.10.5.starlight.jar' + ]) + } + } + tasks.matching { it.name == 'runServer' }.configureEach { + doLast { + def output = file('runs/server/logs/latest.log').getText('UTF-8') + if (providers.gradleProperty('reviewRegression').isPresent()) { + if (!output.contains('PAYLOAD REVIEW PASS') || output.contains('Encountered an unexpected exception')) + throw new GradleException('Payload review regression did not pass.') + return + } + if (!['PAYLOAD PHASE PASS', 'PAYLOAD ENTRY PASS', 'PAYLOAD ASYNC PASS', 'PAYLOAD LIFECYCLE PASS', 'PAYLOAD DIAGNOSTICS PASS'].every { output.contains(it) } + || output.contains('Encountered an unexpected exception')) + throw new GradleException('Payload phase regression did not pass.') + if (providers.gradleProperty('payloadModsDir').isPresent() + && !['SIMULATED HANDLE PACKET PASS', 'CREATE CONFIG PASS', 'JADE REQUEST PASS'].every { output.contains(it) }) + throw new GradleException('Real Simulated/Create/Jade packet regression did not pass.') + } + } +} if (providers.gradleProperty('interactionRegression').isPresent()) { sourceSets.main.java.srcDir('tests/integration') sourceSets.main.java.exclude('**/ChunkHandoffRegression.java') + tasks.matching { it.name == 'runServer' }.configureEach { + doLast { + def output = file('runs/server/logs/latest.log').getText('UTF-8') + if (!['ENTITY ROUTING REGRESSION PASS', 'MISSING CHUNK REGRESSION PASS', + 'REQUIRED CHUNK REGRESSION PASS', 'INTERACTION REGRESSION PASS'].every { output.contains(it) } + || output.contains('Encountered an unexpected exception')) + throw new GradleException('Interaction/water regression did not pass.') + } + } } if (providers.gradleProperty('ownershipRegression').isPresent()) { if (providers.gradleProperty('interactionRegression').isPresent()) { diff --git a/docs/payload-ownership-1.1.17-validation.md b/docs/payload-ownership-1.1.17-validation.md new file mode 100644 index 0000000..17d902e --- /dev/null +++ b/docs/payload-ownership-1.1.17-validation.md @@ -0,0 +1,76 @@ +# 1.1.17 公共收包归属：候选验证记录 + +日期：2026-10-06。结论：公共入口与已列交互路径已实施；不是完整整合包兼容认证，也不是 ServerWarashi 修复。 + +## 环境与范围 + +Java 21 Zulu、Minecraft 1.21.1、NeoForge 21.1.244。仅使用仓库 `runs/server` 的 `interaction-regression` 世界、127.0.0.1:25579。没有修改用户正式世界、启动器 mods 或远端仓库。用户报告过的 NeoForge 21.1.233 未在本轮单独运行。 + +可选模组为用户本机实际 jar：Create 6.0.10.starlight、Create Aeronautics bundled 1.3.0.starlight（含 Simulated）、Sable 2.0.5.starlight、Veil 4.2.0.starlight、Jade 15.10.5.starlight；CC:Tweaked 1.118.0.starlight 单独运行。 + +## 证据层级 + +| 层级 | 已验证内容 | 证据 | +|---|---|---| +| 纯 Java | FIFO、至多一次、256/4096 容量、取消回收、64 项/2 ms 预算、关闭、重入完成、8 生产者 | `runPayloadWorkQueueRegression` 的 `PAYLOAD QUEUE PASS` | +| 实际 Minecraft/NeoForge 上下文 | Runnable/Supplier、返回值、主线程异常身份、异步 guard、取消、换世界失效、配置阶段原路径、嵌套、Sable 同步重入拒绝 | `PAYLOAD PHASE/ENTRY/ASYNC PASS` | +| 实际 Minecraft 生命周期 | 新循环、菜单暂停、退出中循环仍等待、停止不恢复、停止后晚到任务不能复建诊断 | `PAYLOAD LIFECYCLE PASS`、`PAYLOAD REVIEW PASS` | +| 实际可选模组和原生物理 | 真实子世界、真实把手 BE、真实包和 Veil 上下文、Rapier 约束的创建及松手销毁 | `SIMULATED NATIVE RED` 后 `SIMULATED HANDLE PACKET PASS` | +| 实际 Create / Jade | 调速写入、查询 BE、原始距离拒绝 | `CREATE CONFIG PASS`、`JADE REQUEST PASS` | +| 实际 CC | 显示器同一性、回调等待、80 个延迟 token、双维度 ticket/unload、恢复后继续 tick | 四个 `OWNERSHIP … PASS` | +| 已有流水/区块/实体路径 | 600000 次缺失读取、400000 次已加载读取、必须生成请求安全失败、实体路由至多一次、真实水流扩散 | 四个 `… REGRESSION PASS`，其中包括 `INTERACTION REGRESSION PASS` | +| 空载/无可选模组 | 不安装上述可选模组仍启动；100 次空 drain 没有新增批次 | `PAYLOAD IDLE PASS` 及通用 payload 标记 | + +真实把手的失败证明不是“函数调用成功”：测试让维度线程持有 plot 世界归属，直接运行原松手包后物理约束仍然 `isValid=true`；相同真实包经公共安全入口运行后 `isValid=false`。测试使用 FakePlayer（有真实测试监听器），并非真人客户端输入或真实网络连接。 + +四个审查问题均先失败后通过：业务拒绝异常误触发踢人、动态循环丢失菜单暂停、退出中的写入者漏过屏障、停服清理后晚到诊断重新持有服务器。日志 `payload-review-red-2.log` 对应失败，`payload-review-final.log` 对应修正后的通过。 + +诊断分类额外使用 `payload-counter-red.log` 复现“业务失败与容量拒绝混计”，`payload-counter-green.log` 通过；`payload-candidate-real.log` 再次通过真实模组测试并检查换世界旧任务计入 expired。failed、cancelled、expired、rejected 分开统计。 + +旧交互回归第一次运行失败，不能因 Gradle 返回零而算通过：旧夹具没有 connection，且仍要求已经替换掉的单维度桥。夹具改为 FakePlayer 与公共独占断言，保留所有流水/区块检查，并补上日志 PASS/异常检测门禁；`payload-final-interaction-green.log` 才是通过记录。 + +## 本轮取舍 + +1. Windows 无 bash，以 PowerShell/apply_patch 维护执行记录；风险是记录遗漏，测试门禁不变。 +2. 主线程独占标志与菜单暂停分离；否则嵌套 resume 会提前放走维度线程。代价是多一组生命周期状态。 +3. 阶段、入口、生命周期共用夹具及文件，相关提交合并验证，不把中间状态当发布版本。 +4. 保留 NeoForge 异步业务异常返回 null 的 guard 行为，但替换无限逐条异常打印为有界诊断。显式策略失败（旧玩家/停止/溢出）保留异常 future，执行前取消传播到底层队列；这是有意的兼容语义细化，依赖“失败也返回 null”的调用方可能观察到变化。 +5. 普通主线程同步调用为保序与同步完成可连续处理多个有界批次；2 ms 不是整 tick 硬上限，慢维度和慢处理器仍可能拉长等待。独占阶段内嵌套立即运行。 +6. 旧交互夹具适配新连接/归属契约而非放松生产守卫；不能替代真实客户端网络压力验证。 +7. 既有停服 30 秒超时、ServerWarashi、C2ME、任意后台世界/光照访问保留为独立问题；它们仍可能造成完整整合包失败。 +8. 只交付隔离环境验证的候选，不将以下未测项标成通过。审查没有列出待办 Minor。 + +## 未完成的验收，不属于已修复声明 + +- 真人客户端整套飞行、抓握输入、GUI 与完整整合包存档副本测试。 +- 两玩家同时抓握、死亡、断线、换维度、拆把手、卸载子世界及溢出拒绝松手后的完整原生约束清理矩阵。已测普通真实松手、通用旧世界拒绝和溢出断线调用，不能互相替代。 +- 跨服务器实例隔离的双实例并行压力测试；实现按服务器分区，不能仅凭代码视为此项实测。 +- 空载、普通交互、密集抓握、多人查询和慢维度的完整对比压测。空队列无批次及慢 worker 等待机制已验证，不承诺开销比例或全服 20 TPS。 +- 任意 worker 提交后同步 join 主线程队列不受支持。夹具通过 250 ms 超时检测并取消这种等待环，不表示生产代码会自动化解任意 join 死锁。 +- ServerWarashi 分析立即崩溃仍需单独从采样栈修复。 + +## 重跑 + +以下命令均使用 `--no-daemon --no-configuration-cache --console=plain`，先检查隔离世界配置，不能指向正式存档： + +```powershell +.\gradlew.bat check +.\gradlew.bat runServer -PpayloadRegression +.\gradlew.bat runServer -PpayloadRegression -PreviewRegression +.\gradlew.bat runServer -PpayloadRegression '-PpayloadModsDir=<本机上述模组目录>' +.\gradlew.bat runServer -PownershipRegression '-PccTestJar=<本机CC jar>' +.\gradlew.bat runServer -PinteractionRegression +.\gradlew.bat clean build +``` + +每组运行必须完成后再启动下一组。查看对应 PASS 标记及服务器异常，而非只看 Java/Gradle 退出码。正常构建不可带任何回归夹具属性。 + +## 打包门禁 + +`check runServer -PpayloadRegression` 成功（五个纯 Java 回归程序及无可选模组服务器夹具）；随后不带任何夹具属性的 `clean build` 成功，五个纯 Java 回归再次执行通过。真实模组、CC、交互及审查运行见上述分组，未将测试标记缺失的运行计入通过。 + +发布 jar：`build/libs/dimensionalripper-1.1.17.jar`，277211 字节。ZIP 内容检查确认元数据版本为 1.1.17，包含公共入口 mixin/执行器，不含 Regression、FakePlayer、tests 或打包的可选模组 jar/类。 + +SHA256：`CF1DE863DE119272098142CF5F0ABE78543BA55B15074369A1A66FBC922FBAD3`。 + +完整本地日志已保存在 `build/reports/payload/`（忽略提交）。关键最终文件：`payload-clean-build.log`、`payload-candidate-check.log`、`payload-candidate-real.log`、`payload-counter-green.log`、`payload-final-cc.log`、`payload-final-interaction-green.log`。失败前例也保留供对照。执行工作区仍保留未完成验收记录，不把剩余矩阵标为完成。 diff --git a/docs/superpowers/plans/2026-10-06-payload-ownership.md b/docs/superpowers/plans/2026-10-06-payload-ownership.md new file mode 100644 index 0000000..76f4370 --- /dev/null +++ b/docs/superpowers/plans/2026-10-06-payload-ownership.md @@ -0,0 +1,133 @@ +# 公共收包线程归属 Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 将受支持的服务端游戏包交给真正主线程独占处理，修复把手松开路径，同时保留阶段之外的维度并行。 + +**Architecture:** 每服务器有界 FIFO 接收显式主线程工作；服务器外层 tick 建立全维度安全阶段，批量调用原处理器。阶段状态阻止递归收口和新增维度写入，旧组装入口复用同一执行权。生命周期清理与通用包失效分开处理。 + +**Tech Stack:** Java 21、Minecraft 1.21.1、NeoForge、Mixin/MixinExtras、现有 Gradle 隔离服务器回归。 + +**Spec:** `docs/superpowers/specs/2026-10-06-payload-ownership-design.md` + +**执行状态（2026-10-06）：** Task 1 已提交 `691e949`；Task 2–5 的公共入口、阶段、适配与诊断代码已实施并有回归证据。Task 4 的完整原生资源异常清理矩阵、Task 5 的双服务器压力测试、Task 6 的完整性能/客户端矩阵未全部验收，因此不勾选混合这些要求的条目。具体通过项和剩余项以 `docs/payload-ownership-1.1.17-validation.md` 为准，本轮只交付候选包。 + +## Global Constraints + +- 第一版新公共入口默认全维度隔离，不凭“玩家当前维度”或包内坐标猜测整个处理器的访问范围。 +- 初始批次上限为 64 项或 2 ms 的处理时间，达到任一条件后不启动下一项；不强行中断已开始的处理器。 +- 服务器队列上限 4096 项、单连接上限 256 项。 +- 记录上限 128 个 ID，超出归入 other，避免恶意 ID 撑大统计。 +- 登录、握手、配置阶段、客户端接收和纯网络线程处理不被无条件改派。 +- 超时用于诊断，不是允许并发执行的理由。 +- 不自动上传远端或部署正式服务器。测试仅操作仓库隔离世界或显式创建的存档副本。 +- 不改变原处理器权限、距离、强加载规则；不放松 Level 的线程守卫。 +- ServerWarashi 是后续独立修复，需要单独的采样设计及失败测试；本计划不宣称解决该崩溃。 +- C2ME、后台光照查询及既有停服超时不纳入修复声明。 + +## Review Focus + +- 主线程正在服务等待中的 Sable 回调时再次 enqueue：不得等待该回调所属 worker，任务 2、3 验证。 +- 前一个包传送/替换玩家后，同连接后续旧包不得操作新世界：任务 3、4 验证。 +- 嵌套调用导致新世界/循环出现：不得在独占期间启动另一个写入者，任务 2 验证。 +- 队列满载时恰好收到松手：拒绝要显式完成 future，断线清理不能遗留约束，任务 1、4 验证。 +- 停服与阶段 finally、开关/菜单暂停同时发生：不得恢复已停止或仍应暂停的循环，任务 2、5 验证。 + +## 执行与证据约定 + +当前工作树含尚未提交的 1.1.16 归属修复，不能用新 worktree 的裸 HEAD 替代这些依赖。执行开始按 worktree 技能检查隔离方式；保留并核对所有现有变更，禁止 reset 或覆盖。不把原有未验证内容顺手算作本次通过。 + +以下路径相对仓库根。每个任务先写断言、记录失败，再改生产代码；每任务结束显式暂存该任务文件并提交（git 使用本仓库 safe.directory）。混合文件先核对 diff，不能误纳入用户后续改动。 + +命令统一使用 Java 21、`--no-daemon --no-configuration-cache --console=plain`。新的 `-PpayloadRegression` 夹具与 `ownershipRegression`、`interactionRegression` 互斥；运行结束必须有全部指定 PASS 标记且没有 unexpected exception，不能仅依赖进程退出码。证据放 `build/reports/payload/`。 + +## 文件职责 + +- `core/compat/PayloadWorkQueue.java`：不依赖 Minecraft 的 FIFO、容量、取消及批次预算。 +- `core/compat/MainThreadPhaseCoordinator.java`：服务器阶段状态与独占租约。 +- `core/compat/PayloadMainThreadDispatch.java`：实际玩家上下文、future、生命周期与 drain。 +- `core/compat/PayloadDiagnostics.java`：有界计数及耗时快照。 +- `mixin/ServerPayloadContextMixin.java`：两个 enqueueWork 重载，保留原有分支及 guard。 +- `mixin/MinecraftServerPayloadMixin.java`：外层 tick 的 drain，不扩展到通用 pollTask。 +- 已有 ServerManager、LevelTickLoop、Sable 桥、组装兼容、停止入口：接入阶段生命周期。 +- `tests/PayloadWorkQueueRegression.java`：纯 Java 队列测试。 +- `tests/payload/PayloadRegression.java`：实际 NeoForge 上下文与多维度集成。 +- `tests/payload/SimulatedHandleRegression.java`：可选真实模组处理器及约束清理测试。 + +## Task 1: 可测试的有界队列 + +**Files:** 创建 `src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java`、`tests/PayloadWorkQueueRegression.java`；修改 `build.gradle`。 + +**Interfaces:** `PayloadWorkQueue(LongSupplier nanoTime)`；` CompletableFuture submit(K connection, Supplier action)`；`int drainBatch()`；`void close(Throwable reason)`；`int size()`。拒绝返回异常 future；执行前已取消的任务不产生副作用。锁只保护队列及计数，绝不包住 action 或 completion 回调。 + +- [ ] 写 `fifoAndExactlyOnce`、`capacityAndCancellation`、`batchCountAndTime`、`closeAndReentrantCompletion` 断言：同连接顺序 0..255；第 257 项拒绝；16 个连接各 256 项填满后第 4097 项拒绝；取消使容量回收；批次最多 64；假时钟到 2_000_000 ns 后不再启动下一项；慢 action 不被中断；close 后旧 future 全结束且不能再接收。 +- [ ] 添加 `runPayloadWorkQueueRegression` 到 check；先运行并保存失败证据，不能以无测试发现作为 RED。 +- [ ] 实现上述接口；future 完成回调可重新 submit/close 而无锁死；Supplier 异常只影响本项。 +- [ ] 运行 `./gradlew.bat runPayloadWorkQueueRegression`，输出 `PAYLOAD QUEUE PASS`；并行提交/取消压力测试核对计数为零。 +- [ ] 审阅并提交本任务文件。 + +## Task 2: 安全阶段与循环生命周期 + +**Files:** 创建 `src/main/java/dev/modzuozhi/core/compat/MainThreadPhaseCoordinator.java`、`tests/payload/PayloadRegression.java`；修改 `core/dimthread/ServerManager.java`、`core/dimthread/LevelTickLoop.java`、`core/compat/SableMainThreadBridge.java`（均位于 `src/main/java/dev/modzuozhi/`），修改 `build.gradle`。 + +**Interfaces:** coordinator 的 `void run(MinecraftServer server, String reason, Runnable action)`、`boolean owns(MinecraftServer server)`、`boolean blocksWorkerStart(MinecraftServer server)`、`void stopping(MinecraftServer server)`、`void clear(MinecraftServer server)`；Sable 桥新增 `boolean inHandoff(MinecraftServer server)`。ServerManager.runMainThreadPhase 委托 coordinator，保持现有 CC 调用不变。 + +- [ ] 写 `exclusivePhase`、`nestedPhase`、`handoffReentryRejected`、`newLoopDuringPhase`、`stopDuringPhase`：处理期间全部世界 owner 为真主线程、worker 写入计数不增加；等待可完成 Sable 回调；独占内嵌套只建一次屏障；handoff 中建全局屏障立即明确失败；新 loop 在阶段退出前不能开跑；预先暂停、菜单暂停、停止状态不被 finally 恢复。 +- [ ] `./gradlew.bat runServer -PpayloadRegression` 得到预期断言失败，保存日志；夹具另用超时终止测试而非让生产代码越过屏障。 +- [ ] 实现 IDLE/DRAINING/EXCLUSIVE/RESTORING/STOPPING 状态、租约与入口检查。循环注册、scheduleIfNeeded、resume 共享阶段启动门禁；新 loop 参与退出处理。等待不持锁，继续 pump 必要回调；嵌套 drain 不递归。循环已经在途时等待其子任务及线程字段全部归还。 +- [ ] 运行夹具输出 `PAYLOAD PHASE PASS`；复跑 ownershipRegression，确认 CC 预置暂停及 Sable 回调仍通过。 +- [ ] 审阅并提交本任务文件。 + +## Task 3: NeoForge 公共入口与 future 语义 + +**Files:** 创建 `src/main/java/dev/modzuozhi/core/compat/PayloadMainThreadDispatch.java`、`src/main/java/dev/modzuozhi/mixin/ServerPayloadContextMixin.java`、`src/main/java/dev/modzuozhi/mixin/MinecraftServerPayloadMixin.java`；修改 `tests/payload/PayloadRegression.java`、`src/main/resources/dimensionalripper.mixins.json`。 + +**Interfaces:** dispatch 的 ` CompletableFuture submit(ServerPlayer player, ResourceLocation payloadId, Supplier action)`、`boolean owns(MinecraftServer server)`、`void drain(MinecraftServer server)`、`void stop(MinecraftServer server)`。Runnable 适配 Supplier。任务保存实际 connection、player、接收时 ServerLevel 身份及时间；各服务器实例独立。 + +- [ ] 用真实 ServerPayloadContext 的 Runnable/Supplier 路径写 `mainPayloadSeesBlockEntity`、`futureContract`、`phaseEscapeAndStalePlayer`、`configurationPassthrough`。断言并发维度下能取得原 BE、true main、至多一次；Supplier 返回 42；异常、取消和 guard 的对外结果与未修改 NeoForge 一致，内部真实失败仍可见；权限拒绝未发生世界变更；配置阶段不调用 player()；网络计算原线程不变。 +- [ ] 运行 payloadRegression 记录原入口 wrong-owner/null 的失败；记录尚未覆盖的真实 Veil 路径，不冒充实测。 +- [ ] 两个 enqueueWork 仅拦截游戏玩家的显式主线程工作。网络/worker 投递不等待；独占内嵌套立即完成；普通主线程外层进入安全阶段时先遵守同连接已排队顺序；DRAINING 中普通投递只入队；同步 handoff 重入按任务 2 契约明确失败。原 handler、上下文与 NetworkRegistry.guard 保留，不双重执行。 +- [ ] 在 MinecraftServer.tickServer HEAD 的外层主线程 drain 一批；不在 pollTask 或 worker 回调内 drain。执行前验证连接、player 身份及世界；失效 future 明确结束。无队列不建屏障。 +- [ ] 加 `workerWaitCycleDetected` 测试：worker 等待队列 future 时测试必须识别无法收口且取消夹具，不误报自动兼容。运行集成输出 `PAYLOAD ENTRY PASS`，覆盖没有并行 loop 的原版情况。 +- [ ] 审阅并提交本任务文件。 + +## Task 4: 原模组交互与清理路径 + +**Files:** 修改 `src/main/java/dev/modzuozhi/core/compat/SableAssemblyDispatch.java`、`src/main/java/dev/modzuozhi/mixin/CreateConfigurationPacketMixin.java`、`src/main/java/dev/modzuozhi/mixin/SimulatedAssemblePacketMixin.java`；创建 `tests/payload/SimulatedHandleRegression.java`；修改 `build.gradle`。若真实路径证明需要专用适配，仅新增 `src/main/java/dev/modzuozhi/mixin/JadeRequestDispatchMixin.java` 与 `src/main/java/dev/modzuozhi/mixin/SimulatedHandleLifecycleMixin.java`，并登记可选 mixin。 + +**Interfaces:** 保持 `SableAssemblyDispatch.dispatch(ServerPlayer,Runnable)`，改为复用公共阶段/投递，禁止再送进已暂停的维度队列。具体可选模组方法描述符从本地实际 jar 审计后锁定，不能猜注入目标。 + +- [ ] 先记录真实 `UpdatePlayerUsingHandlePacket.handle` 的抓取、距离、松手入口、BE 身份、HandleConstraint 移除结果，建立 RED。核对 HandleBlockEntity.checkPlayers 及断线/死亡/卸载清理，不仅测试模拟 Runnable。 +- [ ] 审计 Veil 注册和 Jade 二次 server.execute；建立逃逸测试。只有已观察的逃逸点使用定点包装，不改写全部 MinecraftServer.execute。旧 Create/组装入口在公共独占内直接调用原 handler；阶段外走相同队列。 +- [ ] 测试顺序与副作用：抓取一次、更新距离一次、松手一次，最后 constraint 不存在；两个玩家不会互相释放；转速配置、组装拆解保留原权限和距离行为，错误坐标不强加载。 +- [ ] 覆盖断线、换维度、死亡、把手拆除、sublevel 卸载和单连接 256 满载后松手被拒绝；若原清理不足，可信生命周期适配在旧世界安全阶段释放资源，不重放过期客户端包。过载断开连接调用原 disconnect/cleanup，日志限频。 +- [ ] 真实运行输出 `SIMULATED HANDLE PASS`、`CREATE CONFIG PASS`、`JADE REQUEST PASS`。原模组依赖或原生物理无法启动时，记录未通过项并暂停“把手已修复”的交付结论，不能用假夹具替代。 +- [ ] 审阅并提交已验证的适配文件；无必要不创建专用 mixin。 + +## Task 5: 停止、可观测性与限频 + +**Files:** 创建 `src/main/java/dev/modzuozhi/core/compat/PayloadDiagnostics.java`；修改 `src/main/java/dev/modzuozhi/core/compat/PayloadMainThreadDispatch.java`、`src/main/java/dev/modzuozhi/mixin/MinecraftServerDimMixin.java`、`src/main/java/dev/modzuozhi/core/compat/ThreadOwnershipDiagnostics.java`、`src/main/java/dev/modzuozhi/command/MasterCommand.java`、`tests/payload/PayloadRegression.java`。 + +**Interfaces:** `PayloadDiagnostics.snapshot(MinecraftServer server): String`，只读且不新建循环；dispatch.stop 拒绝接收、终结待执行 future，阶段退出不恢复停止中的 loop；clear 在清理结束移除服务器引用。 + +- [ ] 写 `boundedDiagnostics`、`stopRejectsAndReleases`、`twoServersIsolated`：1000 个不同 ID 最多 128 个具名项加 other；1000 次失败不输出1000份堆栈；停止后无遗留 future/连接/世界引用；某一服务器 stop 不影响另一服务器。 +- [ ] 运行得到 RED。实现每服务器计数、队列最老等待、最近 32 事件、等待/执行耗时、批次数；异常全局 30 秒限频，快照无包正文。正常/异常停止入口在既有循环停止与保存之前关闭新队列；不要把旧 30 秒停服风险说成已解决。 +- [ ] status 和 crash report 附快照，区分等待 worker、handler 慢和 Sable handoff；测试所有失败 future 最终完成且 handler 从不重复执行。 +- [ ] 运行 payloadRegression 输出 `PAYLOAD LIFECYCLE PASS`、`PAYLOAD DIAGNOSTICS PASS`，审阅并提交。 + +## Task 6: 兼容验收与候选包 + +**Files:** 修改 `README.md`、`gradle.properties`（候选版本 1.1.17）；证据 `build/reports/payload/`。 + +**Interfaces:** 交付明确列出“源码/字节码审计”“夹具通过”“真实模组通过”的覆盖表，不将它们合并为兼容保证。 + +- [ ] 分别运行 `./gradlew.bat check`、`runServer -PpayloadRegression`、`runServer -PownershipRegression -PccTestJar=<已验证本地CC路径>`、`runServer -PinteractionRegression`。逐项确认纯 Java、CC、区块、水流、实体路由和新 PASS 标记，不把预期故障夹具当成功事故。 +- [ ] 无 Create/Veil/Jade/Simulated 可选模组启动通过；真实模组组合另跑，不自动复制到用户 mods 目录。 +- [ ] 对比无交互、普通交互、密集抓握、多人查询及慢维度：主线程 TPS、维度完成 TPS、队列延迟、阶段耗时。无交互新增屏障数必须为零；报告慢维度代价，不以 Carpet MSPT 独立判定。 +- [ ] 正常构建 `./gradlew.bat clean build`，不带任何夹具属性。检查 jar 无 tests/Regression/测试模组类、元数据 1.1.17，记录 SHA256。 +- [ ] 按 requesting-code-review 技能审查完整差异，修复阻塞问题并重跑受影响测试。README 清楚标注 ServerWarashi 仍未由本计划修复以及真实航空学验收状态；提交本任务。 +- [ ] 只交付本地候选 jar 和覆盖/风险说明；如真实物理未通过，交付测试候选而非宣称全部修好。不推远端、不部署。 + +## 自审结果 + +设计第 4 节对应任务 1–3；旧适配、Veil/Jade 与生命周期对应任务 4–5；有界诊断与性能/打包对应任务 5–6。所有五个 Review Focus 已映射到断言。ServerWarashi 明确作为独立后续工作，不以当前方案替代采样栈修复。 diff --git a/docs/superpowers/specs/2026-10-06-payload-ownership-design.md b/docs/superpowers/specs/2026-10-06-payload-ownership-design.md index dc86935..6877451 100644 --- a/docs/superpowers/specs/2026-10-06-payload-ownership-design.md +++ b/docs/superpowers/specs/2026-10-06-payload-ownership-design.md @@ -2,7 +2,7 @@ 日期：2026-10-06（Asia/Shanghai） -状态：用户已同意“共用入口、正确性优先”的方向；本文待审阅，尚未进入实施。 +状态：用户于 2026-10-06 批准直接实施；公共入口已实现并进行候选验证。完整验收尚有未测项，见 `docs/payload-ownership-1.1.17-validation.md`，不得把本文的目标清单当作全部通过。 ## 1. 用户目标与边界 diff --git a/gradle.properties b/gradle.properties index 7c2b53e..2e93d86 100644 --- a/gradle.properties +++ b/gradle.properties @@ -36,7 +36,7 @@ mod_name=Dimensional Ripper # The license of the mod. Review your options at https://choosealicense.com/. All Rights Reserved is the default. mod_license=MIT # The mod version. See https://semver.org/ -mod_version=1.1.15 +mod_version=1.1.17 # The group ID for the mod. It is only important when publishing build artifacts to Maven. mod_group_id=dev.modzuozhi # The authors of the mod. This is a simple text string that is used for display purposes in the mod list. diff --git a/src/main/java/dev/modzuozhi/command/MasterCommand.java b/src/main/java/dev/modzuozhi/command/MasterCommand.java index 6aeb28c..e5ab09d 100644 --- a/src/main/java/dev/modzuozhi/command/MasterCommand.java +++ b/src/main/java/dev/modzuozhi/command/MasterCommand.java @@ -166,6 +166,10 @@ public final class MasterCommand { dimensionName(dimension), EntityTickParallel.missedShards(), ShardGate.teMissed(), ShardGate.chunkMissed()); ctx.getSource().sendSuccess(() -> Component.literal(msg), false); + ctx.getSource().sendSuccess(() -> Component.literal("线程交接：" + + dev.modzuozhi.core.compat.ThreadOwnershipDiagnostics.timing()), false); + ctx.getSource().sendSuccess(() -> Component.literal( + dev.modzuozhi.core.compat.PayloadDiagnostics.snapshot(ctx.getSource().getServer())), false); var loops = DimThreadCore.MANAGER.loopsSnapshot(); if (loops.isEmpty()) { ctx.getSource().sendSuccess(() -> Component.literal("维度循环：无数据（尚未启动或已停止），不按 20 TPS 显示。"), false); diff --git a/src/main/java/dev/modzuozhi/core/compat/MainThreadPhaseCoordinator.java b/src/main/java/dev/modzuozhi/core/compat/MainThreadPhaseCoordinator.java new file mode 100644 index 0000000..8c8e3ed --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/MainThreadPhaseCoordinator.java @@ -0,0 +1,85 @@ +package dev.modzuozhi.core.compat; + +import dev.modzuozhi.core.dimthread.DimThreadCore; +import dev.modzuozhi.core.dimthread.LevelTickLoop; +import net.minecraft.server.MinecraftServer; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Supplier; + +/** Separate compatibility lease; ordinary pause/resume cannot release it. */ +public final class MainThreadPhaseCoordinator { + private enum Phase { IDLE, DRAINING, EXCLUSIVE, RESTORING, STOPPING } + private static final class State { volatile Phase phase = Phase.IDLE; } + private static final ConcurrentHashMap STATES = new ConcurrentHashMap<>(); + private MainThreadPhaseCoordinator() {} + private static State state(MinecraftServer server) { return STATES.computeIfAbsent(server, s -> new State()); } + + public static boolean owns(MinecraftServer server) { + State s = STATES.get(server); + return server.isSameThread() && s != null && s.phase == Phase.EXCLUSIVE; + } + public static boolean blocksWorkerStart(MinecraftServer server) { + State s = STATES.get(server); + return s != null && s.phase != Phase.IDLE; + } + public static String phase(MinecraftServer server) { + State s = STATES.get(server); return s == null ? "IDLE" : s.phase.name(); + } + /** Serializes loop registration with taking/releasing all leases; never waits for workers. */ + public static LevelTickLoop register(MinecraftServer server, Supplier registration) { + State s = state(server); + synchronized (s) { + LevelTickLoop loop = registration.get(); + if (s.phase != Phase.IDLE || PayloadMainThreadDispatch.closed(server)) loop.holdForMainThread(true); + return loop; + } + } + /** Pause changes and loop creation use the same short lock; never hold it while draining. */ + public static void updatePause(MinecraftServer server, Runnable change) { + synchronized(state(server)){change.run();} + } + public static void run(MinecraftServer server, String reason, Runnable action) { + if (!server.isSameThread()) throw new IllegalStateException("Compatibility phase requires server thread"); + State s = state(server); + if (owns(server)) { action.run(); return; } + if (SableMainThreadBridge.inHandoff(server)) + throw new IllegalStateException("Global phase inside synchronous Sable handoff would deadlock: " + reason); + synchronized (s) { + if (s.phase != Phase.IDLE) throw new IllegalStateException("Reentrant global phase during " + s.phase); + s.phase = Phase.DRAINING; + loops(server).forEach(loop -> loop.holdForMainThread(true)); + } + long began = System.nanoTime(), drained = began; + try { + server.managedBlock(() -> { + boolean idle = loops(server).stream().noneMatch(LevelTickLoop::isTicking); + if (!idle) ThreadOwnershipDiagnostics.slowPhase(server, reason, began); + return idle; + }); + synchronized (s) { + if (s.phase == Phase.STOPPING) throw new IllegalStateException("Server stopping during compatibility phase"); + s.phase = Phase.EXCLUSIVE; + } + drained = System.nanoTime(); + action.run(); + } finally { + ThreadOwnershipDiagnostics.phaseTiming(reason, began, drained); + synchronized (s) { + if (s.phase != Phase.STOPPING) { + s.phase = Phase.RESTORING; + loops(server).forEach(loop -> loop.holdForMainThread(false)); + s.phase = Phase.IDLE; + } + } + } + } + private static java.util.List loops(MinecraftServer server) { + return DimThreadCore.MANAGER.ownershipLoopsSnapshot().stream() + .filter(loop -> loop.getLevel().getServer() == server).toList(); + } + public static void stopping(MinecraftServer server) { + State s = state(server); + synchronized (s) { s.phase = Phase.STOPPING; loops(server).forEach(loop -> loop.holdForMainThread(true)); } + } + public static void clear(MinecraftServer server) { STATES.remove(server); } +} diff --git a/src/main/java/dev/modzuozhi/core/compat/PayloadDiagnostics.java b/src/main/java/dev/modzuozhi/core/compat/PayloadDiagnostics.java new file mode 100644 index 0000000..6e84556 --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/PayloadDiagnostics.java @@ -0,0 +1,76 @@ +package dev.modzuozhi.core.compat; + +import dev.modzuozhi.ModZuozhi; +import net.minecraft.server.MinecraftServer; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +/** Bounded metadata only; never records payload bodies or exception messages containing user data. */ +public final class PayloadDiagnostics { + private static final ConcurrentHashMap STATS = new ConcurrentHashMap<>(); + private static final AtomicLong NEXT_WARNING = new AtomicLong(); + public record Ticket(String label, long started) {} + private static final class Stats { + final Map ids = new LinkedHashMap<>(); + // received, completed, handler failure, cancelled, expired, capacity rejected + final long[] other = new long[6]; + final Map pending = new IdentityHashMap<>(); + final ArrayDeque recent = new ArrayDeque<>(); + long batches; double waitMs, actionMs; + } + private PayloadDiagnostics() {} + public static Ticket begin(MinecraftServer server, String id) { + synchronized(PayloadMainThreadDispatch.LIFECYCLE) { + if(PayloadMainThreadDispatch.closed(server))return null; + Stats s = STATS.computeIfAbsent(server, key -> new Stats()); + synchronized (s) { + String label = s.ids.containsKey(id) || s.ids.size()<128 ? id : "other"; + long[] values = label.equals("other") ? s.other : s.ids.computeIfAbsent(label, k -> new long[6]); + values[0]++; + Ticket ticket = new Ticket(label,System.nanoTime()); s.pending.put(ticket,Boolean.TRUE); return ticket; + } + } + } + public static void end(MinecraftServer server, Ticket ticket, Throwable error) { + if(ticket==null)return; + Stats s=STATS.get(server); if(s==null) return; + synchronized(s) { + if(s.pending.remove(ticket)==null) return; + long[] values=ticket.label.equals("other") ? s.other : s.ids.get(ticket.label); + int outcome = error == null ? 1 + : PayloadMainThreadDispatch.expired(error) ? 4 + : error instanceof PayloadWorkQueue.CapacityExceeded ? 5 + : error instanceof java.util.concurrent.CancellationException ? 3 : 2; + values[outcome]++; + if(error!=null) { + if(s.recent.size()==32) s.recent.removeFirst(); + s.recent.addLast(ticket.label+":"+error.getClass().getSimpleName()); + } + } + if(error!=null) warn(ticket.label,error); + } + private static void warn(String id, Throwable error) { + long now=System.nanoTime(),next=NEXT_WARNING.get(); + if((next==0 || now-next>=0) && NEXT_WARNING.compareAndSet(next,now+TimeUnit.SECONDS.toNanos(30))) + ModZuozhi.LOGGER.warn("[PayloadOwnership] Task {} ended with {}; further failures aggregated for 30s",id,error.getClass().getSimpleName()); + } + public static void batch(MinecraftServer server,long start,long acquired) { + Stats s=STATS.get(server);if(s==null)return; + synchronized(s){s.batches++;s.waitMs=(acquired-start)/1_000_000.0;s.actionMs=(System.nanoTime()-acquired)/1_000_000.0;} + } + public static String snapshot(MinecraftServer server) { + Stats s=STATS.get(server);if(s==null)return "payloads: pending=0 batches=0"; + synchronized(s) { + long oldest=s.pending.keySet().stream().mapToLong(Ticket::started).min().orElse(System.nanoTime()); + long received=s.other[0],completed=s.other[1],failed=s.other[2],cancelled=s.other[3],expired=s.other[4],rejected=s.other[5]; + for(long[] v:s.ids.values()){received+=v[0];completed+=v[1];failed+=v[2];cancelled+=v[3];expired+=v[4];rejected+=v[5];} + return "payloads: phase="+MainThreadPhaseCoordinator.phase(server)+" received="+received+" completed="+completed + +" failed="+failed+" cancelled="+cancelled+" expired="+expired+" rejected="+rejected + +" pending="+s.pending.size()+" oldestMs="+(System.nanoTime()-oldest)/1_000_000.0 + +" batches="+s.batches+" waitMs="+s.waitMs+" actionMs="+s.actionMs+" ids="+s.ids.size()+" other="+Arrays.toString(s.other)+" recent="+s.recent; + } + } + public static void clear(MinecraftServer server){STATS.remove(server);} +} diff --git a/src/main/java/dev/modzuozhi/core/compat/PayloadMainThreadDispatch.java b/src/main/java/dev/modzuozhi/core/compat/PayloadMainThreadDispatch.java new file mode 100644 index 0000000..1b61d1b --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/PayloadMainThreadDispatch.java @@ -0,0 +1,114 @@ +package dev.modzuozhi.core.compat; + +import net.minecraft.network.Connection; +import net.minecraft.resources.ResourceLocation; +import net.minecraft.server.MinecraftServer; +import net.minecraft.server.level.ServerPlayer; +import java.util.concurrent.*; +import java.util.function.Supplier; + +/** Shared game payload handoff. No Minecraft world access is authorized merely by thread name. */ +public final class PayloadMainThreadDispatch { + static final Object LIFECYCLE = new Object(); + private static final class StalePayload extends CancellationException { + StalePayload(String reason){super(reason);} + } + private static final ConcurrentHashMap> QUEUES = new ConcurrentHashMap<>(); + private static final java.util.Map STOPPED = new java.util.WeakHashMap<>(); + private static final ConcurrentHashMap> OVERLOADED = new ConcurrentHashMap<>(); + private PayloadMainThreadDispatch() {} + + public static boolean owns(MinecraftServer server) { return MainThreadPhaseCoordinator.owns(server); } + static boolean closed(MinecraftServer server) { synchronized(LIFECYCLE){return STOPPED.containsKey(server)||server.isStopped();} } + static boolean expired(Throwable error) { return error instanceof StalePayload; } + + public static CompletableFuture submit(ServerPlayer player, ResourceLocation payloadId, Supplier action) { + var level = player.serverLevel(); + var server = level.getServer(); + var listener = player.connection; + if (listener == null || player.hasDisconnected() || player.isRemoved() || closed(server)) + return CompletableFuture.failedFuture(new StalePayload("Stale payload player")); + if (owns(server)) return CompletableFuture.completedFuture(action.get()); + if (server.isSameThread() && SableMainThreadBridge.inHandoff(server)) + throw new IllegalStateException("Payload " + payloadId + " requested global ownership from synchronous Sable callback"); + PayloadWorkQueue queue; + synchronized(LIFECYCLE) { + if(closed(server)) return CompletableFuture.failedFuture(new StalePayload("Server stopping")); + queue = QUEUES.computeIfAbsent(server, s -> new PayloadWorkQueue<>(System::nanoTime)); + } + var ticket=PayloadDiagnostics.begin(server,payloadId.toString()); + CompletableFuture result = queue.submit(listener.getConnection(), () -> { + if (player.connection != listener || listener.player != player || player.hasDisconnected() + || player.isRemoved() || player.serverLevel() != level) + throw new StalePayload("Payload player/world changed before execution"); + try{return action.get();} + catch(CancellationException handlerFailure){throw new CompletionException(handlerFailure);} + }); + result.whenComplete((value,error) -> { + PayloadDiagnostics.end(server,ticket,error); + if(error instanceof PayloadWorkQueue.CapacityExceeded) synchronized(LIFECYCLE) { + if(!closed(server)) OVERLOADED.computeIfAbsent(server,s -> new ConcurrentHashMap<>()).putIfAbsent(listener.getConnection(),player); + } + }); + // Preserve synchronous main-thread enqueueWork semantics and observed FIFO order. + // Each acquisition still obeys the per-batch budget. Never drain recursively while pumping callbacks. + if (server.isSameThread() && !MainThreadPhaseCoordinator.blocksWorkerStart(server)) { + while (!result.isDone()) drain(server); + try { result.join(); } + catch (CompletionException failed) { + if (failed.getCause() instanceof RuntimeException runtime) throw runtime; + if (failed.getCause() instanceof Error error) throw error; + throw failed; + } + } + return result; + } + public static void drain(MinecraftServer server) { + if (!server.isSameThread()) throw new IllegalStateException("Payload drain requires server thread"); + var queue = QUEUES.get(server); + var overloaded=OVERLOADED.get(server); + if ((queue == null || queue.size() == 0) && (overloaded==null || overloaded.isEmpty())) return; + if (MainThreadPhaseCoordinator.blocksWorkerStart(server) + || SableMainThreadBridge.inHandoff(server)) return; + long start=System.nanoTime(); + MainThreadPhaseCoordinator.run(server, "custom payloads", () -> { + long acquired=System.nanoTime(); + try { + // Cleanup goes through the original disconnect lifecycle with world ownership. + // A cleanup-only batch avoids bypassing the payload batch count/time budget. + if(overloaded!=null && !overloaded.isEmpty()) { + int count=0; + for(var entry:overloaded.entrySet()) { + if(count++>=64 || System.nanoTime()-acquired>=2_000_000L) break; + if(overloaded.remove(entry.getKey(),entry.getValue())) { + var player=entry.getValue(); + if(player.connection!=null && player.connection.getConnection()==entry.getKey() && !player.hasDisconnected()) + player.connection.disconnect(net.minecraft.network.chat.Component.literal("Too many pending mod interactions; please reconnect.")); + } + } + } else if(queue!=null) queue.drainBatch(); + } finally {PayloadDiagnostics.batch(server,start,acquired);} + }); + } + public static void stop(MinecraftServer server) { + PayloadWorkQueue queue; + synchronized(LIFECYCLE){STOPPED.put(server,Boolean.TRUE);queue=QUEUES.get(server);OVERLOADED.remove(server);} + if(queue!=null)queue.close(new StalePayload("Server stopping")); + MainThreadPhaseCoordinator.stopping(server); + } + public static void clear(MinecraftServer server) { + synchronized(LIFECYCLE){QUEUES.remove(server); OVERLOADED.remove(server); PayloadDiagnostics.clear(server);} + MainThreadPhaseCoordinator.clear(server); + } + /** NeoForge guard's null-on-handler-error contract, with bounded diagnostics already attached to raw result. */ + public static CompletableFuture guard(CompletableFuture result) { + CompletableFuture guarded=result.handle((value,error) -> { + // Preserve NeoForge's handler-error result, not false success for queue policy failures. + if(error instanceof StalePayload || error instanceof PayloadWorkQueue.CapacityExceeded || result.isCancelled()) + throw new CompletionException(error); + return error==null ? value : null; + }); + guarded.whenComplete((value,error) -> {if(guarded.isCancelled())result.cancel(false);}); + return guarded; + } +} diff --git a/src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java b/src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java index 3df67a2..c284a37 100644 --- a/src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java +++ b/src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java @@ -8,6 +8,9 @@ import java.util.function.Supplier; /** Bounded FIFO. Queue locks never encompass handlers or future completion callbacks. */ public final class PayloadWorkQueue { + public static final class CapacityExceeded extends RejectedExecutionException { + private CapacityExceeded() { super("Payload queue capacity exceeded"); } + } private final LongSupplier clock; private final LinkedHashSet> pending = new LinkedHashSet<>(); private final Map counts = new HashMap<>(); @@ -24,7 +27,7 @@ public final class PayloadWorkQueue { synchronized (this) { rejected = closed; if (rejected == null && (pending.size() >= 4096 || counts.getOrDefault(connection, 0) >= 256)) - rejected = new RejectedExecutionException("Payload queue capacity exceeded"); + rejected = new CapacityExceeded(); if (rejected == null) { pending.add(entry); counts.merge(connection, 1, Integer::sum); diff --git a/src/main/java/dev/modzuozhi/core/compat/SableAssemblyDispatch.java b/src/main/java/dev/modzuozhi/core/compat/SableAssemblyDispatch.java index 8835616..f7a00d7 100644 --- a/src/main/java/dev/modzuozhi/core/compat/SableAssemblyDispatch.java +++ b/src/main/java/dev/modzuozhi/core/compat/SableAssemblyDispatch.java @@ -1,38 +1,15 @@ package dev.modzuozhi.core.compat; -import dev.modzuozhi.ModZuozhi; -import dev.modzuozhi.core.dimthread.DimThreadCore; -import dev.modzuozhi.core.dimthread.IPlayerPacketQueue; -import net.minecraft.server.level.ServerLevel; +import net.minecraft.resources.ResourceLocation; import net.minecraft.server.level.ServerPlayer; -/** Main -> dimension safe point -> exclusive main-thread phase, without blocking main on dimension. */ +/** Legacy Create/assembly callers share the generic payload phase; no second dimension queue. */ public final class SableAssemblyDispatch { private SableAssemblyDispatch() {} public static void dispatch(ServerPlayer player, Runnable action) { - ServerLevel level = player.serverLevel(); - var server = level.getServer(); - if (!server.isSameThread()) { - server.execute(() -> dispatch(player, action)); - return; - } if (player.hasDisconnected() || player.isRemoved()) return; - if (!DimThreadCore.MANAGER.isLoopScheduled(level) || SableMainThreadBridge.ownsHandoff(level)) { - action.run(); - return; - } - // Return immediately: waiting here would deadlock against an in-flight Sable tick. - ((IPlayerPacketQueue) level).modzuozhi$playerPacketQueue().offer(() -> - SableMainThreadBridge.runForLevel(level, () -> { - // A queued interaction must not act on a different dimension after teleport/logout. - if (player.hasDisconnected() || player.isRemoved() || player.serverLevel() != level) return; - try { - action.run(); - } catch (RuntimeException failure) { - ModZuozhi.LOGGER.error("[Compatibility] 航空学组装/拆解请求执行失败", failure); - throw failure; - } - })); + PayloadMainThreadDispatch.submit(player, ResourceLocation.fromNamespaceAndPath("dimensionalripper", "legacy_interaction"), + () -> { action.run(); return null; }); } } diff --git a/src/main/java/dev/modzuozhi/core/compat/SableMainThreadBridge.java b/src/main/java/dev/modzuozhi/core/compat/SableMainThreadBridge.java index 81044cc..b4c32bc 100644 --- a/src/main/java/dev/modzuozhi/core/compat/SableMainThreadBridge.java +++ b/src/main/java/dev/modzuozhi/core/compat/SableMainThreadBridge.java @@ -32,6 +32,10 @@ public final class SableMainThreadBridge { public static boolean ownsHandoff(ServerLevel level) { return HANDED_OFF_LEVEL.get() == level; } + public static boolean inHandoff(MinecraftServer server) { + ServerLevel level = HANDED_OFF_LEVEL.get(); + return level != null && level.getServer() == server; + } private static final ConcurrentHashMap, Method> GET_LEVEL_METHODS = new ConcurrentHashMap<>(); private SableMainThreadBridge() { @@ -72,12 +76,14 @@ public final class SableMainThreadBridge { CompletableFuture completion = new CompletableFuture<>(); // 0 = queued, 1 = running, 2 = cancelled before start. AtomicInteger state = new AtomicInteger(); + ThreadOwnershipDiagnostics.handoff(level, "queued"); server.execute(() -> { // A timed-out queued task must not execute later, detached from its waiting level tick. if (!state.compareAndSet(0, 1)) { return; } try { + ThreadOwnershipDiagnostics.handoff(level, "running"); // The parent dimension worker temporarily owns both vanilla thread fields while // it is paused here. Restore real server-thread ownership for the complete Sable // segment, then hand ownership back before waking the worker. @@ -88,8 +94,10 @@ public final class SableMainThreadBridge { } finally { if (previous == null) HANDED_OFF_LEVEL.remove(); else HANDED_OFF_LEVEL.set(previous); } + ThreadOwnershipDiagnostics.handoff(level, "completed-owner-restored"); completion.complete(null); } catch (Throwable throwable) { + ThreadOwnershipDiagnostics.handoff(level, "failed-owner-restored: " + throwable.getClass().getSimpleName()); completion.completeExceptionally(throwable); } }); @@ -98,6 +106,7 @@ public final class SableMainThreadBridge { completion.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); } catch (InterruptedException interrupted) { if (state.compareAndSet(0, 2)) { + ThreadOwnershipDiagnostics.handoff(level, "cancelled-interrupted-before-start"); completion.cancel(false); } else { // A running main-thread phase must finish before world ownership is released. @@ -107,6 +116,7 @@ public final class SableMainThreadBridge { throw new IllegalStateException("Interrupted while waiting for Sable main-thread physics tick", interrupted); } catch (TimeoutException timeout) { if (state.compareAndSet(0, 2)) { + ThreadOwnershipDiagnostics.handoff(level, "cancelled-timeout-before-start"); completion.cancel(false); ModZuozhi.LOGGER.error( "[Compatibility] Sable 物理 Tick 排队超过 {} 秒且尚未开始，交由故障保护降级", @@ -119,6 +129,7 @@ public final class SableMainThreadBridge { ModZuozhi.LOGGER.warn( "[Compatibility] Sable 主线程物理 Tick 已运行超过 {} 秒；为避免世界并发写入，继续等待其安全结束", TIMEOUT_SECONDS); + ThreadOwnershipDiagnostics.handoff(level, "running-over-10s"); awaitRunningTick(completion); } catch (ExecutionException failed) { Throwable cause = failed.getCause(); diff --git a/src/main/java/dev/modzuozhi/core/compat/ThreadOwnershipDiagnostics.java b/src/main/java/dev/modzuozhi/core/compat/ThreadOwnershipDiagnostics.java new file mode 100644 index 0000000..0a2eaab --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/ThreadOwnershipDiagnostics.java @@ -0,0 +1,101 @@ +package dev.modzuozhi.core.compat; + +import dev.modzuozhi.ModZuozhi; +import dev.modzuozhi.core.dimthread.DimThreadCore; +import dev.modzuozhi.core.dimthread.IMutableMainThread; +import net.minecraft.core.BlockPos; +import net.minecraft.server.MinecraftServer; +import net.minecraft.server.level.ServerLevel; +import java.util.ArrayDeque; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +/** Bounded, read-only evidence: never load chunks or query live block-entity collections here. */ +public final class ThreadOwnershipDiagnostics { + private static final ConcurrentHashMap TRACES = new ConcurrentHashMap<>(); + private static final AtomicLong NEXT_SLOW_WARNING = new AtomicLong(); + private static volatile String lastTiming = "no CC ownership phase sampled"; + private ThreadOwnershipDiagnostics() {} + + private static final class Trace { + String phase = "unobserved"; + String handoff = "none"; + final ArrayDeque recent = new ArrayDeque<>(); + long rejected; + long nextWarning; + String lastRejected = "none"; + void add(String entry) { + if (recent.size() == 16) recent.removeFirst(); + recent.addLast(System.nanoTime() + "ns " + entry); + } + } + + public static void phase(ServerLevel level, String phase) { + Trace trace = TRACES.computeIfAbsent(level, key -> new Trace()); + synchronized (trace) { trace.phase = phase; trace.add(phase + " caller=" + thread(Thread.currentThread())); } + } + + public static void handoff(ServerLevel level, String state) { + Trace trace = TRACES.computeIfAbsent(level, key -> new Trace()); + synchronized (trace) { trace.handoff = state; trace.add("handoff=" + state + " caller=" + thread(Thread.currentThread())); } + } + + public static void rejectedLookup(ServerLevel level, BlockPos pos, Thread observedOwner) { + Trace trace = TRACES.computeIfAbsent(level, key -> new Trace()); + long now = System.nanoTime(); + synchronized (trace) { + trace.rejected++; + if (trace.nextWarning != 0 && now - trace.nextWarning < 0) return; + trace.nextWarning = now + TimeUnit.SECONDS.toNanos(30); + String callers = StackWalker.getInstance().walk(frames -> frames + .filter(frame -> !frame.getClassName().equals(ThreadOwnershipDiagnostics.class.getName())) + .limit(8).map(Object::toString).collect(java.util.stream.Collectors.joining(" <- "))); + trace.lastRejected = "pos=" + pos.toShortString() + " caller=" + thread(Thread.currentThread()) + + " observedOwner=" + thread(observedOwner) + " at=" + callers; + trace.add("rejected " + trace.lastRejected); + ModZuozhi.LOGGER.warn("[ThreadOwnership] getBlockEntity owner mismatch at entry (guard may return null): {}", describe(level)); + } + } + + public static String describe(ServerLevel level) { + Trace trace = TRACES.get(level); + String owners = "dimension=" + level.dimension().location() + + " levelOwner=" + thread(((IMutableMainThread) level).dimThreads$getMainThread()) + + " chunkOwner=" + thread(((IMutableMainThread) level.getChunkSource()).dimThreads$getMainThread()); + if (trace == null) return owners + " phase=unobserved"; + synchronized (trace) { + return owners + " phase=" + trace.phase + " handoff=" + trace.handoff + + " rejected=" + trace.rejected + " lastRejected={" + trace.lastRejected + "} recent=" + trace.recent; + } + } + + public static String report(MinecraftServer server) { + StringBuilder out = new StringBuilder("Snapshot (not an atomic world view); observer=") + .append(thread(Thread.currentThread())).append(';').append(lastTiming); + DimThreadCore.MANAGER.loopsSnapshot().forEach((level, loop) -> { + if (level.getServer() == server) out.append('\n').append(describe(level)) + .append(" ticking=").append(loop.isTicking()).append(" paused=").append(loop.isPaused()) + .append(" running=").append(loop.isRunning()); + }); + return out.append('\n').append(PayloadDiagnostics.snapshot(server)).toString(); + } + + public static void slowPhase(MinecraftServer server, String reason, long began) { + long now = System.nanoTime(); + if (now - began < TimeUnit.SECONDS.toNanos(1)) return; + long next = NEXT_SLOW_WARNING.get(); + if ((next == 0 || now - next >= 0) && NEXT_SLOW_WARNING.compareAndSet(next, now + TimeUnit.SECONDS.toNanos(30))) + ModZuozhi.LOGGER.warn("[ThreadOwnership] {} waiting {} ms; {}", reason, (now-began)/1_000_000, report(server)); + } + + public static void phaseTiming(String reason, long began, long drained) { + lastTiming = reason + " lastWaitMs=" + (drained-began)/1_000_000.0 + + " lastActionMs=" + (System.nanoTime()-drained)/1_000_000.0; + } + public static String timing() { return lastTiming; } + public static String thread(Thread thread) { + return thread == null ? "null" : thread.getName() + "#" + thread.threadId() + '/' + thread.getState(); + } + public static void clear() { TRACES.clear(); NEXT_SLOW_WARNING.set(0); lastTiming = "no CC ownership phase sampled"; } +} diff --git a/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java b/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java index 78ed125..8f86443 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java @@ -2,6 +2,7 @@ package dev.modzuozhi.core.dimthread; import com.mojang.logging.LogUtils; import dev.modzuozhi.ModZuozhi; +import dev.modzuozhi.core.compat.ThreadOwnershipDiagnostics; import net.minecraft.network.Connection; import net.minecraft.network.PacketSendListener; import net.minecraft.network.chat.Component; @@ -62,6 +63,7 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu private volatile boolean ticking = false; /** 单机内嵌服务器暂停（ESC 菜单）时置 true：取消下一拍并不再自排，杜绝暂停/保存期间并行写区块。 */ private volatile boolean paused = false; + private volatile boolean mainThreadHeld; private long scheduleGeneration; private final dev.modzuozhi.core.compat.TickProgress progress = new dev.modzuozhi.core.compat.TickProgress(); @@ -106,7 +108,7 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu /** 首次启动（或外部主动重排）本循环。 */ public synchronized void schedule() { - if (!this.running || this.paused || this.ticking) return; + if (!this.running || this.paused || this.mainThreadHeld || this.ticking) return; this.scheduleAfter(0L); } @@ -128,7 +130,16 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu } public boolean isPaused() { - return this.paused; + return this.paused || this.mainThreadHeld; + } + + public synchronized void holdForMainThread(boolean held) { + if (this.mainThreadHeld == held) return; + this.mainThreadHeld = held; + if (held) { + this.scheduleGeneration++; + if (this.currentTickTask != null) this.currentTickTask.cancel(false); + } else if (this.scheduled) this.schedule(); } /** @@ -181,7 +192,7 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu private void runIfCurrent(long generation) { synchronized (this) { - if (generation != this.scheduleGeneration || this.paused || !this.running || this.ticking) return; + if (generation != this.scheduleGeneration || this.paused || this.mainThreadHeld || !this.running || this.ticking) return; this.ticking = true; } DimThreadCore.attach(Thread.currentThread(), this.level); @@ -199,11 +210,14 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu try { DimThreadCore.swapThreadsAndRun(() -> { try { + ThreadOwnershipDiagnostics.phase(this.level, "queued-tasks"); this.processPendingTasks(); + ThreadOwnershipDiagnostics.phase(this.level, "level-tick"); completed[0] = this.internalTick(); } finally { workNanos[0] = System.nanoTime() - tickStart; } + ThreadOwnershipDiagnostics.phase(this.level, "chunk-tasks"); this.drainChunkTasks(deadline); }, this.level, this.level.getChunkSource()); } catch (Throwable t) { @@ -211,10 +225,11 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu ModZuozhi.LOGGER.error("[DimThread] 维度 {} tick 异常", this.level.dimension().location(), t); FaultGuard.onDimTickFailure(t); } finally { + ThreadOwnershipDiagnostics.phase(this.level, "idle"); progress.finish(System.nanoTime(), workNanos[0], completed[0]); synchronized (this) { this.ticking = false; - if (this.running && !this.paused) { + if (this.running && !this.paused && !this.mainThreadHeld) { this.scheduleAfter(Math.max(0L, deadline - System.nanoTime())); } else if (!this.running) { ModZuozhi.LOGGER.info("[DimThread] 维度 {} 独立循环已停止", this.level.dimension().location()); @@ -250,7 +265,9 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu // 区块不加载 → 实体/光照等待区块 → tick 更慢 → 更没有预算排空区块。因此给一个最低保证窗口。 long effectiveDeadline = Math.max(deadline, now + 2_000_000L); // 最低 2ms boolean overBudget = now >= deadline; - while (this.running && System.nanoTime() < effectiveDeadline) { + // A compatibility phase needs completed work, not the remainder of an idle poll budget. + // The in-flight level tick has already finished; leave queued work for its next owner. + while (this.running && !this.paused && !this.mainThreadHeld && System.nanoTime() < effectiveDeadline) { if (chunkSource.pollTask()) { continue; // 有就绪任务，继续排空 } @@ -296,6 +313,8 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu } EventHooks.fireLevelTickPost(this.level, () -> true); + ThreadOwnershipDiagnostics.phase(this.level, "connections"); + if (net.neoforged.fml.ModList.get().isLoaded("starlight_extra")) { // Farm inventory persistence and logout callbacks require the actual server thread. // Keep the whole connection phase together, including disconnection cleanup. diff --git a/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java b/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java index c5df22a..98696dc 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java @@ -38,6 +38,8 @@ public class ServerManager { /** 每维度独立 tick 循环（维度隔离模型，见 {@link LevelTickLoop}）。key 为维度。 */ private final Map loops = new ConcurrentHashMap<>(); + private final java.util.Set retiringLoops = ConcurrentHashMap.newKeySet(); + private final Map serverPauses = new ConcurrentHashMap<>(); /** 承载所有维度 loop 自排程的调度池（{@code schedule(task, nspt)} 需要 Scheduled 能力）。 */ private volatile ScheduledThreadPoolExecutor masterPool; /** 玩家连接当前所属维度（key 为 Connection，value 为维度），供跨维度/登录时幂等迁移连接。 */ @@ -143,7 +145,26 @@ public class ServerManager { /** 获取（懒创建）某维度的独立 tick 循环。 */ public LevelTickLoop getLoop(ServerLevel level) { - return this.loops.computeIfAbsent(level, l -> new LevelTickLoop(l, getMasterPool())); + return dev.modzuozhi.core.compat.MainThreadPhaseCoordinator.register(level.getServer(), + () -> { + for(var retired:retiringLoops) { + if(retired.getLevel()==level && retired.isTicking()) + throw new IllegalStateException("Cannot replace dimension loop before retired tick drains"); + } + return this.loops.computeIfAbsent(level, l -> { + var loop=new LevelTickLoop(l,getMasterPool()); + if(serverPauses.getOrDefault(l.getServer(),false))loop.pause(); + return loop; + }); + }); + } + + /** Includes removed-but-still-running writers; removing a map entry does not return ownership. */ + public java.util.List ownershipLoopsSnapshot() { + retiringLoops.removeIf(loop -> !loop.isTicking()); + var snapshot=new java.util.HashSet<>(this.loops.values()); + snapshot.addAll(retiringLoops); + return java.util.List.copyOf(snapshot); } /** Read-only inventory; observing status must never create or start a loop. */ @@ -151,6 +172,15 @@ public class ServerManager { return Map.copyOf(this.loops); } + /** + * Short main-thread compatibility phase. Never hold a mod lock while draining workers. + * Call from the server tick, not from a synchronous worker-to-main handoff: that caller's + * worker cannot drain until the callback returns. CC TickScheduler is a server-tick hook. + */ + public void runMainThreadPhase(MinecraftServer server, String reason, Runnable action) { + dev.modzuozhi.core.compat.MainThreadPhaseCoordinator.run(server, reason, action); + } + /** * 该维度 loop 是否已启动接管（{@code tickChildren → scheduleIfNeeded} 之后才为 true）。 *

@@ -159,6 +189,7 @@ public class ServerManager { * 来驱动出生点区块生成；直到 loop 接管后才由 {@code ServerChunkCachePollMixin} 跳过主线程 poll。 */ public boolean isLoopScheduled(ServerLevel level) { + if(retiringLoops.stream().anyMatch(loop -> loop.getLevel()==level && loop.isTicking()))return true; LevelTickLoop loop = this.loops.get(level); return loop != null && dev.modzuozhi.core.compat.PauseOwnership.workerOwns( loop.isScheduled(), loop.isPaused(), loop.isTicking()); @@ -227,21 +258,25 @@ public class ServerManager { if (!server.isSameThread()) { throw new IllegalStateException("Pause ownership must be changed on the server thread"); } + dev.modzuozhi.core.compat.MainThreadPhaseCoordinator.updatePause(server,() -> { + serverPauses.put(server,paused); + ownershipLoopsSnapshot().stream().filter(loop -> loop.getLevel().getServer()==server) + .forEach(loop -> {if(paused)loop.pause();else loop.resume();}); + }); if (paused) { - this.loops.values().forEach(LevelTickLoop::pause); // In-flight ticks may be waiting for Sable on this very thread. Pump callbacks // until they return ownership; a timeout must not authorize concurrent saving. - server.managedBlock(() -> this.loops.values().stream().noneMatch(LevelTickLoop::isTicking)); - } else { - this.loops.values().forEach(LevelTickLoop::resume); + server.managedBlock(() -> ownershipLoopsSnapshot().stream().filter(loop -> loop.getLevel().getServer()==server).noneMatch(LevelTickLoop::isTicking)); } } /** 移除并停止某维度的 loop（维度卸载时调用）。 */ public void removeLoop(ServerLevel level) { - LevelTickLoop loop = this.loops.remove(level); + LevelTickLoop loop = this.loops.get(level); if (loop != null) { loop.kill(); + retiringLoops.add(loop); + this.loops.remove(level,loop); } } @@ -260,9 +295,10 @@ public class ServerManager { * 带 30s 看门狗防止异常场景下永久阻塞。 */ public void stopLoopsAndWaitIdle() { - this.loops.values().forEach(LevelTickLoop::kill); + var stopping = ownershipLoopsSnapshot(); + stopping.forEach(LevelTickLoop::kill); long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); - for (LevelTickLoop loop : this.loops.values()) { + for (LevelTickLoop loop : stopping) { while (loop.isTicking()) { if (System.nanoTime() > deadline) { ModZuozhi.LOGGER.warn("[DimThread] 等待维度 {} loop 收口超时（>30s），强制继续", @@ -277,6 +313,7 @@ public class ServerManager { /** 停止所有维度 loop 并关闭调度池（服务器停机 / 全局开关关闭时调用）。 */ public void killAllLoops() { this.loops.values().forEach(LevelTickLoop::kill); + this.loops.values().stream().filter(LevelTickLoop::isTicking).forEach(retiringLoops::add); this.loops.clear(); // 清空连接路由：重新启用时 routeConnection 会把连接当作「首次」重新挂到新 loop。 // 否则 old == level 会误判为 no-op，导致重启后玩家连接丢失、不再被任何 loop tick。 @@ -289,8 +326,11 @@ public class ServerManager { } public void clear() { + dev.modzuozhi.core.compat.ThreadOwnershipDiagnostics.clear(); TpsTracker.reset(); actives.clear(); + serverPauses.clear(); + retiringLoops.removeIf(loop -> !loop.isTicking()); threadPools.clear(); } } diff --git a/src/main/java/dev/modzuozhi/mixin/ComputerCraftTickSchedulerMixin.java b/src/main/java/dev/modzuozhi/mixin/ComputerCraftTickSchedulerMixin.java new file mode 100644 index 0000000..580b38a --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/ComputerCraftTickSchedulerMixin.java @@ -0,0 +1,41 @@ +package dev.modzuozhi.mixin; + +import com.llamalad7.mixinextras.injector.wrapmethod.WrapMethod; +import com.llamalad7.mixinextras.injector.wrapoperation.Operation; +import dev.modzuozhi.core.dimthread.DimThreadCore; +import net.minecraft.server.level.ServerLevel; +import net.minecraft.world.level.chunk.LevelChunk; +import net.neoforged.neoforge.server.ServerLifecycleHooks; +import org.spongepowered.asm.mixin.*; +import java.util.Queue; + +/** CC remains optional. Keep its identity check intact, but run it with exclusive ownership. */ +@Pseudo +@Mixin(targets = "dan200.computercraft.shared.util.TickScheduler", remap = false) +public abstract class ComputerCraftTickSchedulerMixin { + @Shadow @Final private static Queue toTick; + @Unique private static final Object dimensionalripper$schedulerLock = new Object(); + + @WrapMethod(method = "tick()V") + private static void dimensionalripper$safeTick(Operation original) { + // No tokens -> no world access, no global barrier. New tokens can wait until next tick. + if (toTick.isEmpty()) return; + var server = ServerLifecycleHooks.getCurrentServer(); + if (server == null) throw new IllegalStateException("CC tick without a running server"); + DimThreadCore.MANAGER.runMainThreadPhase(server, "CC TickScheduler", () -> { + synchronized (dimensionalripper$schedulerLock) { original.call(); } + }); + } + + // CC's delayed map is global across dimensions. These callbacks only manipulate tokens; + // neither loads chunks nor waits on the main thread. Never take this lock before draining. + @WrapMethod(method = "onChunkTicketChanged") + private static void dimensionalripper$ticket(ServerLevel level, long pos, int oldTicket, int newTicket, Operation original) { + synchronized (dimensionalripper$schedulerLock) { original.call(level, pos, oldTicket, newTicket); } + } + + @WrapMethod(method = "onChunkUnload") + private static void dimensionalripper$unload(LevelChunk chunk, Operation original) { + synchronized (dimensionalripper$schedulerLock) { original.call(chunk); } + } +} diff --git a/src/main/java/dev/modzuozhi/mixin/LevelMixin.java b/src/main/java/dev/modzuozhi/mixin/LevelMixin.java index 9f1ce60..24a7a9f 100644 --- a/src/main/java/dev/modzuozhi/mixin/LevelMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/LevelMixin.java @@ -53,4 +53,14 @@ public abstract class LevelMixin { boolean ownerIsActive = DimThreadCore.owns(this.thread); return callerIsActive && ownerIsActive ? this.thread : t; } + + @org.spongepowered.asm.mixin.injection.Inject(method = "getBlockEntity", at = @At("HEAD")) + private void dimensionalripper$observeLookup(net.minecraft.core.BlockPos pos, + org.spongepowered.asm.mixin.injection.callback.CallbackInfoReturnable ci) { + if (!((Object) this instanceof net.minecraft.server.level.ServerLevel level) || level.isOutsideBuildHeight(pos)) return; + Thread caller = Thread.currentThread(); + Thread owner = this.thread; + if (caller != owner && !((DimThreadCore.owns(caller) || FineGrainScheduler.inSubTask()) && DimThreadCore.owns(owner))) + dev.modzuozhi.core.compat.ThreadOwnershipDiagnostics.rejectedLookup(level, pos, owner); + } } diff --git a/src/main/java/dev/modzuozhi/mixin/MinecraftServerDimMixin.java b/src/main/java/dev/modzuozhi/mixin/MinecraftServerDimMixin.java index 223ebe1..fa43892 100644 --- a/src/main/java/dev/modzuozhi/mixin/MinecraftServerDimMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/MinecraftServerDimMixin.java @@ -81,6 +81,7 @@ public abstract class MinecraftServerDimMixin { /** 服务器停止时：先等所有维度 tick 收口，再停 loop、关线程池，防止保存与 worker 并发卡死。 */ @Inject(method = "stopServer", at = @At("HEAD")) public void modzuozhi_shutdownThreadpool(CallbackInfo ci) { + dev.modzuozhi.core.compat.PayloadMainThreadDispatch.stop((MinecraftServer)(Object)this); // 必须先等正在执行的维度 tick 完全退出再进入保存：否则主线程 saveAllChunks 会与 // 仍在飞的维度 worker（level.tick → processUnloads）并发修改 ChunkMap.toDrop // （非线程安全 fastutil 结构），导致「正在保存中」无限卡死（只能强杀进程）。 @@ -90,4 +91,9 @@ public abstract class MinecraftServerDimMixin { DimThreadCore.MANAGER.threadPools.forEach((server, pool) -> pool.shutdown()); DimThreadCore.MANAGER.clear(); } -} \ No newline at end of file + + @Inject(method = "stopServer", at = @At("RETURN")) + private void dimensionalripper$clearPayloads(CallbackInfo ci) { + dev.modzuozhi.core.compat.PayloadMainThreadDispatch.clear((MinecraftServer)(Object)this); + } +} diff --git a/src/main/java/dev/modzuozhi/mixin/MinecraftServerPayloadMixin.java b/src/main/java/dev/modzuozhi/mixin/MinecraftServerPayloadMixin.java new file mode 100644 index 0000000..8602973 --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/MinecraftServerPayloadMixin.java @@ -0,0 +1,16 @@ +package dev.modzuozhi.mixin; + +import dev.modzuozhi.core.compat.PayloadMainThreadDispatch; +import net.minecraft.server.MinecraftServer; +import org.spongepowered.asm.mixin.Mixin; +import org.spongepowered.asm.mixin.injection.*; +import org.spongepowered.asm.mixin.injection.callback.CallbackInfo; +import java.util.function.BooleanSupplier; + +@Mixin(MinecraftServer.class) +public abstract class MinecraftServerPayloadMixin { + @Inject(method = "tickServer", at = @At("HEAD")) + private void dimensionalripper$drainPayloads(BooleanSupplier budget, CallbackInfo ci) { + PayloadMainThreadDispatch.drain((MinecraftServer) (Object) this); + } +} diff --git a/src/main/java/dev/modzuozhi/mixin/ServerPayloadContextMixin.java b/src/main/java/dev/modzuozhi/mixin/ServerPayloadContextMixin.java new file mode 100644 index 0000000..73d9a9e --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/ServerPayloadContextMixin.java @@ -0,0 +1,38 @@ +package dev.modzuozhi.mixin; + +import com.llamalad7.mixinextras.injector.wrapmethod.WrapMethod; +import com.llamalad7.mixinextras.injector.wrapoperation.Operation; +import dev.modzuozhi.core.compat.PayloadMainThreadDispatch; +import net.minecraft.network.protocol.common.ServerCommonPacketListener; +import net.minecraft.resources.ResourceLocation; +import net.minecraft.server.network.ServerGamePacketListenerImpl; +import net.neoforged.neoforge.network.handling.ServerPayloadContext; +import org.spongepowered.asm.mixin.*; +import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; + +@Mixin(value = ServerPayloadContext.class, remap = false) +public abstract class ServerPayloadContextMixin { + @Shadow @Final private ServerCommonPacketListener listener; + @Shadow @Final private ResourceLocation payloadId; + + @WrapMethod(method = "enqueueWork(Ljava/lang/Runnable;)Ljava/util/concurrent/CompletableFuture;") + private CompletableFuture dimensionalripper$runnable(Runnable task, Operation> original) { + if (!(listener instanceof ServerGamePacketListenerImpl game)) return original.call(task); + return dimensionalripper$submit(game, () -> { task.run(); return null; }); + } + + @WrapMethod(method = "enqueueWork(Ljava/util/function/Supplier;)Ljava/util/concurrent/CompletableFuture;") + private CompletableFuture dimensionalripper$supplier(Supplier task, Operation> original) { + if (!(listener instanceof ServerGamePacketListenerImpl game)) return original.call(task); + return dimensionalripper$submit(game, task); + } + + @Unique private CompletableFuture dimensionalripper$submit(ServerGamePacketListenerImpl game, Supplier task) { + if (game.player.connection != game) + return CompletableFuture.failedFuture(new java.util.concurrent.CancellationException("Replaced payload listener")); + boolean main = game.getMainThreadEventLoop().isSameThread(); + var result = PayloadMainThreadDispatch.submit(game.player, payloadId, task); + return main ? result : PayloadMainThreadDispatch.guard(result); + } +} diff --git a/src/main/java/dev/modzuozhi/mixin/ThreadOwnershipReportMixin.java b/src/main/java/dev/modzuozhi/mixin/ThreadOwnershipReportMixin.java new file mode 100644 index 0000000..9d0e28b --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/ThreadOwnershipReportMixin.java @@ -0,0 +1,18 @@ +package dev.modzuozhi.mixin; + +import dev.modzuozhi.core.compat.ThreadOwnershipDiagnostics; +import net.minecraft.SystemReport; +import net.minecraft.server.MinecraftServer; +import org.spongepowered.asm.mixin.Mixin; +import org.spongepowered.asm.mixin.injection.At; +import org.spongepowered.asm.mixin.injection.Inject; +import org.spongepowered.asm.mixin.injection.callback.CallbackInfoReturnable; + +@Mixin(MinecraftServer.class) +public abstract class ThreadOwnershipReportMixin { + @Inject(method = "fillSystemReport", at = @At("HEAD")) + private void dimensionalripper$ownershipReport(SystemReport report, CallbackInfoReturnable ci) { + report.setDetail("DimensionalRipper Thread Ownership", () -> + ThreadOwnershipDiagnostics.report((MinecraftServer) (Object) this)); + } +} diff --git a/src/main/resources/dimensionalripper.mixins.json b/src/main/resources/dimensionalripper.mixins.json index 5be612f..5d9a908 100644 --- a/src/main/resources/dimensionalripper.mixins.json +++ b/src/main/resources/dimensionalripper.mixins.json @@ -18,6 +18,8 @@ "ClassInstanceMultiMapMixin", "LithiumEntityGroupsMixin", "CollectingNeighborUpdaterMixin", + "ComputerCraftTickSchedulerMixin", + "ThreadOwnershipReportMixin", "DistanceManagerMixin", "EntityChangeDimensionMixin", "EntityGetEntityCollisionsMixin", @@ -38,6 +40,8 @@ "LevelTicksMixin", "LivingEntityGetInBlockStateMixin", "MinecraftServerDimMixin", + "MinecraftServerPayloadMixin", + "ServerPayloadContextMixin", "MinecraftServerSaveMixin", "MinecraftServerTimeSyncAccessor", "NavigatingMobsMixin", diff --git a/tests/integration/InteractionFlowRegression.java b/tests/integration/InteractionFlowRegression.java index fd60398..29c9f5a 100644 --- a/tests/integration/InteractionFlowRegression.java +++ b/tests/integration/InteractionFlowRegression.java @@ -2,13 +2,11 @@ package dev.modzuozhi; import com.mojang.authlib.GameProfile; import dev.modzuozhi.core.compat.SableAssemblyDispatch; -import dev.modzuozhi.core.compat.SableMainThreadBridge; +import dev.modzuozhi.core.compat.PayloadMainThreadDispatch; import dev.modzuozhi.core.dimthread.DimThreadCore; import dev.modzuozhi.core.dimthread.FineGrainScheduler; import dev.modzuozhi.core.dimthread.FaultGuard; import net.minecraft.core.BlockPos; -import net.minecraft.server.level.ClientInformation; -import net.minecraft.server.level.ServerPlayer; import net.minecraft.world.level.block.Blocks; import net.minecraft.world.level.material.Fluids; import net.minecraft.world.level.chunk.status.ChunkStatus; @@ -41,14 +39,14 @@ public final class InteractionFlowRegression { level.setBlockAndUpdate(new BlockPos(x, 100, z), Blocks.AIR.defaultBlockState()); } level.setBlockAndUpdate(SOURCE.above(3), Blocks.CHEST.defaultBlockState()); - var player = new ServerPlayer(server, level, new GameProfile(UUID.randomUUID(), "Regression"), - ClientInformation.createDefault()) { - @Override public boolean hasDisconnected() { return false; } - }; + // A real listener is now part of the shared dispatch's stale-connection contract. + var player = new net.neoforged.neoforge.common.util.FakePlayer(level, + new GameProfile(UUID.randomUUID(), "Regression")); startTime = level.getGameTime(); DimThreadCore.MANAGER.setPaused(server, false); SableAssemblyDispatch.dispatch(player, () -> { - if (!server.isSameThread() || !SableMainThreadBridge.ownsHandoff(level)) + if (!server.isSameThread() || !PayloadMainThreadDispatch.owns(server) + || DimThreadCore.MANAGER.ownershipLoopsSnapshot().stream().anyMatch(loop -> loop.isTicking())) throw new AssertionError("Interaction did not obtain exclusive server ownership"); if (level.getBlockEntity(SOURCE.above(3)) == null) throw new AssertionError("Main-thread block entity lookup silently failed"); diff --git a/tests/ownership/OwnershipRegression.java b/tests/ownership/OwnershipRegression.java new file mode 100644 index 0000000..63210af --- /dev/null +++ b/tests/ownership/OwnershipRegression.java @@ -0,0 +1,200 @@ +package dev.modzuozhi; + +import dev.modzuozhi.core.dimthread.DimThreadCore; +import net.minecraft.core.BlockPos; +import net.minecraft.core.registries.BuiltInRegistries; +import net.minecraft.resources.ResourceLocation; +import net.neoforged.bus.api.SubscribeEvent; +import net.neoforged.fml.common.EventBusSubscriber; +import net.neoforged.neoforge.event.tick.ServerTickEvent; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +/** Local opt-in fixture. Removing the CC safe phase must fail the real identity check. */ +@EventBusSubscriber(modid = "dimensionalripper") +public final class OwnershipRegression { + private static boolean ran; + private static boolean verified; + private static long worldTime; + @SubscribeEvent + public static void tick(ServerTickEvent.Post event) throws Exception { + var server = event.getServer(); + if (verified && server.getTickCount() >= 80) { + DimThreadCore.MANAGER.setPaused(server, true); + if (server.overworld().getGameTime() <= worldTime) throw new AssertionError("Parallel worlds did not resume"); + ModZuozhi.LOGGER.info("OWNERSHIP RESUME PASS: world continued ticking after compatibility phases"); + verified = false; + server.halt(false); + } + if (ran || server.getTickCount() < 20) return; + ran = true; + var level = server.overworld(); + var manager = DimThreadCore.MANAGER; + var release = new CountDownLatch(1); + float originalRate = server.tickRateManager().tickrate(); + try { + manager.setPaused(server, true); + level.setChunkForced(0, 0, true); + level.getChunk(0, 0); + var pos = new BlockPos(8, 110, 8); + var block = BuiltInRegistries.BLOCK.get(ResourceLocation.parse("computercraft:monitor_normal")); + level.setBlockAndUpdate(pos, block.defaultBlockState()); + var monitor = level.getBlockEntity(pos); + if (monitor == null) throw new AssertionError("Monitor fixture not installed"); + var scheduler = Class.forName("dan200.computercraft.shared.util.TickScheduler"); + var tokenClass = Class.forName("dan200.computercraft.shared.util.TickScheduler$Token"); + var token = tokenClass.getConstructor(net.minecraft.world.level.block.entity.BlockEntity.class).newInstance(monitor); + scheduler.getMethod("schedule", tokenClass).invoke(null, token); + var entered = new CountDownLatch(1); + var bridgeFailure = new java.util.concurrent.atomic.AtomicReference(); + manager.getLoop(level).execute(() -> { + // A Sable-like main-thread dependency: safe waiting must pump this callback. + server.execute(release::countDown); + entered.countDown(); + try { + if (!release.await(10, TimeUnit.SECONDS)) throw new AssertionError("Main callback was not pumped"); + dev.modzuozhi.core.compat.SableMainThreadBridge.runForLevel(level, () -> { + if (!server.isSameThread() || level.getBlockEntity(pos) != monitor) + throw new AssertionError("Bridge did not obtain main-thread world ownership"); + }); + } catch (Throwable e) { bridgeFailure.set(e); } + }); + // Amplify idle polling waste: stopping an idle poll must not wait a 1-second tick budget. + server.tickRateManager().setTickRate(1.0F); + manager.setPaused(server, false); + if (!entered.await(10, TimeUnit.SECONDS)) throw new AssertionError("Worker did not start"); + if (level.getBlockEntity(pos) != null) throw new AssertionError("Did not reproduce guarded main-thread lookup"); + for (int i = 0; i < 1000; i++) level.getBlockEntity(pos); + long start = System.nanoTime(); + scheduler.getMethod("tick").invoke(null); + server.tickRateManager().setTickRate(originalRate); + if (System.nanoTime() - start > TimeUnit.MILLISECONDS.toNanos(750)) + throw new AssertionError("Safe phase waited out idle polling budget after pause request"); + if (bridgeFailure.get() != null) throw new AssertionError("Sable bridge failed", bridgeFailure.get()); + ModZuozhi.LOGGER.info("CC PHASE MEASUREMENT: {}", dev.modzuozhi.core.compat.ThreadOwnershipDiagnostics.timing()); + manager.setPaused(server, true); + if (level.getBlockEntity(pos) != monitor) throw new AssertionError("Monitor identity changed"); + var state = tokenClass.getDeclaredField("$state"); + state.setAccessible(true); + if (!state.get(token).toString().equals("IDLE")) throw new AssertionError("CC token was not completed"); + // The same scope must preserve an existing pause, even when nested or throwing. + var expected = new IllegalArgumentException("regression sentinel"); + try { + manager.runMainThreadPhase(server, "outer-test", () -> manager.runMainThreadPhase(server, "inner-test", () -> { throw expected; })); + throw new AssertionError("Phase swallowed exception"); + } catch (IllegalArgumentException failure) { + if (failure != expected) throw failure; + } + if (!manager.getLoop(level).isPaused()) throw new AssertionError("Existing pause was lost"); + manager.setPaused(server, false); + try { + manager.runMainThreadPhase(server, "exception-resume-test", () -> { throw expected; }); + throw new AssertionError("Running phase swallowed exception"); + } catch (IllegalArgumentException failure) { + if (failure != expected) throw failure; + } + if (manager.getLoop(level).isPaused()) throw new AssertionError("Failed phase did not resume running world"); + manager.setPaused(server, true); + // Removed owners are valid stale tokens. Real identity mismatches must still throw. + var removed = net.minecraft.world.level.block.entity.BlockEntityType.CHEST.create(pos, net.minecraft.world.level.block.Blocks.CHEST.defaultBlockState()); + removed.setLevel(level); + var removedToken = tokenClass.getConstructor(net.minecraft.world.level.block.entity.BlockEntity.class).newInstance(removed); + scheduler.getMethod("schedule", tokenClass).invoke(null, removedToken); + removed.setRemoved(); + scheduler.getMethod("tick").invoke(null); + if (!state.get(removedToken).toString().equals("IDLE")) throw new AssertionError("Removed token not discarded"); + removed.clearRemoved(); + scheduler.getMethod("schedule", tokenClass).invoke(null, removedToken); + try { + scheduler.getMethod("tick").invoke(null); + throw new AssertionError("Real mismatch was hidden"); + } catch (java.lang.reflect.InvocationTargetException failure) { + if (!(failure.getCause() instanceof IllegalStateException)) throw failure; + } + var far = new BlockPos(1600000, 100, 1600000); + var unloaded = net.minecraft.world.level.block.entity.BlockEntityType.CHEST.create(far, net.minecraft.world.level.block.Blocks.CHEST.defaultBlockState()); + unloaded.setLevel(level); + var unloadedToken = tokenClass.getConstructor(net.minecraft.world.level.block.entity.BlockEntity.class).newInstance(unloaded); + scheduler.getMethod("schedule", tokenClass).invoke(null, unloadedToken); + int chunks = level.getChunkSource().getLoadedChunksCount(); + scheduler.getMethod("tick").invoke(null); + if (!state.get(unloadedToken).toString().equals("UNLOADED") || chunks != level.getChunkSource().getLoadedChunksCount()) + throw new AssertionError("Unloaded token forced chunk loading or was dropped"); + scheduler.getMethod("onChunkUnload", net.minecraft.world.level.chunk.LevelChunk.class).invoke(null, + new net.minecraft.world.level.chunk.LevelChunk(level, new net.minecraft.world.level.ChunkPos(far))); + if (!state.get(unloadedToken).toString().equals("IDLE")) throw new AssertionError("Unload callback lost delayed token"); + scheduler.getMethod("schedule", tokenClass).invoke(null, unloadedToken); + scheduler.getMethod("tick").invoke(null); + // Ticket promotion requeues a delayed token; unload discards it without masking errors. + scheduler.getMethod("onChunkTicketChanged", net.minecraft.server.level.ServerLevel.class, long.class, int.class, int.class) + .invoke(null, level, net.minecraft.world.level.ChunkPos.asLong(far), 40, 31); + if (!state.get(unloadedToken).toString().equals("SCHEDULED")) throw new AssertionError("Ticket callback lost token"); + unloaded.setRemoved(); + scheduler.getMethod("tick").invoke(null); + var report = server.fillSystemReport(new net.minecraft.SystemReport()).toLineSeparatedString(); + if (!report.contains("DimensionalRipper Thread Ownership") || !report.contains("observedOwner=") + || !report.contains("8, 110, 8") || !report.contains("rejected=1001")) + throw new AssertionError("Crash report lacks ownership evidence: " + report); + if (!report.contains("handoff=completed-owner-restored")) throw new AssertionError("Bridge ownership history missing"); + concurrentCallbacks(server, scheduler, tokenClass, state); + ModZuozhi.LOGGER.info("OWNERSHIP EDGE PASS: nested/exception pauses, removed/mismatched/unloaded tokens, ticket callback, crash diagnostics"); + ModZuozhi.LOGGER.info("OWNERSHIP REGRESSION PASS: real CC monitor identity, callback pumping, scheduled tick; elapsed={}ms", (System.nanoTime()-start)/1_000_000.0); + worldTime = level.getGameTime(); + level.setBlockAndUpdate(pos, net.minecraft.world.level.block.Blocks.AIR.defaultBlockState()); + verified = true; + } finally { + server.tickRateManager().setTickRate(originalRate); + release.countDown(); + manager.setPaused(server, !verified); + if (!verified) server.halt(false); + } + } + + private static void concurrentCallbacks(net.minecraft.server.MinecraftServer server, Class scheduler, + Class tokenClass, java.lang.reflect.Field state) throws Exception { + var levels = java.util.List.of(server.overworld(), server.getLevel(net.minecraft.world.level.Level.NETHER)); + var tokens = new Object[2][40]; + var owners = new net.minecraft.world.level.block.entity.BlockEntity[2][40]; + for (int d = 0; d < 2; d++) for (int i = 0; i < 40; i++) { + var pos = new BlockPos(1600000 + i * 16, 100, 1600000); + var owner = net.minecraft.world.level.block.entity.BlockEntityType.CHEST.create(pos, net.minecraft.world.level.block.Blocks.CHEST.defaultBlockState()); + owner.setLevel(levels.get(d)); + owners[d][i] = owner; + tokens[d][i] = tokenClass.getConstructor(net.minecraft.world.level.block.entity.BlockEntity.class).newInstance(owner); + scheduler.getMethod("schedule", tokenClass).invoke(null, tokens[d][i]); + } + scheduler.getMethod("tick").invoke(null); // all 80 are deferred, no chunks are loaded + var gate = new CountDownLatch(2); + var done = new CountDownLatch(2); + var failure = new java.util.concurrent.atomic.AtomicReference(); + for (int d = 0; d < 2; d++) { + final int dimension = d; + var level = levels.get(d); + DimThreadCore.MANAGER.getLoop(level).execute(() -> { + try { + gate.countDown(); + if (!gate.await(10, TimeUnit.SECONDS)) throw new AssertionError("Dimension callback overlap not reached"); + for (int i = 0; i < 40; i++) { + var pos = owners[dimension][i].getBlockPos(); + if (i % 2 == 0) scheduler.getMethod("onChunkUnload", net.minecraft.world.level.chunk.LevelChunk.class) + .invoke(null, new net.minecraft.world.level.chunk.LevelChunk(level, new net.minecraft.world.level.ChunkPos(pos))); + else scheduler.getMethod("onChunkTicketChanged", net.minecraft.server.level.ServerLevel.class, long.class, int.class, int.class) + .invoke(null, level, net.minecraft.world.level.ChunkPos.asLong(pos), 40, 31); + } + } catch (Throwable error) { failure.set(error); } + finally { done.countDown(); } + }); + } + DimThreadCore.MANAGER.setPaused(server, false); + server.managedBlock(() -> done.getCount() == 0); + DimThreadCore.MANAGER.setPaused(server, true); + if (failure.get() != null) throw new AssertionError("Concurrent callbacks failed", failure.get()); + for (int d = 0; d < 2; d++) for (int i = 0; i < 40; i++) { + String expected = i % 2 == 0 ? "IDLE" : "SCHEDULED"; + if (!state.get(tokens[d][i]).toString().equals(expected)) throw new AssertionError("Delayed token lost across dimensions"); + owners[d][i].setRemoved(); + } + scheduler.getMethod("tick").invoke(null); + ModZuozhi.LOGGER.info("OWNERSHIP CALLBACK PASS: 80 deferred tokens, concurrent overworld/nether ticket/unload callbacks"); + } +} diff --git a/tests/payload/ModInteractionRegression.java b/tests/payload/ModInteractionRegression.java new file mode 100644 index 0000000..0a991c9 --- /dev/null +++ b/tests/payload/ModInteractionRegression.java @@ -0,0 +1,78 @@ +package dev.modzuozhi; + +import net.minecraft.core.*; +import net.minecraft.core.registries.BuiltInRegistries; +import net.minecraft.resources.ResourceLocation; +import net.minecraft.server.MinecraftServer; +import net.minecraft.server.level.ServerPlayer; +import net.minecraft.world.phys.*; +import net.neoforged.neoforge.network.handling.ServerPayloadContext; + +/** Actual optional mod handlers, invoked through the shared context. No production dependencies. */ +public final class ModInteractionRegression { + public static void run(MinecraftServer server,ServerPayloadContext ctx) throws Exception { + if(!net.neoforged.fml.ModList.get().isLoaded("create"))return; + var level=server.overworld();var pos=new BlockPos(12,115,8);var player=ctx.player(); + player.setPos(12,114,8); + var block=BuiltInRegistries.BLOCK.get(ResourceLocation.parse("create:rotation_speed_controller")); + level.setBlockAndUpdate(pos,block.defaultBlockState());var be=level.getBlockEntity(pos); + PayloadRegression.check(be!=null,"Create speed controller fixture"); + var speed=be.getClass().getField("targetSpeed").get(be); + var valueType=Class.forName("com.simibubi.create.foundation.blockEntity.behaviour.ValueSettingsBehaviour"); + int netId=(int)SimulatedHandleRegression.call(valueType,speed,"netId",int.class,new Class[]{}); + var packetType=Class.forName("com.simibubi.create.foundation.blockEntity.behaviour.ValueSettingsPacket"); + var hit=new BlockHitResult(Vec3.atCenterOf(pos),Direction.UP,pos,false); + var packet=packetType.getConstructor(BlockPos.class,int.class,int.class,net.minecraft.world.InteractionHand.class, + BlockHitResult.class,Direction.class,boolean.class,int.class).newInstance(pos,0,32,null,hit,Direction.UP,false,netId); + var packetBase=Class.forName("com.simibubi.create.foundation.networking.BlockEntityConfigurationPacket"); + var speedType=Class.forName("com.simibubi.create.foundation.blockEntity.behaviour.scrollValue.ScrollValueBehaviour"); + try { + ctx.enqueueWork(() -> { + try {SimulatedHandleRegression.call(packetBase,packet,"handle",void.class,new Class[]{ServerPlayer.class},player);} + catch(Exception e){throw new RuntimeException(e);} + }).join(); + int configured=(int)SimulatedHandleRegression.call(speedType,speed,"getValue",int.class,new Class[]{}); + PayloadRegression.check(Math.abs(configured)==32,"actual Create speed packet failed: "+configured); + player.setPos(10000,115,10000); + var denied=packetType.getConstructor(BlockPos.class,int.class,int.class,net.minecraft.world.InteractionHand.class, + BlockHitResult.class,Direction.class,boolean.class,int.class).newInstance(pos,0,64,null,hit,Direction.UP,false,netId); + ctx.enqueueWork(() -> { + try {SimulatedHandleRegression.call(packetBase,denied,"handle",void.class,new Class[]{ServerPlayer.class},player);} + catch(Exception e){throw new RuntimeException(e);} + }).join(); + PayloadRegression.check((int)SimulatedHandleRegression.call(speedType,speed,"getValue",int.class,new Class[]{})==configured,"Create distance check bypassed"); + ModZuozhi.LOGGER.info("CREATE CONFIG PASS: real speed packet and original distance denial"); + player.setPos(12,114,8); + if(net.neoforged.fml.ModList.get().isLoaded("jade"))jade(ctx,pos,be,hit); + } finally {level.setBlockAndUpdate(pos,net.minecraft.world.level.block.Blocks.AIR.defaultBlockState());} + } + static void jade(ServerPayloadContext ctx,BlockPos pos,Object be,BlockHitResult hit) throws Exception { + var dataType=Class.forName("snownee.jade.impl.BlockAccessorImpl$SyncData"); + var data=dataType.getConstructor(boolean.class,BlockHitResult.class,net.minecraft.world.level.block.state.BlockState.class,net.minecraft.world.item.ItemStack.class) + .newInstance(false,hit,ctx.player().level().getBlockState(pos),net.minecraft.world.item.ItemStack.EMPTY); + var requestType=Class.forName("snownee.jade.network.RequestBlockPacket"); + var request=requestType.getConstructor(dataType,java.util.List.class).newInstance(data,java.util.List.of()); + var jadeContext=Class.forName("snownee.jade.network.ServerPayloadContext"); + var proxy=java.lang.reflect.Proxy.newProxyInstance(jadeContext.getClassLoader(),new Class[]{jadeContext},(self,method,args) -> { + if(method.getName().equals("player"))return ctx.player(); + return java.lang.reflect.InvocationHandler.invokeDefault(self,method,args); + }); + var accessorType=Class.forName("snownee.jade.api.BlockAccessor"); + var accessorImpl=Class.forName("snownee.jade.impl.BlockAccessorImpl"); + var seen=new java.util.concurrent.atomic.AtomicInteger(); + java.util.function.Consumer consumer=tag -> { + PayloadRegression.check(tag.getInt("x")==pos.getX(),"Jade response coordinate");seen.incrementAndGet(); + }; + Runnable query=() -> { + try { + var accessor=SimulatedHandleRegression.call(dataType,data,"unpack",accessorType,new Class[]{ServerPlayer.class},ctx.player()); + PayloadRegression.check(SimulatedHandleRegression.call(accessorType,accessor,"getBlockEntity",net.minecraft.world.level.block.entity.BlockEntity.class,new Class[]{})==be,"Jade BE identity"); + SimulatedHandleRegression.call(accessorImpl,null,"handleRequest",void.class,new Class[]{requestType,jadeContext,java.util.function.Consumer.class},request,proxy,consumer); + } catch(Exception e){throw new RuntimeException(e);} + }; + ctx.enqueueWork(query).join();PayloadRegression.check(seen.get()==1,"Jade query callback missing"); + ctx.player().setPos(10000,115,10000);ctx.enqueueWork(query).join(); + PayloadRegression.check(seen.get()==1,"Jade range guard bypassed"); + ModZuozhi.LOGGER.info("JADE REQUEST PASS: real lookup/execute/query and original range denial"); + } +} diff --git a/tests/payload/PayloadRegression.java b/tests/payload/PayloadRegression.java new file mode 100644 index 0000000..84b6e58 --- /dev/null +++ b/tests/payload/PayloadRegression.java @@ -0,0 +1,200 @@ +package dev.modzuozhi; + +import dev.modzuozhi.core.dimthread.DimThreadCore; +import dev.modzuozhi.core.compat.SableMainThreadBridge; +import net.minecraft.core.BlockPos; +import net.minecraft.world.level.block.Blocks; +import net.neoforged.bus.api.SubscribeEvent; +import net.neoforged.fml.common.EventBusSubscriber; +import net.neoforged.neoforge.event.tick.ServerTickEvent; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +@EventBusSubscriber(modid="dimensionalripper") +public final class PayloadRegression { + static boolean ran; + static void check(boolean value, String reason) { if(!value) throw new AssertionError(reason); } + @SubscribeEvent + public static void tick(ServerTickEvent.Post event) throws Exception { + var server=event.getServer(); + if(ran || server.getTickCount()<20) return; + ran=true; + if(Boolean.getBoolean("dimensionalripper.reviewRegression")) { + try {ReviewRegression.run(server);} finally {DimThreadCore.MANAGER.setPaused(server,true);server.halt(false);} + return; + } + var manager=DimThreadCore.MANAGER; var level=server.overworld(); + var release=new CountDownLatch(1); var entered=new CountDownLatch(1); + var failure=new AtomicReference(); + try { + String idleStats=dev.modzuozhi.core.compat.PayloadDiagnostics.snapshot(server); + for(int i=0;i<100;i++) dev.modzuozhi.core.compat.PayloadMainThreadDispatch.drain(server); + check(idleStats.equals(dev.modzuozhi.core.compat.PayloadDiagnostics.snapshot(server)),"empty drain created payload barrier/statistics"); + ModZuozhi.LOGGER.info("PAYLOAD IDLE PASS: 100 empty drains created no payload batches"); + manager.setPaused(server,true); level.setChunkForced(0,0,true); level.getChunk(0,0); + var pos=new BlockPos(8,115,8); level.setBlockAndUpdate(pos,Blocks.CHEST.defaultBlockState()); + var chest=level.getBlockEntity(pos); check(chest!=null,"fixture chest"); + manager.getLoop(level).execute(() -> { + server.execute(release::countDown); entered.countDown(); + try { + check(release.await(5,TimeUnit.SECONDS),"callback not pumped"); + SableMainThreadBridge.runForLevel(level, () -> check(level.getBlockEntity(pos)==chest,"handoff lookup")); + } catch(Throwable e) { failure.set(e); } + }); + manager.setPaused(server,false); check(entered.await(5,TimeUnit.SECONDS),"worker start"); + check(level.getBlockEntity(pos)==null,"must reproduce guarded null before phase"); + manager.runMainThreadPhase(server,"payload-test",() -> { + check(server.isSameThread() && level.getBlockEntity(pos)==chest,"exclusive lookup"); + manager.runMainThreadPhase(server,"nested",() -> check(level.getBlockEntity(pos)==chest,"nested lookup")); + // Regression: a normal resume during exclusive handling must not drop its lease. + manager.setPaused(server,false); + manager.getLoop(level).scheduleIfNeeded(); + try { Thread.sleep(100); } catch(InterruptedException e) { throw new RuntimeException(e); } + check(!manager.getLoop(level).isTicking() && level.getBlockEntity(pos)==chest,"menu resume broke exclusive lease"); + }); + check(failure.get()==null,"Sable callback failed: "+failure.get()); + manager.setPaused(server,true); + manager.runMainThreadPhase(server,"pre-paused",() -> {}); + check(manager.getLoop(level).isPaused(),"pre-pause lost"); + var expected=new IllegalStateException("sentinel"); + try { manager.runMainThreadPhase(server,"failure",() -> {throw expected;}); throw new AssertionError("swallowed"); } + catch(IllegalStateException e) {check(e==expected,"exception identity");} + check(manager.getLoop(level).isPaused(),"exception lost pre-pause"); + ModZuozhi.LOGGER.info("PAYLOAD PHASE PASS"); + entry(server, pos, chest); + } finally { release.countDown(); manager.setPaused(server,true); server.halt(false); } + } + static void entry(net.minecraft.server.MinecraftServer server, BlockPos pos, + net.minecraft.world.level.block.entity.BlockEntity chest) throws Exception { + var manager=DimThreadCore.MANAGER; var level=server.overworld(); + var player=new net.neoforged.neoforge.common.util.FakePlayer(level, + new com.mojang.authlib.GameProfile(java.util.UUID.randomUUID(), "PayloadTest")); + var ctx=new net.neoforged.neoforge.network.handling.ServerPayloadContext(player.connection, + net.minecraft.resources.ResourceLocation.parse("dimensionalripper:test")); + var entered=new CountDownLatch(1); var release=new CountDownLatch(1); + manager.getLoop(level).execute(() -> { + server.execute(release::countDown); entered.countDown(); + try { check(release.await(5,TimeUnit.SECONDS),"entry callback pump"); } + catch(InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(e); } + }); + manager.setPaused(server,false); check(entered.await(5,TimeUnit.SECONDS),"entry worker start"); + try { + var future=ctx.enqueueWork(() -> { + check(server.isSameThread(),"payload ran off main"); + check(level.getBlockEntity(pos)==chest,"public payload entry returned guarded null"); + return ctx.enqueueWork(() -> 42).join(); + }); + check(future.join()==42,"nested supplier result"); + var expected=new IllegalArgumentException("payload sentinel"); + try { ctx.enqueueWork((Runnable) () -> { throw expected; }); throw new AssertionError("main exception hidden"); } + catch(IllegalArgumentException e) { check(e==expected,"main exception identity"); } + ModZuozhi.LOGGER.info("PAYLOAD ENTRY PASS"); + asyncEntry(server,ctx,player); + manager.setPaused(server,false); + var assembled=new AtomicInteger(); + dev.modzuozhi.core.compat.SableAssemblyDispatch.dispatch(player,() -> { + check(server.isSameThread(),"legacy assembly not main"); + check(manager.loopsSnapshot().values().stream().noneMatch(loop -> loop.isTicking()),"legacy assembly lacks global ownership"); + assembled.incrementAndGet(); + }); + check(assembled.get()==1,"legacy dispatch did not use common phase"); + manager.setPaused(server,true); + ModInteractionRegression.run(server,ctx); + SimulatedHandleRegression.run(server,ctx); + var lifecyclePlayer=new net.neoforged.neoforge.common.util.FakePlayer(level, + new com.mojang.authlib.GameProfile(java.util.UUID.randomUUID(),"LifecycleTest")); + lifecycle(server,new net.neoforged.neoforge.network.handling.ServerPayloadContext(lifecyclePlayer.connection,ctx.payloadId()),lifecyclePlayer); + } finally { release.countDown(); manager.setPaused(server,true); } + } + static void asyncEntry(net.minecraft.server.MinecraftServer server, + net.neoforged.neoforge.network.handling.ServerPayloadContext ctx, + net.minecraft.server.level.ServerPlayer player) throws Exception { + var seen=new java.util.ArrayList(); + var fs=new java.util.ArrayList>(); + Thread producer=new Thread(() -> { + for(int i=0;i<100;i++) {int n=i; fs.add(ctx.enqueueWork(() -> {check(server.isSameThread(),"async handler main");seen.add(n);return n;}));} + },"payload-network-fixture"); + producer.start(); producer.join(5000); check(!producer.isAlive(),"network submit blocks"); + check(seen.isEmpty(),"off-thread handler executed inline"); + while(fs.stream().anyMatch(f -> !f.isDone())) dev.modzuozhi.core.compat.PayloadMainThreadDispatch.drain(server); + check(seen.size()==100,"async handler count"); + for(int i=0;i<100;i++) check(seen.get(i)==i && fs.get(i).join()==i,"async FIFO/result"); + var cancelled=new AtomicReference>();var changed=new AtomicInteger(); + Thread cancelProducer=new Thread(() -> cancelled.set(ctx.enqueueWork(changed::incrementAndGet))); + cancelProducer.start();cancelProducer.join(5000);cancelled.get().cancel(false); + dev.modzuozhi.core.compat.PayloadMainThreadDispatch.drain(server); + check(changed.get()==0,"cancelled public future still mutated world"); + var configListener=new net.minecraft.server.network.ServerConfigurationPacketListenerImpl(server, + new net.minecraft.network.Connection(net.minecraft.network.protocol.PacketFlow.SERVERBOUND), + net.minecraft.server.network.CommonListenerCookie.createInitial(player.getGameProfile(),false)); + var configCtx=new net.neoforged.neoforge.network.handling.ServerPayloadContext(configListener,ctx.payloadId()); + check(configCtx.enqueueWork(() -> 17).join()==17,"configuration phase not preserved"); + var guarded=new AtomicReference>(); + Thread failing=new Thread(() -> guarded.set(ctx.enqueueWork(() -> {throw new IllegalArgumentException("expected test failure");}))); + failing.start();failing.join(5000);dev.modzuozhi.core.compat.PayloadMainThreadDispatch.drain(server); + check(guarded.get().join()==null,"NeoForge async handler failure contract changed"); + var raw=new AtomicReference>(); + Thread stale=new Thread(() -> raw.set(dev.modzuozhi.core.compat.PayloadMainThreadDispatch.submit(player,ctx.payloadId(),() -> {throw new AssertionError("stale handler executed");}))); + stale.start();stale.join(5000); + var old=player.serverLevel(); player.setServerLevel(server.getLevel(net.minecraft.world.level.Level.NETHER)); + dev.modzuozhi.core.compat.PayloadMainThreadDispatch.drain(server); + check(raw.get().isCompletedExceptionally(),"stale future pending"); player.setServerLevel(old); + check(dev.modzuozhi.core.compat.PayloadDiagnostics.snapshot(server).contains("expired=1"),"stale work not counted separately"); + var nested=new AtomicReference(); var entered=new CountDownLatch(1); + var manager=DimThreadCore.MANAGER; + manager.setPaused(server,true); + manager.getLoop(old).execute(() -> { + entered.countDown(); + SableMainThreadBridge.runForLevel(old,() -> { + try {ctx.enqueueWork(() -> 1); nested.set(new AssertionError("unsafe handoff reentry accepted"));} + catch(IllegalStateException expected) {} + }); + }); + manager.setPaused(server,false);check(entered.await(5,TimeUnit.SECONDS),"handoff fixture start"); + manager.runMainThreadPhase(server,"handoff-test",() -> {}); + check(nested.get()==null,"handoff reentry result"); + manager.setPaused(server,true); + var blockedFuture=new AtomicReference>();var waitDetected=new AtomicBoolean(); + var waiting=new CountDownLatch(1); + manager.getLoop(old).execute(() -> { + CompletableFuture f=dev.modzuozhi.core.compat.PayloadMainThreadDispatch.submit(player,ctx.payloadId(),() -> {throw new AssertionError("cancelled worker wait action ran");}); + blockedFuture.set(f);waiting.countDown(); + try { f.get(250,TimeUnit.MILLISECONDS); } + catch(TimeoutException expected) {waitDetected.set(true);f.cancel(false);} + catch(Exception e) {throw new RuntimeException(e);} + }); + manager.setPaused(server,false);check(waiting.await(5,TimeUnit.SECONDS),"worker wait fixture start"); + manager.runMainThreadPhase(server,"unsupported-worker-wait",() -> {}); + check(waitDetected.get() && blockedFuture.get().isCancelled(),"worker blocking cycle not detected"); + manager.setPaused(server,true); + ModZuozhi.LOGGER.info("PAYLOAD ASYNC PASS"); + } + static void lifecycle(net.minecraft.server.MinecraftServer server, + net.neoforged.neoforge.network.handling.ServerPayloadContext ctx, + net.minecraft.server.level.ServerPlayer player) throws Exception { + var manager=DimThreadCore.MANAGER; var other=server.getLevel(net.minecraft.world.level.Level.NETHER); + manager.setPaused(server,true); manager.removeLoop(other); + manager.runMainThreadPhase(server,"new-loop",() -> { + var created=manager.getLoop(other);created.scheduleIfNeeded(); + check(created.isPaused() && !created.isTicking(),"new loop escaped phase"); + }); + manager.setPaused(server,true); + // Bounded ID labels, not an unbounded record for every packet. + for(int i=0;i<1000;i++) { + dev.modzuozhi.core.compat.PayloadMainThreadDispatch.submit(player, + net.minecraft.resources.ResourceLocation.parse("dimensionalripper:test_"+i),() -> 1).join(); + } + String stats=dev.modzuozhi.core.compat.PayloadDiagnostics.snapshot(server); + check(stats.contains("ids=128") && stats.contains("other="),"diagnostics ID cap: "+stats); + check(stats.contains("pending=0"),"diagnostic pending count: "+stats); + var pending=new AtomicReference>(); + Thread t=new Thread(() -> pending.set(dev.modzuozhi.core.compat.PayloadMainThreadDispatch.submit(player,ctx.payloadId(),() -> {throw new AssertionError("work after stop");}))); + t.start();t.join(5000); + manager.runMainThreadPhase(server,"stop-inside-phase",() -> dev.modzuozhi.core.compat.PayloadMainThreadDispatch.stop(server)); + check(pending.get().isDone() && pending.get().isCompletedExceptionally(),"stop leaves future pending"); + check(manager.getLoop(other).isPaused(),"stop restored worker"); + check(dev.modzuozhi.core.compat.PayloadMainThreadDispatch.submit(player,ctx.payloadId(),() -> 1).isCompletedExceptionally(),"stop accepts new work"); + ModZuozhi.LOGGER.info("PAYLOAD LIFECYCLE PASS"); + ModZuozhi.LOGGER.info("PAYLOAD DIAGNOSTICS PASS"); + } +} diff --git a/tests/payload/ReviewRegression.java b/tests/payload/ReviewRegression.java new file mode 100644 index 0000000..8b47c59 --- /dev/null +++ b/tests/payload/ReviewRegression.java @@ -0,0 +1,81 @@ +package dev.modzuozhi; + +import dev.modzuozhi.core.compat.*; +import dev.modzuozhi.core.dimthread.*; +import net.minecraft.server.MinecraftServer; +import net.minecraft.resources.ResourceLocation; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; +import java.util.*; + +public final class ReviewRegression { + static final List failures=new ArrayList<>(); + static void expect(boolean ok,String name){if(!ok){failures.add(name);ModZuozhi.LOGGER.error("REVIEW RED: {}",name);}} + public static void run(MinecraftServer server) throws Exception { + var manager=DimThreadCore.MANAGER;manager.setPaused(server,true); + var player=new net.neoforged.neoforge.common.util.FakePlayer(server.overworld(), + new com.mojang.authlib.GameProfile(UUID.randomUUID(),"ReviewTest")); + var disconnects=new AtomicInteger(); + new net.minecraft.server.network.ServerGamePacketListenerImpl(server, + new net.minecraft.network.Connection(net.minecraft.network.protocol.PacketFlow.SERVERBOUND),player, + net.minecraft.server.network.CommonListenerCookie.createInitial(player.getGameProfile(),false)) { + @Override public void disconnect(net.minecraft.network.chat.Component reason){disconnects.incrementAndGet();} + }; + var ctx=new net.neoforged.neoforge.network.handling.ServerPayloadContext(player.connection,ResourceLocation.parse("dimensionalripper:review")); + var result=new AtomicReference>(); + Thread rejected=new Thread(() -> result.set(ctx.enqueueWork(() -> {throw new RejectedExecutionException("handler executor closed");}))); + rejected.start();rejected.join(5000);PayloadMainThreadDispatch.drain(server);PayloadMainThreadDispatch.drain(server); + expect(disconnects.get()==0,"handler rejection must not disconnect player"); + try{expect(result.get().join()==null,"guard preserves handler rejection result");} + catch(CompletionException e){expect(false,"guard preserves handler rejection result");} + Thread cancelledHandler=new Thread(() -> result.set(ctx.enqueueWork(() -> {throw new CancellationException("handler cancelled its own computation");}))); + cancelledHandler.start();cancelledHandler.join(5000);PayloadMainThreadDispatch.drain(server); + expect(result.get().join()==null,"guard preserves handler cancellation result"); + var overflow=new AtomicReference>(); + Thread flood=new Thread(() -> {for(int i=0;i<257;i++)overflow.set(ctx.enqueueWork(() -> 1));}); + flood.start();flood.join(5000); + expect(overflow.get().isCompletedExceptionally(),"capacity rejection remains exceptional"); + PayloadMainThreadDispatch.drain(server); + expect(disconnects.get()==1,"genuine capacity rejection disconnects once"); + for(int i=0;i<10;i++)PayloadMainThreadDispatch.drain(server); + String counters=PayloadDiagnostics.snapshot(server); + expect(counters.contains("failed=2") && counters.contains("rejected=1"), + "diagnostics distinguish handler failures from capacity rejection"); + + var other=server.getLevel(net.minecraft.world.level.Level.NETHER); + manager.removeLoop(other); + manager.runMainThreadPhase(server,"review-new-loop",() -> manager.getLoop(other).scheduleIfNeeded()); + expect(manager.getLoop(other).isPaused(),"new loop inherits menu pause after phase exit"); + manager.setPaused(server,true); + + var loop=manager.getLoop(other);var entered=new CountDownLatch(1);var release=new CountDownLatch(1); + var finished=new AtomicBoolean(); + loop.execute(() -> { + server.execute(() -> manager.removeLoop(other));entered.countDown(); + while(release.getCount()!=0) try{release.await();}catch(InterruptedException ignored){} + finished.set(true); + }); + manager.setPaused(server,false);PayloadRegression.check(entered.await(5,TimeUnit.SECONDS),"retiring loop fixture start"); + Thread releaser=new Thread(() -> {try{Thread.sleep(250);}catch(InterruptedException ignored){}release.countDown();}); + releaser.start(); + try{manager.runMainThreadPhase(server,"review-retire",() -> expect(finished.get() && !loop.isTicking(),"retired loop must drain before exclusive phase"));} + finally{release.countDown();releaser.join(5000);} + manager.setPaused(server,true); + + var labelled=new CountDownLatch(1);var labelRelease=new CountDownLatch(1); + var late=new AtomicReference>(); + Thread submitter=new Thread(() -> { + labelled.countDown();try{labelRelease.await(5,TimeUnit.SECONDS);}catch(InterruptedException e){throw new RuntimeException(e);} + // Exact late-registration boundary after the submitter had obtained its queue. + var ticket=PayloadDiagnostics.begin(server,"dimensionalripper:shutdown_race"); + PayloadDiagnostics.end(server,ticket,new CancellationException("stopped")); + late.set(PayloadMainThreadDispatch.submit(player,ctx.payloadId(),() -> {throw new AssertionError("late action");})); + }); + submitter.start();PayloadRegression.check(labelled.await(5,TimeUnit.SECONDS),"late submit fixture pause"); + PayloadMainThreadDispatch.stop(server);PayloadMainThreadDispatch.clear(server);labelRelease.countDown();submitter.join(5000); + expect(late.get()!=null && late.get().isCompletedExceptionally(),"late future completed"); + expect(PayloadDiagnostics.snapshot(server).equals("payloads: pending=0 batches=0"),"late submission must not recreate diagnostics after clear"); + if(!failures.isEmpty())throw new AssertionError(String.join("; ",failures)); + ModZuozhi.LOGGER.info("PAYLOAD REVIEW PASS"); + } +} diff --git a/tests/payload/SimulatedHandleRegression.java b/tests/payload/SimulatedHandleRegression.java new file mode 100644 index 0000000..beaef33 --- /dev/null +++ b/tests/payload/SimulatedHandleRegression.java @@ -0,0 +1,101 @@ +package dev.modzuozhi; + +import net.minecraft.core.BlockPos; +import net.minecraft.core.registries.BuiltInRegistries; +import net.minecraft.resources.ResourceLocation; +import net.minecraft.server.MinecraftServer; +import net.neoforged.neoforge.network.handling.ServerPayloadContext; + +/** Isolated real packet, plot block entity and native constraint lifecycle fixture. */ +public final class SimulatedHandleRegression { + public static void run(MinecraftServer server, ServerPayloadContext ctx) throws Exception { + if (!net.neoforged.fml.ModList.get().isLoaded("simulated")) return; + var level=server.overworld(); + var handleId=ResourceLocation.parse("simulated:iron_handle"); + ModZuozhi.LOGGER.info("Handle fixture block: {}",handleId); + var block=BuiltInRegistries.BLOCK.get(handleId); + var packet=Class.forName("dev.simulated_team.simulated.network.packets.UpdatePlayerUsingHandlePacket"); + var ctor=packet.getConstructor(float.class,boolean.class,BlockPos.class); + var veilType=Class.forName("foundry.veil.api.network.handler.ServerPacketContext"); + var veil=Class.forName("foundry.veil.forge.network.NeoForgeServerPacketContext") + .getConstructor(net.neoforged.neoforge.network.handling.IPayloadContext.class).newInstance(ctx); + var handle=packet.getMethod("handle",veilType); + var containerType=Class.forName("dev.ryanhcode.sable.api.sublevel.SubLevelContainer"); + var subType=Class.forName("dev.ryanhcode.sable.sublevel.SubLevel"); + var serverSubType=Class.forName("dev.ryanhcode.sable.sublevel.ServerSubLevel"); + var poseType=Class.forName("dev.ryanhcode.sable.companion.math.Pose3d"); + var container=call(containerType,null,"getContainer",Class.forName("dev.ryanhcode.sable.api.sublevel.ServerSubLevelContainer"), + new Class[]{net.minecraft.server.level.ServerLevel.class},level); + var sub=call(containerType,container,"allocateNewSubLevel",subType,new Class[]{poseType},poseType.getConstructor().newInstance()); + var plot=subType.getMethod("getPlot").invoke(sub); + var plotType=Class.forName("dev.ryanhcode.sable.sublevel.plot.LevelPlot"); + var pos=((BlockPos)plotType.getMethod("getCenterBlock").invoke(plot)).atY(115); + plotType.getMethod("newEmptyChunk",net.minecraft.world.level.ChunkPos.class).invoke(plot,new net.minecraft.world.level.ChunkPos(pos)); + level.setBlockAndUpdate(pos,block.defaultBlockState()); + var be=level.getBlockEntity(pos); + PayloadRegression.check(be!=null && be.getClass().getName().endsWith("HandleBlockEntity"),"real handle plot fixture"); + var players=be.getClass().getDeclaredField("players");players.setAccessible(true); + var systemType=Class.forName("dev.ryanhcode.sable.sublevel.system.SubLevelPhysicsSystem"); + var system=systemType.getMethod("get",net.minecraft.world.level.Level.class).invoke(null,level); + var body=systemType.getMethod("getPhysicsHandle",serverSubType).invoke(system,sub); + var bodyType=Class.forName("dev.ryanhcode.sable.api.physics.handle.RigidBodyHandle"); + var jointType=Class.forName("dev.ryanhcode.sable.api.physics.constraint.PhysicsConstraintHandle"); + Object joint=null; + var helper=Class.forName("dev.ryanhcode.sable.Sable").getField("HELPER").get(null); + var global=(net.minecraft.world.phys.Vec3)call(helper.getClass(),helper,"projectOutOfSubLevel",net.minecraft.world.phys.Vec3.class, + new Class[]{net.minecraft.world.level.Level.class,net.minecraft.world.phys.Vec3.class},level,net.minecraft.world.phys.Vec3.atCenterOf(pos)); + var player=ctx.player();player.setPos(global.x,global.y-1,global.z);player.getAbilities().flying=true; + level.addNewPlayer(player); + var release=new java.util.concurrent.CountDownLatch(1); + try { + for(boolean remove:new boolean[]{false,false,true}) { + var request=ctor.newInstance(4.0f,remove,pos); + if(remove) { + var entered=new java.util.concurrent.CountDownLatch(1); + var manager=dev.modzuozhi.core.dimthread.DimThreadCore.MANAGER; + manager.getLoop(level).execute(() -> { + server.execute(release::countDown);entered.countDown(); + try {PayloadRegression.check(release.await(5,java.util.concurrent.TimeUnit.SECONDS),"native release callback pump");} + catch(InterruptedException e){throw new RuntimeException(e);} + }); + manager.setPaused(server,false); + PayloadRegression.check(entered.await(5,java.util.concurrent.TimeUnit.SECONDS),"native release worker start"); + // The old unprotected path clears player holding state but leaves the joint. + handle.invoke(request,veil); + PayloadRegression.check((boolean)jointType.getMethod("isValid").invoke(joint),"unprotected release did not reproduce stranded joint"); + ModZuozhi.LOGGER.info("SIMULATED NATIVE RED: unprotected release left joint valid while worker owned plot"); + } + ctx.enqueueWork(() -> { + try {handle.invoke(request,veil);} catch(ReflectiveOperationException e) {throw new RuntimeException(e);} + }).join(); + var map=(java.util.Map)players.get(be); + PayloadRegression.check(remove ? map.isEmpty() : map.containsKey(ctx.player().getUUID()),"real handle grab/release state"); + if(!remove) { + be.getClass().getMethod("sable$physicsTick",serverSubType,bodyType,double.class).invoke(be,sub,body,0.05d); + var holding=map.get(player.getUUID()); + var jointField=holding.getClass().getDeclaredField("constraintHandle");jointField.setAccessible(true); + joint=jointField.get(holding); + PayloadRegression.check(joint!=null && (boolean)jointType.getMethod("isValid").invoke(joint),"native handle joint created"); + } else { + PayloadRegression.check(joint!=null && !(boolean)jointType.getMethod("isValid").invoke(joint),"native joint survived release packet"); + dev.modzuozhi.core.dimthread.DimThreadCore.MANAGER.setPaused(server,true); + } + } + ModZuozhi.LOGGER.info("SIMULATED HANDLE PACKET PASS: real packet, BE and native joint invalidated on release"); + } finally { + release.countDown(); + dev.modzuozhi.core.dimthread.DimThreadCore.MANAGER.setPaused(server,true); + level.setBlockAndUpdate(pos,net.minecraft.world.level.block.Blocks.AIR.defaultBlockState()); + level.removePlayerImmediately(player,net.minecraft.world.entity.Entity.RemovalReason.DISCARDED); + var reasonType=Class.forName("dev.ryanhcode.sable.sublevel.storage.SubLevelRemovalReason"); + call(containerType,container,"removeSubLevel",void.class,new Class[]{subType,reasonType},sub,reasonType.getField("REMOVED").get(null)); + } + } + static Object call(Class owner,Object target,String name,Class result,Class[] types,Object...args) throws Exception { + var signature=java.lang.invoke.MethodType.methodType(result,types); + var lookup=java.lang.invoke.MethodHandles.publicLookup(); + var method=target==null?lookup.findStatic(owner,name,signature):lookup.findVirtual(owner,name,signature); + var values=new java.util.ArrayList();if(target!=null)values.add(target);java.util.Collections.addAll(values,args); + try{return method.invokeWithArguments(values);}catch(Throwable failure){throw new RuntimeException(failure);} + } +}