diff --git a/README.md b/README.md index a964c04..1691f1a 100644 --- a/README.md +++ b/README.md @@ -42,9 +42,59 @@ Dimensional Ripper 是一个基于 **NeoForge 1.21.1** 的多线程维度并行 - NeoForge **21.1.x** - JDK **21** -## ⚠️ 兼容性警告 +## ⚠️ 兼容性说明 -> **不兼容警告**：本模组**不建议与任何其他多线程优化模组或跨维度交互类模组同时使用**。经实际测试，与这类模组共同加载会发生冲突并导致游戏崩溃。若需使用本模组，请单独加载，或移除其它多线程优化模组 / 跨维度交互模组。 +**已知未修复问题（1.1.15）**：与 C2ME 的 `c2me_fixes_worldgen_threading_issues` 同时加载时，双方对 `Level` 构造器内随机源初始化的 `@Redirect` 冲突，可能导致启动失败。日志已确认 C2ME 优先级 1100 的重定向覆盖本模组优先级 1000 的 `modzuozhi_threadSafeRandom`，继而触发注入检查失败。当前源码尚未包含此兼容修复，不能视为可直接部署的稳定版。 + +1.1.15 针对 Sable 碰撞查询期间反复创建实心占位区块的问题： + +- 删除 `ModZuozhiSolidChunk`。并行子任务的 `getChunk(..., required=false)` 在区块缺失时返回 `null`，不再伪造“已加载”结果、不分配区块段，也不触发加载/生成。已有区块仍返回当前活动对象。 +- `required=true` 的缺失 FULL 区块与缺失其它状态统一快速失败，并关闭后续细粒度并行；不会等待正处于屏障的维度线程，也不在部分执行后重跑同一个实体。 +- 实体预计移动包围盒（外扩 1 格）涉及未加载区块，或扫描跨度超过 8 个区块时，在实体批提交前交所属维度线程串行执行；已加载区域仍保留并行。此预检不是对任意模组远距离查询的保证，遗漏的必须加载请求仍由上一条保护。 +- 每个超时阶段只输出一次详细线程栈，限服务器/本模组非空闲线程；后续保留简短等待信息。超时不会强行放走仍在操作世界的任务。 + +隔离 NeoForge 21.1.244 / Java 21 运行测试：两条实际子任务共执行 60 万次缺失查询、40 万次已加载查询，确认返回值/对象身份、区块数量不增加且并行未降级；验证缺失 FULL/EMPTY 必须加载请求快速失败，加载区域实体并行、超大移动范围实体串行且不重复执行。主线程交接、水流计划刻和维度完成进度测试同时通过。测试中的一次 FaultGuard ERROR 是必须加载失败用例主动触发并在测试中复位，不代表正常运行需要复位。运行方式：`gradlew runServer -PinteractionRegression`，仅用于隔离测试世界。 + +本次未运行用户完整服务端整合包（尤其其额外的 Sable Chunk Guard 1.0.0）；以上不是完整飞行器/区块发送验收，也不宣称此前 DriveByWire 并发异常已修复。请在存档副本测试，不直接用于正式服。 + +1.1.14 增加每维度进度监控：`/dimensionalripper status` 列出所有已创建维度循环的实际完成 TPS（约 10 秒滚动窗口，100ms 分桶）、工作 MSPT 均值/最近值、当前 Tick 持续时间、距最近成功完成时间及失败 Tick 总数。TPS 不再由工作耗时推算，也不钳制到 20；没有新完成时约 10 秒内降至 0，当前 Tick 或无完成时间超过 1 秒单独提示，提示不等于断定死锁。暂停/停止显示状态和 `—` TPS；恢复后的窗口包含暂停时间，约 10 秒后更新完毕。无采样不显示 20。统计随维度循环销毁，不跨存档复用。 + +工作 MSPT 含玩家包处理、主线程交接等待和世界工作，排除末尾区块队列轮询；完成记录在整个循环归还世界所有权后写入。世界 Tick 抛异常不计成功完成；细粒度内部自行捕获的错误仍应结合故障状态判断。这些指标描述执行进度，不能证明水流、组装等游戏逻辑正确，也不修改 Carpet 的 HUD。 + +1.1.13 为水流问题的**诊断版，不宣称修复停流**。`flowstatus` 改为优先检查准星所指的 8 格内方块/水源，报告实体加载状态、LevelTicks 完整执行条件、最近调度时间、该区块队首计划刻的位置/剩余时间和调度索引。此前 1.1.12 仅显示玩家区块状态及全维度计数，不能据此判断同两项流体任务一直被卡住。本版不绕过实体加载门槛、不重排计划刻、不改写用户存档。 + +1.1.12 候选修复：Simulated 的 AssemblePacket（组装/拆解）和 Create 的 BlockEntityConfigurationPacket（包含转速 ValueSettingsPacket）先进入所属维度安全点，再借用服务器主线程执行原处理器。避免维度线程接管期间，主线程 getBlockEntity 返回空值而请求静默失效；保留原处理器的距离和权限检查，排队后换维度/离线则丢弃请求。其余维度继续运行，不全局关闭并行。 + +隔离运行测试 `InteractionFlowRegression` 验证了真实 ServerLevel 中的独占主线程方块实体读取，以及放水后的计划刻扩散。测试不加载 Create/Simulated，不能代替完整整合包的实际拉杆、转速和拆解验收；报告的整合包水流停滞尚未复现、未宣称修复。可站在问题水源附近间隔数秒执行两次 `/dimensionalripper flowstatus`，获取世界时间、区块更新条件和计划刻数量；诊断只应用户请求输出，不持续刷日志。 + +1.1.11 修正：加载界面的暂停状态不再停止维度循环。暂停交接移到服务器线程，在原版暂停保存之前等待在途维度 tick 完成，并继续处理 Sable 回调；已暂停且完全空闲的维度将区块处理权交还服务器主线程。调度入口与暂停共用同步保护和代次校验，防止取消后的旧任务在恢复时重复启动。 + +验证：Lithium 0.15.3 + Sable 2.0.5 临时服务端完成 10 次暂停状态下的真实 ServerPlayer 构造、同步生成新区块、恢复循环，以及 10,000 次运行中区块请求，细粒度保持开启。纯 Java 测试覆盖加载/登录暂停条件及在途/空闲所有权。服务端测试不加载客户端暂停 Mixin，因此不能替代完整 StarLight 单机进世界、ESC 暂停及飞行器验证。临时测试停机时仍记录 Sable 等待被中断的异常，不能将本测试描述为全程无错误。 + +1.1.10 修正：1.1.9 在 `getChunk` 入口检查 `mainThread` 后才进入原版方法；维度 worker 可在两次检查之间接管，导致真正服务器线程仍落入原版阻塞 `join`，同时 Sable 排队等服务器线程。现在维度循环启动后，外部服务器区块请求一律通过 `tell` 排队，等待期间处理服务器回调；只有明确暂停了维度 worker 的同步 Sable/玩家交接阶段才允许直接读取。维度异常也不再强行清空仍有区块请求的线程队列。 + +本次在 Lithium 0.15.3 + Sable 2.0.5 的临时服务端测试了 10,000 次 server-post-tick 同步区块请求，未卡死且细粒度保持启用，保存退出完成。测试夹具保留于 `tests/integration/ChunkHandoffRegression.java`，不进入发布 jar；完整 StarLight 客户端整合包尚未验证。 + +1.1.9 修正： + +- 子任务屏障改为按完整提交阶段计数，修复“先完成一批、随后又提交任务”时一次性 latch 提前放行的问题。超时和中断均不再允许在任务未完成时继续下一阶段。 +- 10 秒是诊断阈值，不是正常加载预算；超时后关闭后续细粒度阶段，但保留当前维度所有权并记录完整线程栈、锁持有者、阶段和池状态。真正卡死的任务不会被此机制强行解锁，仍需根据诊断定位。 +- 实体查询先在锁内收集结果，再在锁外执行回调，避免回调移动/删除实体导致读锁升级死锁。 +- 子任务读取非 FULL 区块也走已就绪路径；必须同步生成时明确报错并关闭后续细粒度并行，避免等待正在等待自己的维度线程。 +- **Lithium 环境下村民改在所属维度线程执行**：实测 65 村民并行查询 POI 可写坏 Lithium 的 `ListeningLong2ObjectOpenHashMap`。这是定向限制，Sable 物理主线程桥接、其它生物、方块实体和区块环境的并行开关不因此关闭。 +- Lithium 的实体分类缓存添加同一实例上的读写同步，并返回快照，修复碰撞查询绕过 section 锁后与实体移动并发修改分类集合的问题。 + +等待逻辑回归测试：`tests/TaskBarrierRegression.java`（提交间隙、超时、中断、可见性、2 万次并发完成）和 `tests/ManagedFutureAwaitRegression.java`。这些测试与小规模 Lithium + Sable 临时世界测试不替代完整整合包及真实飞行器验证；旧日志缺少子任务栈，无法唯一确定旧 10 秒超时的具体阻塞点。 + +1.1.8 修正：真实服务器主线程同步等待区块 Future 时，使用服务器事件循环的 managedBlock 继续处理回调，避免主线程等待区块、维度线程等待物理/玩家主线程回调的循环等待。区块任务仍由原区块执行器处理。独立等待关系回归测试位于 `tests/ManagedFutureAwaitRegression.java`；它验证回调归属、完成、异常和取消，不替代整合包内的进服及飞行器测试。 + +1.1.7 修正：加载 StarLight Extra 时，玩家连接 Tick（包括玩家 Tick 事件和断开清理）同步交回服务器主线程，对应维度等待完成后继续。区块反序列化恢复使用原有区块执行器，避免后台反序列化同时写入 Lithium 的 POI 存储。其它维度及细粒度并行开关不因此关闭。这是针对已观察到异常的修补，编译或空服启动成功不代表完整整合包、玩家交互和飞行器压力测试已通过。 + +> **Create: Aeronautics / Sable**：1.1.5 起内置主线程桥接。Sable 的物理子世界 Tick 同步交回真正的服务器主线程，完成后维度 worker 继续运行。已开始的物理阶段必须结束后才能归还所有权，不能通过超时强行中断。可通过 `/dimensionalripper status` 确认桥接及 Lithium 村民限制。 +> +> **Lithium**：1.1.6 起会优先保留 Lithium 的实体移动碰撞注入，避免双方争抢同一调用点导致启动崩溃。本模组的实体互推关闭仍会生效；移动碰撞查询交由 Lithium 优化。 +> +> 其它多线程优化模组或跨维度交互类模组仍不建议与本模组同时使用；尚未建立专门兼容路径的组合可能发生冲突。 ## 指令 diff --git a/build.gradle b/build.gradle index 02d2053..84aed20 100644 --- a/build.gradle +++ b/build.gradle @@ -76,6 +76,12 @@ dependencies { implementation "net.neoforged:neoforge:${neo_version}" } +// Opt-in isolated runtime fixture; excluded from normal builds. +if (providers.gradleProperty('interactionRegression').isPresent()) { + sourceSets.main.java.srcDir('tests/integration') + sourceSets.main.java.exclude('**/ChunkHandoffRegression.java') +} + tasks.withType(ProcessResources).configureEach { var replaceProperties = [ minecraft_version : minecraft_version, @@ -107,3 +113,26 @@ idea { downloadJavadoc = true } } + +// Pure-Java scheduling regressions run in check without bootstrapping Minecraft. +def regressionClasses = layout.buildDirectory.dir('regression') +tasks.register('compileConcurrencyRegression', JavaCompile) { + source = files('tests/TaskBarrierRegression.java', 'tests/ManagedFutureAwaitRegression.java', + 'src/main/java/dev/modzuozhi/core/compat/TaskBarrier.java', + 'src/main/java/dev/modzuozhi/core/compat/ManagedFutureAwait.java', + 'src/main/java/dev/modzuozhi/core/compat/PauseOwnership.java', 'tests/PauseOwnershipRegression.java', + 'src/main/java/dev/modzuozhi/core/compat/TickProgress.java', 'tests/TickProgressRegression.java') + classpath = files() + destinationDirectory = regressionClasses +} +['TaskBarrierRegression', 'ManagedFutureAwaitRegression', 'PauseOwnershipRegression', 'TickProgressRegression'].each { regressionMain -> + def regressionTask = tasks.register("run${regressionMain}", JavaExec) { + dependsOn 'compileConcurrencyRegression' + classpath = files(regressionClasses) + mainClass = regressionMain + javaLauncher = javaToolchains.launcherFor { + languageVersion = JavaLanguageVersion.of(21) + } + } + tasks.named('check') { dependsOn regressionTask } +} diff --git a/gradle.properties b/gradle.properties index 2724a98..7c2b53e 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1,6 +1,5 @@ # Sets default memory used for gradle commands. Can be overridden by user or command line properties. # This is required to provide enough memory for the Minecraft decompilation process. -org.gradle.java.home=C:\\Program Files\\Java\\jdk-21 org.gradle.jvmargs=-Xmx3G org.gradle.daemon=true org.gradle.parallel=true @@ -37,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.4 +mod_version=1.1.15 # 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 1199b85..6aeb28c 100644 --- a/src/main/java/dev/modzuozhi/command/MasterCommand.java +++ b/src/main/java/dev/modzuozhi/command/MasterCommand.java @@ -3,12 +3,12 @@ package dev.modzuozhi.command; import com.mojang.brigadier.CommandDispatcher; import com.mojang.brigadier.context.CommandContext; import dev.modzuozhi.ModZuozhi; +import dev.modzuozhi.core.compat.SableMainThreadBridge; import dev.modzuozhi.core.dimthread.DimThreadCore; import dev.modzuozhi.core.dimthread.EntityTickParallel; import dev.modzuozhi.core.dimthread.FaultGuard; import dev.modzuozhi.core.dimthread.FineGrainScheduler; import dev.modzuozhi.core.dimthread.ShardGate; -import dev.modzuozhi.core.dimthread.TpsTracker; import dev.modzuozhi.core.filter.SerDesFilter; import net.minecraft.commands.CommandSourceStack; import net.minecraft.commands.arguments.ResourceLocationArgument; @@ -37,6 +37,7 @@ public final class MasterCommand { public static void register(CommandDispatcher dispatcher) { dispatcher.register( literal("dimensionalripper") + .then(literal("flowstatus").executes(MasterCommand::flowStatus)) .then(literal("status").executes(MasterCommand::status)) .then(literal("fine") .then(literal("on").executes(ctx -> setFine(ctx, true))) @@ -67,12 +68,8 @@ public final class MasterCommand { ModZuozhi.LOGGER.info("[ModZuozhi] 全局随 /dimensionalripper fine on 自动恢复为开启"); } FineGrainScheduler.setEnabled(enabled); - if (!enabled) { - // 关闭细粒度后必须先排空 SUB_POOL 残留子任务：此时维度 loop 已退化为串行 tick， - // 残留实体子任务会与它并发改 EntitySection（靠 isLockDisabled 的残留保护仍持锁互斥， - // 见 EntityTickParallel），等排空后锁才回落到零开销纯串行。 - FineGrainScheduler.awaitIdle(5000L); - } + // Barriers drain each current phase; keep the server event loop available + // for any in-flight task that depends on a server callback. if (enabled) { FaultGuard.reset(); } @@ -117,6 +114,38 @@ public final class MasterCommand { return 1; } + private static int flowStatus(CommandContext ctx) { + var player = ctx.getSource().getPlayer(); + if (player == null) { + ctx.getSource().sendFailure(Component.literal("请由玩家站在不流动的水附近执行。")); + return 0; + } + dev.modzuozhi.core.compat.SableAssemblyDispatch.dispatch(player, () -> { + var level = player.serverLevel(); + var hit = player.pick(8.0, 0.0F, true); + var pos = hit instanceof net.minecraft.world.phys.BlockHitResult blockHit + && hit.getType() == net.minecraft.world.phys.HitResult.Type.BLOCK + ? blockHit.getBlockPos() : player.blockPosition(); + var targetChunk = new net.minecraft.world.level.ChunkPos(pos); + long chunk = targetChunk.toLong(); + String message = "水流诊断：维度=" + level.dimension().location() + + "，目标=" + pos.toShortString() + "，区块=" + targetChunk + + "，世界时间=" + level.getGameTime() + + "，正常刻=" + level.tickRateManager().runsNormally() + + "，调试世界=" + level.isDebug() + + "，区块可更新=" + level.getChunkSource().isPositionTicking(chunk) + + "，实体已加载=" + level.areEntitiesLoaded(chunk) + + "，目标流体=" + level.getFluidState(pos).getType() + + "，待处理流体刻=" + level.getFluidTicks().count() + + "，待处理方块刻=" + level.getBlockTicks().count(); + ctx.getSource().sendSuccess(() -> Component.literal(message), false); + String detail = ((dev.modzuozhi.core.compat.ScheduledTickDiagnostics) level.getFluidTicks()) + .dimensionalripper$describeTicks(pos, level.getGameTime()); + ctx.getSource().sendSuccess(() -> Component.literal("流体调度明细：" + detail), false); + }); + return 1; + } + private static int status(CommandContext ctx) { String fine = FineGrainScheduler.isEnabled() ? "开启" : "关闭"; String noCollide = ModZuozhi.MOB_COLLISION_OFF ? "开启" : "关闭"; @@ -126,25 +155,54 @@ public final class MasterCommand { ServerLevel level = ctx.getSource().getLevel(); ResourceKey dimension = level.dimension(); - // 维度并行激活时：维度 loop 的真实 MSPT（recordDimension 持续更新）； - // 全局被 FaultGuard 降级关闭（退回原版串行，维度 loop 已停、recordDimension 不再更新）时： - // 用主线程 tick 间隔样本，它才是当前真实 tick 周期。 - double tps = DimThreadCore.MANAGER.isActive(level.getServer()) - ? TpsTracker.tps(dimension) - : TpsTracker.tps(); - String msg = String.format( "modzuozhi 状态：细粒度=%s%s，分片并行=%s，禁碰撞=%s，故障原因：%s\n" - + "当前维度：%s，TPS：%.1f，掉拍分片累计：实体 %d / TE %d / 区块环境 %d", + + "兼容状态：%s\n" + + "当前维度：%s，掉拍分片累计：实体 %d / TE %d / 区块环境 %d", fine, fineMark, EntityTickParallel.isSharded() ? "开启" : "关闭", noCollide, reason, - dimensionName(dimension), tps, + SableMainThreadBridge.statusReason(), + dimensionName(dimension), EntityTickParallel.missedShards(), ShardGate.teMissed(), ShardGate.chunkMissed()); ctx.getSource().sendSuccess(() -> Component.literal(msg), false); + var loops = DimThreadCore.MANAGER.loopsSnapshot(); + if (loops.isEmpty()) { + ctx.getSource().sendSuccess(() -> Component.literal("维度循环：无数据（尚未启动或已停止），不按 20 TPS 显示。"), false); + } else { + ctx.getSource().sendSuccess(() -> Component.literal( + "维度实测：完成TPS按最近10秒计数（启动不足10秒用已观测时间，含暂停时间）；" + + "工作MSPT为最近100次均值/最近一次，包含交接等待，不含末尾区块队列轮询。"), false); + long now = System.nanoTime(); + loops.entrySet().stream().sorted(java.util.Comparator.comparing(e -> e.getKey().dimension().location().toString())) + .forEach(entry -> { + var loop = entry.getValue(); + var p = loop.progress(now); + String state = !loop.isRunning() ? "已停止" : loop.isPaused() + ? (loop.isTicking() ? "暂停收口中" : "已暂停") + : !loop.isScheduled() ? "未启动" + : p.windowSeconds() == 0 ? "无采样" + : p.runningMs() >= 1000 ? "当前Tick超过1秒" + : p.sinceCompletedMs() >= 1000 ? "超过1秒无完成" + : Double.isNaN(p.sinceCompletedMs()) ? "尚无成功完成" : "运行中"; + String line = String.format(java.util.Locale.ROOT, + "%s [%s] 完成TPS=%s (窗口%.1fs)，工作MSPT(均/末)=%s/%s ms，当前Tick=%s ms，距上次完成=%s ms，失败Tick=%d", + entry.getKey().dimension().location(), state, + loop.isPaused() || !loop.isRunning() ? "—" : metric(p.tps()), + p.windowSeconds(), + metric(p.averageWorkMs()), metric(p.lastWorkMs()), + Double.isNaN(p.runningMs()) ? "未执行" : metric(p.runningMs()), + Double.isNaN(p.sinceCompletedMs()) ? "无记录" : metric(p.sinceCompletedMs()), p.failures()); + ctx.getSource().sendSuccess(() -> Component.literal(line), false); + }); + } return 1; } + private static String metric(double value) { + return Double.isNaN(value) ? "无数据" : String.format(java.util.Locale.ROOT, "%.1f", value); + } + /** 把维度 key 映射成玩家可读的中文名。 */ private static String dimensionName(ResourceKey dimension) { return switch (dimension.location().getPath()) { @@ -154,4 +212,4 @@ public final class MasterCommand { default -> dimension.location().toString(); }; } -} \ No newline at end of file +} diff --git a/src/main/java/dev/modzuozhi/core/compat/ManagedFutureAwait.java b/src/main/java/dev/modzuozhi/core/compat/ManagedFutureAwait.java new file mode 100644 index 0000000..8700dcd --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/ManagedFutureAwait.java @@ -0,0 +1,30 @@ +package dev.modzuozhi.core.compat; + +import java.util.concurrent.CompletableFuture; +import java.util.function.BooleanSupplier; +import java.util.function.Consumer; +import java.util.function.Supplier; + +/** Lets the owning event loop service callbacks while waiting for work on another thread. */ +public final class ManagedFutureAwait { + private ManagedFutureAwait() {} + + /** enqueue must always queue (e.g. BlockableEventLoop.tell), never run inline. */ + public static T callQueued(Supplier work, Consumer enqueue, + Consumer managedBlock) { + CompletableFuture future = new CompletableFuture<>(); + enqueue.accept(() -> { + try { future.complete(work.get()); } + catch (Throwable failure) { future.completeExceptionally(failure); } + }); + return join(future, managedBlock); + } + + public static T join(CompletableFuture future, Consumer managedBlock) { + if (!future.isDone()) { + managedBlock.accept(future::isDone); + } + // Preserve exceptional completion and cancellation semantics. + return future.join(); + } +} diff --git a/src/main/java/dev/modzuozhi/core/compat/PauseOwnership.java b/src/main/java/dev/modzuozhi/core/compat/PauseOwnership.java new file mode 100644 index 0000000..e42dff6 --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/PauseOwnership.java @@ -0,0 +1,15 @@ +package dev.modzuozhi.core.compat; + +/** Shared pause policy, independent of client classes so it can be regression-tested. */ +public final class PauseOwnership { + private PauseOwnership() {} + + public static boolean shouldPause(boolean menuPaused, boolean connected, boolean hasLevel, boolean hasPlayer) { + return menuPaused && connected && hasLevel && hasPlayer; + } + + public static boolean workerOwns(boolean scheduled, boolean paused, boolean ticking) { + // A paused in-flight tick still owns the level until its finally restores the fields. + return scheduled && (!paused || ticking); + } +} diff --git a/src/main/java/dev/modzuozhi/core/compat/SableAssemblyDispatch.java b/src/main/java/dev/modzuozhi/core/compat/SableAssemblyDispatch.java new file mode 100644 index 0000000..8835616 --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/SableAssemblyDispatch.java @@ -0,0 +1,38 @@ +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.server.level.ServerPlayer; + +/** Main -> dimension safe point -> exclusive main-thread phase, without blocking main on dimension. */ +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; + } + })); + } +} diff --git a/src/main/java/dev/modzuozhi/core/compat/SableMainThreadBridge.java b/src/main/java/dev/modzuozhi/core/compat/SableMainThreadBridge.java new file mode 100644 index 0000000..81044cc --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/SableMainThreadBridge.java @@ -0,0 +1,172 @@ +package dev.modzuozhi.core.compat; + +import dev.modzuozhi.ModZuozhi; +import dev.modzuozhi.core.dimthread.DimThreadCore; +import dev.modzuozhi.core.dimthread.FaultGuard; +import net.minecraft.server.MinecraftServer; +import net.minecraft.server.level.ServerLevel; +import net.neoforged.fml.ModList; + +import java.lang.reflect.Method; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Routes Sable's sub-level/physics tick back to the real Minecraft server thread. + * + *

Dimensional Ripper normally runs {@code ServerLevel.tick} on a dimension worker and swaps the + * two vanilla thread-owner fields for the duration. Sable retains additional main-thread-bound + * sub-level and native physics state, so running its container tick on that worker violates Sable's + * threading contract. The Sable compatibility mixin calls this bridge at the start of the parent + * level tick. The dimension worker waits while only the Sable segment runs on the server thread; + * after it completes, normal dimension and fine-grained parallelism continues unchanged.

+ */ +public final class SableMainThreadBridge { + private static final long TIMEOUT_SECONDS = 10L; + private static final ThreadLocal HANDED_OFF_LEVEL = new ThreadLocal<>(); + + public static boolean ownsHandoff(ServerLevel level) { + return HANDED_OFF_LEVEL.get() == level; + } + private static final ConcurrentHashMap, Method> GET_LEVEL_METHODS = new ConcurrentHashMap<>(); + + private SableMainThreadBridge() { + } + + public static boolean isSableLoaded() { + return ModList.get().isLoaded("sable"); + } + + public static String statusReason() { + String reason = isSableLoaded() + ? "航空学/Sable 物理 Tick 已桥接至服务器主线程" + : "未加载航空学/Sable"; + return ModList.get().isLoaded("lithium") + ? reason + "；Lithium 村民 POI 兼容：村民留在维度线程，其余负载按并行开关运行" + : reason; + } + + /** + * Execute one Sable container tick with vanilla server-thread ownership. + * + * @param container Sable's ServerSubLevelContainer (kept as Object so Sable stays optional) + * @param tick original container tick + */ + public static void run(Object container, Runnable tick) { + ServerLevel level = getLevel(container); + runForLevel(level, tick); + } + + /** Synchronously hands off a phase while its owning dimension worker waits. */ + public static void runForLevel(ServerLevel level, Runnable tick) { + MinecraftServer server = level.getServer(); + if (server.isSameThread()) { + tick.run(); + return; + } + + CompletableFuture completion = new CompletableFuture<>(); + // 0 = queued, 1 = running, 2 = cancelled before start. + AtomicInteger state = new AtomicInteger(); + server.execute(() -> { + // A timed-out queued task must not execute later, detached from its waiting level tick. + if (!state.compareAndSet(0, 1)) { + return; + } + try { + // 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. + ServerLevel previous = HANDED_OFF_LEVEL.get(); + HANDED_OFF_LEVEL.set(level); + try { + DimThreadCore.swapThreadsAndRun(tick, level, level.getChunkSource()); + } finally { + if (previous == null) HANDED_OFF_LEVEL.remove(); else HANDED_OFF_LEVEL.set(previous); + } + completion.complete(null); + } catch (Throwable throwable) { + completion.completeExceptionally(throwable); + } + }); + + try { + completion.get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (InterruptedException interrupted) { + if (state.compareAndSet(0, 2)) { + completion.cancel(false); + } else { + // A running main-thread phase must finish before world ownership is released. + completion.handle((result, failure) -> null).join(); + } + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted while waiting for Sable main-thread physics tick", interrupted); + } catch (TimeoutException timeout) { + if (state.compareAndSet(0, 2)) { + completion.cancel(false); + ModZuozhi.LOGGER.error( + "[Compatibility] Sable 物理 Tick 排队超过 {} 秒且尚未开始，交由故障保护降级", + TIMEOUT_SECONDS); + FaultGuard.onBarrierTimeout(); + throw new IllegalStateException("Timed out queueing Sable main-thread physics tick", timeout); + } + // The server thread is already inside native physics. Resuming/falling back here would + // race the same world state; wait for that in-flight step to finish instead. + ModZuozhi.LOGGER.warn( + "[Compatibility] Sable 主线程物理 Tick 已运行超过 {} 秒；为避免世界并发写入，继续等待其安全结束", + TIMEOUT_SECONDS); + awaitRunningTick(completion); + } catch (ExecutionException failed) { + Throwable cause = failed.getCause(); + if (cause instanceof RuntimeException runtimeException) { + throw runtimeException; + } + if (cause instanceof Error error) { + throw error; + } + throw new IllegalStateException("Sable main-thread physics tick failed", cause); + } + } + + private static void awaitRunningTick(CompletableFuture completion) { + try { + completion.get(); + } catch (InterruptedException interrupted) { + completion.handle((result, failure) -> null).join(); + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted during an in-flight Sable physics tick", interrupted); + } catch (ExecutionException failed) { + Throwable cause = failed.getCause(); + if (cause instanceof RuntimeException runtimeException) { + throw runtimeException; + } + if (cause instanceof Error error) { + throw error; + } + throw new IllegalStateException("Sable main-thread physics tick failed", cause); + } + } + + private static ServerLevel getLevel(Object container) { + try { + Method method = GET_LEVEL_METHODS.computeIfAbsent(container.getClass(), type -> { + try { + return type.getMethod("getLevel"); + } catch (ReflectiveOperationException exception) { + throw new IllegalStateException("Unsupported Sable container: " + type.getName(), exception); + } + }); + Object level = method.invoke(container); + if (level instanceof ServerLevel serverLevel) { + return serverLevel; + } + throw new IllegalStateException("Sable container did not return a ServerLevel"); + } catch (ReflectiveOperationException exception) { + throw new IllegalStateException("Unable to read Sable container level", exception); + } + } +} diff --git a/src/main/java/dev/modzuozhi/core/compat/ScheduledTickDiagnostics.java b/src/main/java/dev/modzuozhi/core/compat/ScheduledTickDiagnostics.java new file mode 100644 index 0000000..57892e4 --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/ScheduledTickDiagnostics.java @@ -0,0 +1,7 @@ +package dev.modzuozhi.core.compat; + +import net.minecraft.core.BlockPos; + +public interface ScheduledTickDiagnostics { + String dimensionalripper$describeTicks(BlockPos pos, long gameTime); +} diff --git a/src/main/java/dev/modzuozhi/core/compat/TaskBarrier.java b/src/main/java/dev/modzuozhi/core/compat/TaskBarrier.java new file mode 100644 index 0000000..8f0a845 --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/TaskBarrier.java @@ -0,0 +1,51 @@ +package dev.modzuozhi.core.compat; + +import java.util.concurrent.TimeUnit; + +/** Counts a whole submission phase, including gaps between successive submissions. */ +public final class TaskBarrier { + private int pending; + private boolean sealed; + + public synchronized void register() { + if (sealed) throw new IllegalStateException("Submission after barrier was sealed"); + pending++; + } + + public synchronized void complete() { + if (pending == 0) throw new IllegalStateException("Unbalanced task completion"); + if (--pending == 0) notifyAll(); + } + + public synchronized int pending() { + return pending; + } + + /** A timeout is diagnostic, never permission to resume while tasks still own world state. */ + public void await(long warningMillis, Runnable onSlow) { + boolean interrupted = false; + long interval = TimeUnit.MILLISECONDS.toNanos(warningMillis); + long nextWarning = System.nanoTime() + interval; + try { + while (true) { + synchronized (this) { + sealed = true; + if (pending == 0) return; + long remaining = nextWarning - System.nanoTime(); + if (remaining > 0) { + try { + TimeUnit.NANOSECONDS.timedWait(this, remaining); + } catch (InterruptedException ignored) { + interrupted = true; + } + continue; + } + } + onSlow.run(); + nextWarning = System.nanoTime() + Math.max(interval, TimeUnit.SECONDS.toNanos(30)); + } + } finally { + if (interrupted) Thread.currentThread().interrupt(); + } + } +} diff --git a/src/main/java/dev/modzuozhi/core/compat/TickProgress.java b/src/main/java/dev/modzuozhi/core/compat/TickProgress.java new file mode 100644 index 0000000..2deb5a6 --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/TickProgress.java @@ -0,0 +1,52 @@ +package dev.modzuozhi.core.compat; + +import java.util.Arrays; + +/** Bounded 10-second completion-rate window, independent of Minecraft for deterministic tests. */ +public final class TickProgress { + private static final long BUCKET_NS = 100_000_000L; + private static final int BUCKETS = 100; + private final long[] epochs = new long[BUCKETS], completions = new long[BUCKETS]; + private final long[] work = new long[100]; + private long observed = Long.MIN_VALUE, started = Long.MIN_VALUE, lastCompleted = Long.MIN_VALUE; + private long failures, lastWork; + private int workIndex, workCount; + + public TickProgress() { Arrays.fill(epochs, Long.MIN_VALUE); } + + public synchronized void begin(long now) { + if (observed == Long.MIN_VALUE) observed = now; + started = now; + } + + public synchronized void finish(long now, long workNanos, boolean success) { + lastWork = Math.max(0, workNanos); + work[workIndex] = lastWork; + workIndex = (workIndex + 1) % work.length; + workCount = Math.min(work.length, workCount + 1); + started = Long.MIN_VALUE; + if (!success) { failures++; return; } + lastCompleted = now; + long epoch = Math.floorDiv(now, BUCKET_NS); + int slot = Math.floorMod(epoch, BUCKETS); + if (epochs[slot] != epoch) { epochs[slot] = epoch; completions[slot] = 0; } + completions[slot]++; + } + + public synchronized Snapshot snapshot(long now) { + long epoch = Math.floorDiv(now, BUCKET_NS), count = 0, total = 0; + for (int i = 0; i < BUCKETS; i++) + if (epochs[i] <= epoch && epochs[i] > epoch - BUCKETS) count += completions[i]; + for (int i = 0; i < workCount; i++) total += work[i]; + double seconds = observed == Long.MIN_VALUE ? 0 : Math.min(10, Math.max(0, now - observed) / 1e9); + return new Snapshot(seconds == 0 ? Double.NaN : count / seconds, + workCount == 0 ? Double.NaN : total / (double) workCount / 1e6, + workCount == 0 ? Double.NaN : lastWork / 1e6, + started == Long.MIN_VALUE ? Double.NaN : Math.max(0, now - started) / 1e6, + lastCompleted == Long.MIN_VALUE ? Double.NaN : Math.max(0, now - lastCompleted) / 1e6, + failures, seconds); + } + + public record Snapshot(double tps, double averageWorkMs, double lastWorkMs, + double runningMs, double sinceCompletedMs, long failures, double windowSeconds) {} +} diff --git a/src/main/java/dev/modzuozhi/core/dimthread/DimThreadCore.java b/src/main/java/dev/modzuozhi/core/dimthread/DimThreadCore.java index 86881eb..675fd77 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/DimThreadCore.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/DimThreadCore.java @@ -69,10 +69,9 @@ public final class DimThreadCore { /** * 单机内嵌服务器暂停/恢复信号（见 {@link ServerManager#setPaused}）。 *

- * 由客户端 mixin {@code MixinMinecraftPauseDetector} 在每帧 {@code Minecraft.runTick} 检测 - * {@code isPaused()} 变化后调用；dedicated server 无客户端，永不会触发。 + * 由 {@code IntegratedServerPauseMixin} 在服务器线程同步；加载界面不触发暂停。 */ - public static void setPaused(boolean paused) { - MANAGER.setPaused(paused); + public static void setPaused(MinecraftServer server, boolean paused) { + MANAGER.setPaused(server, paused); } -} \ No newline at end of file +} diff --git a/src/main/java/dev/modzuozhi/core/dimthread/EntityTickParallel.java b/src/main/java/dev/modzuozhi/core/dimthread/EntityTickParallel.java index 4038204..e8b1ad9 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/EntityTickParallel.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/EntityTickParallel.java @@ -156,7 +156,9 @@ public final class EntityTickParallel { * 在两次调用间翻转而对未持有的锁 unlock。 */ public static boolean isLockDisabled() { - return !FineGrainScheduler.isEnabled() && FineGrainScheduler.activeSubTasks() == 0; + // A phase created before fine-off may still submit work after a temporary zero + // pending count. Keep content locks active across mode transitions as well. + return false; } /** 当前线程的实体屏障（worker 端）；无则 null。 */ @@ -237,7 +239,7 @@ public final class EntityTickParallel { /** 实体循环开始：为本维度 tick 建立实体子任务屏障（仅 worker 线程调用）。 */ public static void begin() { - CURRENT.set(FineGrainScheduler.beginBarrier()); + CURRENT.set(FineGrainScheduler.beginBarrier("entity")); } /** diff --git a/src/main/java/dev/modzuozhi/core/dimthread/FaultGuard.java b/src/main/java/dev/modzuozhi/core/dimthread/FaultGuard.java index 3dcdd8f..695ba09 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/FaultGuard.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/FaultGuard.java @@ -55,8 +55,8 @@ public final class FaultGuard { } } - /** 上报维度 tick 异常；达阈值自动回退原版串行。 */ - public static void onDimTickFailure(Throwable t) { + /** Report repeated dimension failures without abandoning outstanding chunk requests. */ + public static synchronized void onDimTickFailure(Throwable t) { if (dimDegraded) { return; } @@ -64,32 +64,34 @@ public final class FaultGuard { ModZuozhi.LOGGER.error("[FaultGuard] 维度 tick 异常（第 {} 次）", dimFailures, t); if (dimFailures >= DIM_FAULT_THRESHOLD) { dimDegraded = true; - reason = "维度 tick 连续异常 " + dimFailures + " 次，已自动回退原版串行"; + fineDegraded = true; + reason = "维度 tick 连续异常 " + dimFailures + " 次，关闭后续细粒度；保留维度所有权和队列，禁止强制切回主线程"; ModZuozhi.LOGGER.error("[FaultGuard] {}", reason); - // 关闭顺序与 setMaster 一致：先关细粒度并排空残留子任务（此时 ENABLED 仍 true， - // section 锁与碰撞拦截仍生效，残留任务安全跑完），再关全局，最后熄灭维度 loop - // （否则主线程退回原版串行 tick 时仍会经 mixin 提交并行子任务 → 双 tick/CME 竞态）。 + // Killing the loops here strands queued chunk futures and races other levels. + // Leave dimension ownership intact; a global handoff requires a separate protocol. FineGrainScheduler.setEnabled(false); - FineGrainScheduler.awaitIdle(5000L); - ModZuozhi.ENABLED = false; - DimThreadCore.MANAGER.killAllLoops(); } } /** 上报子任务屏障等待超时（子任务卡死/死锁）：立即降级，避免维度 worker 永久阻塞导致整服冻结。 */ - public static void onBarrierTimeout() { - if (dimDegraded) { + public static synchronized void onBarrierTimeout() { + if (fineDegraded) { return; } - dimDegraded = true; - reason = "子任务屏障等待超时（疑似死锁），已自动回退原版串行"; + fineDegraded = true; + reason = "子任务等待超过 10 秒，关闭后续细粒度并行；当前任务继续收口，保留维度线程所有权"; + ModZuozhi.LOGGER.error("[FaultGuard] {}", reason); + // Disabling a switch cannot release locks or stop a running task. The owning + // barrier must finish before the next phase; never hand this world to the server here. + FineGrainScheduler.setEnabled(false); + } + + public static synchronized void onUnsafeChunkRequest(String detail) { + if (fineDegraded) return; + fineDegraded = true; + reason = "子任务请求同步生成区块，关闭后续细粒度并行（保留维度并行）: " + detail; ModZuozhi.LOGGER.error("[FaultGuard] {}", reason); - // 与 onDimTickFailure 一致：先关细粒度并尽力排空残留子任务，再关全局， - // 最后熄灭维度 loop，释放可能被卡死子任务持有的锁，恢复主线程原版串行。 FineGrainScheduler.setEnabled(false); - FineGrainScheduler.awaitIdle(2000L); - ModZuozhi.ENABLED = false; - DimThreadCore.MANAGER.killAllLoops(); } /** @@ -112,7 +114,10 @@ public final class FaultGuard { // 与 onDimTickFailure 一致：先关细粒度并排空残留子任务，再关全局， // 避免主线程串行 tick 时仍提交并行子任务（双 tick / section 无锁遍历 CME 竞态）。 FineGrainScheduler.setEnabled(false); - FineGrainScheduler.awaitIdle(5000L); + if (!FineGrainScheduler.awaitIdle(5000L)) { + reason = "维度超时且子任务尚未排空，拒绝切回主线程"; + return false; + } ModZuozhi.ENABLED = false; return true; } diff --git a/src/main/java/dev/modzuozhi/core/dimthread/FineGrainScheduler.java b/src/main/java/dev/modzuozhi/core/dimthread/FineGrainScheduler.java index f378ad1..5223435 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/FineGrainScheduler.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/FineGrainScheduler.java @@ -6,7 +6,9 @@ import net.minecraft.server.MinecraftServer; import net.minecraft.world.entity.Entity; import java.util.List; -import java.util.concurrent.CountDownLatch; +import dev.modzuozhi.core.compat.TaskBarrier; +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -27,9 +29,9 @@ import static dev.modzuozhi.core.dimthread.DimThreadCore.MOD_ID; *

    *
  • 独立线程池：子任务使用 {@link #SUB_POOL} 专用池，与维度 worker 池解耦。 * 否则"维度 worker 阻塞 await 等待子任务"会占满维度池线程，导致子任务永远无人执行 - * 而死锁（这也是之前必现死锁的根因）。独立池固定 CPU 线程，从结构上不可能死锁。
    • - *
  • 独立开关：{@link #isEnabled()} 默认关闭（保守）。关闭时 {@link Barrier#submit} - * 直接同步执行，行为完全退化为串行，绝不引入并发风险；由 {@code /dimensionalripper fine on|off} + * 而死锁。独立线程池只能避免池饥饿，不能消除锁或回调依赖造成的死锁。
    • + *
  • 独立开关：{@link #isEnabled()} 默认开启。关闭时 {@link Barrier#submit} + * 在新阶段同步执行；已开始的阶段保持原模式直到屏障收口；由 {@code /dimensionalripper fine on|off} * 运行时切换。
    • *
  • 异常隔离：子任务异常只记录日志，不中断本维度 tick，也不污染外层维度收口。
    • *
@@ -37,7 +39,7 @@ import static dev.modzuozhi.core.dimthread.DimThreadCore.MOD_ID; public final class FineGrainScheduler { /** - * 细粒度并行总开关（默认开启，已长时间测试稳定；出错时 {@link FaultGuard} 会自动降级关闭）。 + * 细粒度并行总开关（默认开启；出错时 {@link FaultGuard} 会自动降级关闭）。 * 可由 {@code /dimensionalripper fine on|off} 运行时切换。 */ private static volatile boolean enabled = true; @@ -49,7 +51,7 @@ public final class FineGrainScheduler { 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), r -> { Thread t = new Thread(r); t.setDaemon(true); - t.setName(MOD_ID + "_subtask"); + t.setName(MOD_ID + "_subtask_" + t.threadId()); return t; }); @@ -100,17 +102,18 @@ public final class FineGrainScheduler { * 目的：{@code /dimensionalripper off} 或自动降级后，{@link #SUB_POOL} 里可能残留上一拍提交、 * 尚未跑完的实体子任务；此时主线程已退回原版串行 tick 同一批实体，会与残留子任务 * 并发 tick 同一实体导致异常（表现为"关闭时反复出现子任务异常"）。先排空 - * SUB_POOL 再让主线程接管，消除该竞态。超时则强制继续，避免异常场景下永久阻塞。 + * SUB_POOL 再让主线程接管。超时返回 false，调用方必须拒绝不安全的交接。 */ - public static void awaitIdle(long timeoutMs) { + public static boolean awaitIdle(long timeoutMs) { long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMs); while (System.nanoTime() < deadline) { if (ACTIVE_SUBTASKS.get() == 0) { - return; + return true; } LockSupport.parkNanos(100_000L); // 0.1ms 退避 } - ModZuozhi.LOGGER.warn("[FineGrain] 等待子任务排空超时（>{}ms），强制继续", timeoutMs); + ModZuozhi.LOGGER.error("[FineGrain] 等待排空超时，仍有 {} 个子任务，拒绝交接世界所有权", ACTIVE_SUBTASKS.get()); + return false; } /** 当前并行子任务的延迟执行队列；非子任务上下文返回 null。 */ @@ -120,9 +123,11 @@ public final class FineGrainScheduler { /** 为本维度 tick 创建一个新的子任务屏障。 */ public static Barrier beginBarrier() { - return new Barrier(); + return beginBarrier("unspecified"); } + public static Barrier beginBarrier(String phase) { return new Barrier(phase); } + /** * 维度内子任务屏障：提交多个可并行子任务，{@link #await()} 等待全部完成。 *

@@ -130,14 +135,20 @@ public final class FineGrainScheduler { * {@link #submit} 直接在当前线程同步执行，保证开关无关的正确性。 */ public static final class Barrier { - private final CountDownLatch latch = new CountDownLatch(1); + private final TaskBarrier tasks = new TaskBarrier(); + // Freeze mode per phase: never mix inline ticks with still-running siblings. + private final boolean parallel = enabled; + private final String owner = Thread.currentThread().getName(); + private final String phase; + private boolean dumpedThreads; + + private Barrier(String phase) { this.phase = phase; } private final PostExecuteQueue postQueue = new PostExecuteQueue(); - private int pending; /** * 子任务屏障等待超时（毫秒）。一批子任务（方块实体/区块/实体）正常应在几十 ms 内完成， * 超时意味着某个子任务卡死（例如阻塞在另一个线程持有的 {@link ChunkLock} 或 section 锁上）。 - * 此时无条件 {@code latch.await()} 会让维度 worker 永久阻塞，进而死锁主线程保存流程。 + * 超时输出线程和锁诊断，不再允许未完成的任务越过本阶段边界。 */ private static final long AWAIT_TIMEOUT_MS = 10_000L; @@ -172,21 +183,18 @@ public final class FineGrainScheduler { * （排序加锁防死锁），供跨区块访问的子任务使用 */ public void submit(MinecraftServer server, Runnable task, BlockPos pos, int radius) { - if (!FineGrainScheduler.enabled) { + if (!parallel) { task.run(); return; } - synchronized (this) { - pending++; - } + tasks.register(); ACTIVE_SUBTASKS.incrementAndGet(); - SUB_POOL.execute(() -> { - AutoCloseable chunkLock = (pos == null) - ? null - : (radius > 0 ? ChunkLock.lock(pos, radius) : ChunkLock.lock(pos)); + executeOrRun(() -> { + AutoCloseable chunkLock = null; PostExecuteQueue prev = CURRENT_QUEUE.get(); CURRENT_QUEUE.set(postQueue); try { + chunkLock = pos == null ? null : radius > 0 ? ChunkLock.lock(pos, radius) : ChunkLock.lock(pos); task.run(); } catch (Throwable t) { ModZuozhi.LOGGER.error("[FineGrain] 子任务异常", t); @@ -199,13 +207,8 @@ public final class FineGrainScheduler { } catch (Exception ignore) { } } - synchronized (this) { - pending--; - if (pending == 0) { - latch.countDown(); - } - } ACTIVE_SUBTASKS.decrementAndGet(); + tasks.complete(); } }); } @@ -224,7 +227,7 @@ public final class FineGrainScheduler { if (entities.isEmpty()) { return; } - if (!FineGrainScheduler.enabled) { + if (!parallel) { // 细粒度关闭：当前线程逐实体串行 tick（行为等价原版） for (Entity entity : entities) { if (!entity.isRemoved()) { @@ -236,14 +239,12 @@ public final class FineGrainScheduler { int poolSize = SUB_POOL.getCorePoolSize(); int chunkSize = Math.max(1, (entities.size() + poolSize - 1) / poolSize); int chunkCount = (entities.size() + chunkSize - 1) / chunkSize; - synchronized (this) { - pending += chunkCount; - } + for (int i = 0; i < chunkCount; i++) tasks.register(); ACTIVE_SUBTASKS.addAndGet(chunkCount); for (int i = 0; i < entities.size(); i += chunkSize) { int end = Math.min(i + chunkSize, entities.size()); List chunk = entities.subList(i, end); - SUB_POOL.execute(() -> modzuozhi_runChunk(chunk, consumer)); + executeOrRun(() -> modzuozhi_runChunk(chunk, consumer)); } } @@ -262,16 +263,13 @@ public final class FineGrainScheduler { * @param task 子任务（如某一片实体的 tick） */ public void submitShard(MinecraftServer server, ShardGate gate, int shard, Runnable task) { - if (!FineGrainScheduler.enabled) { - gate.release(shard); - task.run(); + if (!parallel) { + try { task.run(); } finally { gate.release(shard); } return; } - synchronized (this) { - pending++; - } + tasks.register(); ACTIVE_SUBTASKS.incrementAndGet(); - SUB_POOL.execute(() -> { + executeOrRun(() -> { PostExecuteQueue prev = CURRENT_QUEUE.get(); CURRENT_QUEUE.set(postQueue); try { @@ -282,13 +280,8 @@ public final class FineGrainScheduler { } finally { CURRENT_QUEUE.set(prev); gate.release(shard); - synchronized (this) { - pending--; - if (pending == 0) { - latch.countDown(); - } - } ACTIVE_SUBTASKS.decrementAndGet(); + tasks.complete(); } }); } @@ -310,36 +303,48 @@ public final class FineGrainScheduler { } } finally { CURRENT_QUEUE.set(prev); - synchronized (this) { - pending--; - if (pending == 0) { - latch.countDown(); - } - } ACTIVE_SUBTASKS.decrementAndGet(); + tasks.complete(); } } - /** 阻塞等待本维度所有已提交子任务完成；超时触发 FaultGuard 降级，避免永久死锁。 */ - public void await() { - // 关键：若从未提交任何子任务（pending==0），latch 永远不会 countDown， - // 无条件 await 会永久死锁。因此先检查 pending，为 0 直接返回。 - synchronized (this) { - if (pending == 0) { - return; - } - } - boolean done; + private static void executeOrRun(Runnable task) { try { - done = latch.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; + SUB_POOL.execute(task); + } catch (java.util.concurrent.RejectedExecutionException rejected) { + task.run(); // Preserve the same accounting and shard-release finally path. } - if (!done) { - ModZuozhi.LOGGER.error("[FineGrain] 子任务屏障等待超时（>{}ms），存在卡死的子任务，触发降级", AWAIT_TIMEOUT_MS); + } + + /** Await the whole phase, retaining ownership after a timeout or interruption. */ + public void await() { + tasks.await(AWAIT_TIMEOUT_MS, () -> { FaultGuard.onBarrierTimeout(); - } + ModZuozhi.LOGGER.error("[FineGrain] 等待超过 10 秒: owner={}, phase={}, pending={}, poolActive={}, queued={}", + owner, phase, tasks.pending(), SUB_POOL.getActiveCount(), SUB_POOL.getQueue().size()); + // One detailed snapshot per stalled phase. Later warnings remain compact: + // repeatedly printing every mod's idle threads can overwhelm launchers. + if (dumpedThreads) return; + dumpedThreads = true; + try { + for (ThreadInfo info : ManagementFactory.getThreadMXBean().dumpAllThreads(true, true)) { + if (info == null) continue; + if (!info.getThreadName().equals("Server thread") + && !info.getThreadName().startsWith(MOD_ID + "_")) continue; + if (java.util.Arrays.stream(info.getStackTrace()).anyMatch(frame -> + frame.getClassName().equals("java.util.concurrent.ThreadPoolExecutor") + && frame.getMethodName().equals("getTask"))) continue; + StringBuilder dump = new StringBuilder().append(info.getThreadName()) + .append(" #").append(info.getThreadId()).append(' ').append(info.getThreadState()) + .append(" waiting=").append(info.getLockInfo()).append(" owner=").append(info.getLockOwnerName()); + for (StackTraceElement frame : info.getStackTrace()) dump.append("\n\tat ").append(frame); + for (var lock : info.getLockedSynchronizers()) dump.append("\n\towned: ").append(lock); + ModZuozhi.LOGGER.error("[FineGrain thread dump] {}", dump); + } + } catch (RuntimeException failure) { + ModZuozhi.LOGGER.error("[FineGrain] Unable to capture thread diagnostics", failure); + } + }); } /** 维度 worker 收口处：串行执行子任务投递的延迟回调。 */ diff --git a/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java b/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java index ba89087..78ed125 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java @@ -62,6 +62,12 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu private volatile boolean ticking = false; /** 单机内嵌服务器暂停（ESC 菜单）时置 true：取消下一拍并不再自排，杜绝暂停/保存期间并行写区块。 */ private volatile boolean paused = false; + private long scheduleGeneration; + private final dev.modzuozhi.core.compat.TickProgress progress = new dev.modzuozhi.core.compat.TickProgress(); + + public dev.modzuozhi.core.compat.TickProgress.Snapshot progress(long now) { + return progress.snapshot(now); + } /** 本循环独立的 tick 计数（用于每 20 tick 的时间同步）。 */ private int tickCount; @@ -89,8 +95,9 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu } /** 申请停止自排程（维度卸载 / 服务器停机时调用）。 */ - public void kill() { + public synchronized void kill() { this.running = false; + this.scheduleGeneration++; Future task = this.currentTickTask; if (task != null && !task.isCancelled()) { task.cancel(true); @@ -98,16 +105,22 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu } /** 首次启动（或外部主动重排）本循环。 */ - public void schedule() { + public synchronized void schedule() { + if (!this.running || this.paused || this.ticking) return; + this.scheduleAfter(0L); + } + + private void scheduleAfter(long delay) { + long generation = ++this.scheduleGeneration; try { - this.currentTickTask = this.masterPool.submit(this); + this.currentTickTask = this.masterPool.schedule(() -> this.runIfCurrent(generation), delay, TimeUnit.NANOSECONDS); } catch (java.util.concurrent.RejectedExecutionException ignored) { // masterPool 已被关闭（停机/降级/重启用）：放弃排程，run 内的 running 检查兜底 } } /** 每 tick 由主线程调用：保证本 loop 只启动一次自排程。 */ - public void scheduleIfNeeded() { + public synchronized void scheduleIfNeeded() { if (!this.scheduled) { this.scheduled = true; this.schedule(); @@ -125,11 +138,12 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu * 不再有任何并行写入 LevelChunk 的随机tick/TE/实体，消灭「暂停/保存期间区块数据包序列化 * 读到中间态 → 客户端解码越界 → 网络协议错误掉线」的竞态。 */ - public void pause() { + public synchronized void pause() { if (this.paused) { return; } this.paused = true; + this.scheduleGeneration++; Future task = this.currentTickTask; if (task != null && !task.isCancelled()) { task.cancel(false); @@ -137,7 +151,7 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu } /** 恢复本维度循环：立即重排下一拍。 */ - public void resume() { + public synchronized void resume() { if (!this.paused) { return; } @@ -160,15 +174,17 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu @Override public void run() { - DimThreadCore.attach(Thread.currentThread(), this.level); - // 暂停期间若仍有取消失败/竞态残留的任务被执行，直接放弃本拍（不进入维度 tick，不续排）； - // 正常路径下 pause() 会 cancel 掉当前拍，这里只是兜底。 - if (this.paused || !this.running) { - this.currentTickTask = null; - return; - } - this.ticking = true; + long generation; + synchronized (this) { generation = this.scheduleGeneration; } + this.runIfCurrent(generation); + } + private void runIfCurrent(long generation) { + synchronized (this) { + if (generation != this.scheduleGeneration || this.paused || !this.running || this.ticking) return; + this.ticking = true; + } + DimThreadCore.attach(Thread.currentThread(), this.level); MinecraftServer server = this.level.getServer(); long nspt = server.tickRateManager().nanosecondsPerTick(); if (server.tickRateManager().isSprinting()) { @@ -176,31 +192,33 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu } long tickStart = System.nanoTime(); + progress.begin(tickStart); + boolean[] completed = {false}; + long[] workNanos = {0L}; long deadline = tickStart + nspt; try { - this.processPendingTasks(); DimThreadCore.swapThreadsAndRun(() -> { - long workStart = System.nanoTime(); - this.internalTick(); - // 只统计真实 tick 工作量（MSPT），不含后面 drainChunkTasks 的按预算轮询等待， - // 避免把定时器唤醒延迟等「非负载时间」算进 TPS（否则空闲也会显示 19.7）。 - TpsTracker.recordDimension(this.level.dimension(), System.nanoTime() - workStart); + try { + this.processPendingTasks(); + completed[0] = this.internalTick(); + } finally { + workNanos[0] = System.nanoTime() - tickStart; + } this.drainChunkTasks(deadline); }, this.level, this.level.getChunkSource()); } catch (Throwable t) { + completed[0] = false; ModZuozhi.LOGGER.error("[DimThread] 维度 {} tick 异常", this.level.dimension().location(), t); FaultGuard.onDimTickFailure(t); } finally { - this.ticking = false; - if (this.running && !this.paused) { - long remaining = Math.max(0L, deadline - System.nanoTime()); - try { - this.currentTickTask = this.masterPool.schedule(this, remaining, TimeUnit.NANOSECONDS); - } catch (java.util.concurrent.RejectedExecutionException ignored) { - // masterPool 已关闭：放弃续排 + progress.finish(System.nanoTime(), workNanos[0], completed[0]); + synchronized (this) { + this.ticking = false; + if (this.running && !this.paused) { + this.scheduleAfter(Math.max(0L, deadline - System.nanoTime())); + } else if (!this.running) { + ModZuozhi.LOGGER.info("[DimThread] 维度 {} 独立循环已停止", this.level.dimension().location()); } - } else if (!this.running) { - ModZuozhi.LOGGER.info("[DimThread] 维度 {} 独立循环已停止", this.level.dimension().location()); } } } @@ -252,7 +270,8 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu * 连接 tick（玩家 doTick）→ 区块发送 + 恢复刷新。注意：本阶段暂不把自动保存下沉到 loop， * 仍由主线程全局保存，避免与 {@code MinecraftServerSaveMixin} 双保存。 */ - private void internalTick() { + private boolean internalTick() { + boolean successful = true; // 先排空玩家包队列（命令等），再进沉重的 level.tick：实体/AI 卡顿时命令仍能及时执行， // 不会像原来那样被饿死到 level.tick 结束（RETURN）才处理。 ((IPlayerPacketQueue) this.level).modzuozhi$playerPacketQueue().drain(); @@ -272,17 +291,25 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu try { this.level.tick(() -> true); } catch (Throwable throwable) { + successful = false; ModZuozhi.LOGGER.error("[DimThread] 维度 {} tick 抛异常", this.level.dimension().location(), throwable); } EventHooks.fireLevelTickPost(this.level, () -> true); - this.tickConnections(); + 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. + dev.modzuozhi.core.compat.SableMainThreadBridge.runForLevel(this.level, this::tickConnections); + } else { + this.tickConnections(); + } for (Connection connection : this.connections) { ServerGamePacketListenerImpl handler = (ServerGamePacketListenerImpl) connection.getPacketListener(); handler.chunkSender.sendNextChunks(handler.player); handler.resumeFlushing(); } + return successful; } /** tick 本维度玩家的连接（正是这里触发原版 {@code ServerPlayer.doTick()}）。 */ @@ -332,4 +359,4 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu private boolean serverTelemetry() { return this.level.getServer().logIPs(); } -} \ No newline at end of file +} diff --git a/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java b/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java index a785786..c5df22a 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java @@ -146,6 +146,11 @@ public class ServerManager { return this.loops.computeIfAbsent(level, l -> new LevelTickLoop(l, getMasterPool())); } + /** Read-only inventory; observing status must never create or start a loop. */ + public Map loopsSnapshot() { + return Map.copyOf(this.loops); + } + /** * 该维度 loop 是否已启动接管（{@code tickChildren → scheduleIfNeeded} 之后才为 true）。 *

@@ -155,7 +160,8 @@ public class ServerManager { */ public boolean isLoopScheduled(ServerLevel level) { LevelTickLoop loop = this.loops.get(level); - return loop != null && loop.isScheduled(); + return loop != null && dev.modzuozhi.core.compat.PauseOwnership.workerOwns( + loop.isScheduled(), loop.isPaused(), loop.isTicking()); } /** @@ -208,7 +214,7 @@ public class ServerManager { /** * 单机内嵌服务器暂停/恢复（由客户端 {@code Minecraft.isPaused()} 驱动，见 - * {@code MixinMinecraftPauseDetector}）。 + * {@code IntegratedServerPauseMixin}，只在服务器线程交接）。 *

* 暂停时：先取消所有维度 loop 的下一拍，再等待正在执行的那一拍完全收口（含其内部 * 并行子任务屏障），使所有维度 worker 彻底停摆——暂停/保存期间不再有任何线程并行写入 @@ -217,23 +223,15 @@ public class ServerManager { *

* 恢复时：重新排程所有维度 loop。LAN 开放时客户端 {@code isPaused()} 恒 false，不会误暂停。 */ - public void setPaused(boolean paused) { + public void setPaused(MinecraftServer server, boolean paused) { + if (!server.isSameThread()) { + throw new IllegalStateException("Pause ownership must be changed on the server thread"); + } if (paused) { this.loops.values().forEach(LevelTickLoop::pause); - long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); - boolean anyTicking = true; - while (anyTicking && System.nanoTime() < deadline) { - anyTicking = false; - for (LevelTickLoop loop : this.loops.values()) { - if (loop.isTicking()) { - anyTicking = true; - break; - } - } - if (anyTicking) { - LockSupport.parkNanos(100_000L); // 0.1ms 退避 - } - } + // 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); } @@ -291,7 +289,8 @@ public class ServerManager { } public void clear() { + TpsTracker.reset(); actives.clear(); threadPools.clear(); } -} \ No newline at end of file +} diff --git a/src/main/java/dev/modzuozhi/core/dimthread/TpsTracker.java b/src/main/java/dev/modzuozhi/core/dimthread/TpsTracker.java index 48cad28..43c3851 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/TpsTracker.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/TpsTracker.java @@ -1,125 +1,32 @@ package dev.modzuozhi.core.dimthread; -import net.minecraft.resources.ResourceKey; -import net.minecraft.world.level.Level; - -import java.util.Map; -import java.util.TreeMap; -import java.util.concurrent.ConcurrentHashMap; - -/** - * 服务器 TPS 估算（供 {@code /dimensionalripper status} 展示，验证优化效果）。 - *

- * 维度独立 tick 循环改造后，主线程退化为「协调者」，恒按 20 拍运行；真实负载体现在每个维度 loop - * 的单次 tick 实际耗时（MSPT）上。因此这里按维度记录每次 {@code internalTick} 的真实耗时， - * {@code TPS = min(20, 1e9 / 平均MSPT)}： - *

    - *
  • 空闲时 MSPT 远小于 50ms → TPS = 20；
    • - *
  • 某维度单 tick 超过 50ms 预算 → TPS < 20（真实掉 TPS）。
    • - *
- * 这比「记录相邻 tick 的墙钟间隔」更准确——后者会把自排程定时器的唤醒延迟（每拍约 0.5~0.8ms） - * 一并计入，导致完全空闲时仍显示 19.7。 - *

- * 主线程样本 {@link #tick()} / {@link #tps()} 仅在维度并行关闭（退回原版串行）时有对比意义， - * 此时主线程间隔即真实 tick 周期。 - */ +/** Main coordinator cadence only; dimension progress lives on each LevelTickLoop. */ public final class TpsTracker { - /** 采样窗口（tick 数）。 */ - private static final int WINDOW = 100; + private static final long[] INTERVALS = new long[100]; + private static int index, count; + private static long lastTick = Long.MIN_VALUE; + private TpsTracker() {} - /** 主线程最近 N 个 tick 间隔（纳秒）。 */ - private static final long[] INTERVALS = new long[WINDOW]; - private static int index; - private static boolean primed; - private static long lastTick = -1L; - - /** 各维度的独立 MSPT 采样（key 为维度，如 {@code minecraft:overworld}）。 */ - private static final Map, Sample> DIMENSIONS = new ConcurrentHashMap<>(); - - private TpsTracker() { - } - - /** 服务器主线程每 tick 调用一次。 */ - public static void tick() { + public static synchronized void tick() { long now = System.nanoTime(); - if (lastTick >= 0L) { - long interval = now - lastTick; - synchronized (TpsTracker.class) { - INTERVALS[index] = interval; - index = (index + 1) % WINDOW; - primed = true; - } + if (lastTick != Long.MIN_VALUE) { + INTERVALS[index] = now - lastTick; + index = (index + 1) % INTERVALS.length; + count = Math.min(count + 1, INTERVALS.length); } lastTick = now; } - /** 某维度 loop 每次完成 {@code internalTick} 后调用，上报本次真实 tick 耗时（纳秒）。 */ - public static void recordDimension(ResourceKey dimension, long tickNanos) { - Sample sample = DIMENSIONS.computeIfAbsent(dimension, k -> new Sample()); - sample.record(tickNanos); + public static synchronized double tps() { + if (count == 0) return Double.NaN; + long total = 0; + for (int i = 0; i < count; i++) total += INTERVALS[i]; + return total <= 0 ? Double.NaN : 1e9 * count / total; } - /** 主线程 (协调者) TPS。数据不足时返回 20。 */ - public static double tps() { - synchronized (TpsTracker.class) { - if (!primed) { - return 20.0; - } - long sum = 0L; - for (long interval : INTERVALS) { - if (interval > 0L) { - sum += interval; - } - } - return sum <= 0L ? 20.0 : tpsFromAvg(sum / (double) WINDOW); - } + public static synchronized void reset() { + index = count = 0; + lastTick = Long.MIN_VALUE; + java.util.Arrays.fill(INTERVALS, 0L); } - - /** 指定维度的真实 TPS。无采样时返回 20。 */ - public static double tps(ResourceKey dimension) { - Sample sample = DIMENSIONS.get(dimension); - return sample == null ? 20.0 : sample.tps(); - } - - /** 按维度名排序返回每维度的 TPS，供 {@code /dimensionalripper status} 展示。 */ - public static Map allDimensionTps() { - Map result = new TreeMap<>(); - for (Map.Entry, Sample> entry : DIMENSIONS.entrySet()) { - result.put(entry.getKey().location().toString(), entry.getValue().tps()); - } - return result; - } - - private static double tpsFromAvg(double avgNanos) { - if (avgNanos <= 0.0) { - return 20.0; - } - return Math.min(20.0, 1_000_000_000.0 / avgNanos); - } - - /** 单个维度的独立 MSPT 采样窗口。只在所属维度 loop 线程写入，读走 {@code synchronized}。 */ - private static final class Sample { - private final long[] work = new long[WINDOW]; - private int index; - private int count; - - synchronized void record(long nanos) { - work[index] = nanos; - index = (index + 1) % WINDOW; - if (count < WINDOW) { - count++; - } - } - - synchronized double tps() { - if (count == 0) { - return 20.0; - } - long sum = 0L; - for (int i = 0; i < count; i++) { - sum += work[i]; - } - return tpsFromAvg(sum / (double) count); - } - } -} \ No newline at end of file +} diff --git a/src/main/java/dev/modzuozhi/core/filter/EntityTickFilter.java b/src/main/java/dev/modzuozhi/core/filter/EntityTickFilter.java index 1683b88..3a86cff 100644 --- a/src/main/java/dev/modzuozhi/core/filter/EntityTickFilter.java +++ b/src/main/java/dev/modzuozhi/core/filter/EntityTickFilter.java @@ -4,6 +4,8 @@ import net.minecraft.world.entity.Entity; import net.minecraft.world.entity.EntityType; import net.minecraft.world.entity.LivingEntity; import net.minecraft.world.entity.player.Player; +import net.minecraft.world.entity.npc.Villager; +import net.neoforged.fml.ModList; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -27,6 +29,10 @@ import java.util.concurrent.ConcurrentHashMap; * 玩家始终由 {@link PlayerList} 单独处理，不参与本并行路径。 */ public final class EntityTickFilter { + // Lithium replaces SectionStorage.storage with its own listening hash map after our + // initializer. Concurrent villager POI searches can rehash/corrupt that non-thread-safe + // map. Keep this proven unsafe workload on its dimension owner, not the subtask pool. + private static final boolean SERIAL_LITHIUM_VILLAGERS = ModList.get().isLoaded("lithium"); /** 显式放行名单：命中即并行（村民已验证，作为兜底显式保留）。 */ private static final Set WHITELIST = ConcurrentHashMap.newKeySet(); @@ -64,6 +70,9 @@ public final class EntityTickFilter { /** 判断实体是否可并行 tick。 */ public static boolean isParallel(Entity entity) { + if (SERIAL_LITHIUM_VILLAGERS && entity instanceof Villager) { + return false; + } // 快路径：非生物或玩家一定不并行（黑名单中的末影水晶/TNT矿车非生物，白名单只有村民）。 if (!(entity instanceof LivingEntity) || entity instanceof Player) { return false; diff --git a/src/main/java/dev/modzuozhi/mixin/ChunkMapExecutorAccessor.java b/src/main/java/dev/modzuozhi/mixin/ChunkMapExecutorAccessor.java new file mode 100644 index 0000000..5c60ebf --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/ChunkMapExecutorAccessor.java @@ -0,0 +1,12 @@ +package dev.modzuozhi.mixin; + +import net.minecraft.server.level.ChunkMap; +import net.minecraft.util.thread.BlockableEventLoop; +import org.spongepowered.asm.mixin.Mixin; +import org.spongepowered.asm.mixin.gen.Accessor; + +@Mixin(ChunkMap.class) +public interface ChunkMapExecutorAccessor { + @Accessor("mainThreadExecutor") + BlockableEventLoop modzuozhi_chunkExecutor(); +} diff --git a/src/main/java/dev/modzuozhi/mixin/ChunkMapLoadOffloadMixin.java b/src/main/java/dev/modzuozhi/mixin/ChunkMapLoadOffloadMixin.java index 2e055e2..4f25981 100644 --- a/src/main/java/dev/modzuozhi/mixin/ChunkMapLoadOffloadMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/ChunkMapLoadOffloadMixin.java @@ -32,7 +32,10 @@ import java.util.concurrent.Executor; public abstract class ChunkMapLoadOffloadMixin { /** 反序列化下放开关（与 {@link ModZuozhi#isEnabled()} 叠加，可一键回退）。Mixin 禁止非私有静态字段。 */ - private static volatile boolean ENABLED = true; + // ChunkSerializer.read also mutates POI storage. A concurrent replacement of one vanilla + // map does not protect Lithium's listening map or its column indexes. Keep deserialization + // on the chunk executor until all these side effects have an explicit ownership protocol. + private static volatile boolean ENABLED = false; @Shadow @Final @@ -55,4 +58,4 @@ public abstract class ChunkMapLoadOffloadMixin { private Executor modzuozhi$deserializeOffMainThread(Executor executor) { return (ModZuozhi.isEnabled() && ENABLED) ? Util.backgroundExecutor() : executor; } -} \ No newline at end of file +} diff --git a/src/main/java/dev/modzuozhi/mixin/CreateConfigurationPacketMixin.java b/src/main/java/dev/modzuozhi/mixin/CreateConfigurationPacketMixin.java new file mode 100644 index 0000000..799050a --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/CreateConfigurationPacketMixin.java @@ -0,0 +1,22 @@ +package dev.modzuozhi.mixin; + +import com.llamalad7.mixinextras.injector.wrapmethod.WrapMethod; +import com.llamalad7.mixinextras.injector.wrapoperation.Operation; +import dev.modzuozhi.core.compat.SableAssemblyDispatch; +import net.minecraft.server.level.ServerPlayer; +import org.spongepowered.asm.mixin.Mixin; +import org.spongepowered.asm.mixin.Pseudo; + +/** Includes ValueSettingsPacket (rotation speed); preserve Create's range/permission checks. */ +@Pseudo +@Mixin(targets = "com.simibubi.create.foundation.networking.BlockEntityConfigurationPacket", remap = false) +public abstract class CreateConfigurationPacketMixin { + @WrapMethod(method = "handle(Lnet/minecraft/server/level/ServerPlayer;)V") + private void dimensionalripper$configWithOwnership(ServerPlayer player, Operation original) { + if (player == null) { + original.call(player); + return; + } + SableAssemblyDispatch.dispatch(player, () -> original.call(player)); + } +} diff --git a/src/main/java/dev/modzuozhi/mixin/DimensionalRipperMixinPlugin.java b/src/main/java/dev/modzuozhi/mixin/DimensionalRipperMixinPlugin.java new file mode 100644 index 0000000..9d68b73 --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/DimensionalRipperMixinPlugin.java @@ -0,0 +1,61 @@ +package dev.modzuozhi.mixin; + +import net.neoforged.fml.loading.FMLLoader; +import org.objectweb.asm.tree.ClassNode; +import org.spongepowered.asm.mixin.extensibility.IMixinConfigPlugin; +import org.spongepowered.asm.mixin.extensibility.IMixinInfo; + +import java.util.List; +import java.util.Set; + +/** Selects compatibility-safe mixins before Minecraft target classes are transformed. */ +public final class DimensionalRipperMixinPlugin implements IMixinConfigPlugin { + private boolean lithiumLoaded; + + @Override + public void onLoad(String mixinPackage) { + try { + var loadingMods = FMLLoader.getLoadingModList(); + this.lithiumLoaded = loadingMods != null && loadingMods.getModFileById("lithium") != null; + } catch (Throwable ignored) { + // If loader discovery is unavailable, preserve the standalone behavior. + this.lithiumLoaded = false; + } + } + + @Override + public boolean shouldApplyMixin(String targetClassName, String mixinClassName) { + if (mixinClassName.endsWith(".LithiumEntityGroupsMixin")) { + return this.lithiumLoaded; + } + // Lithium replaces Level#getChunk(int, int) wholesale. Injecting into that merged method is + // forbidden by Mixin and aborts startup, while Lithium already owns this access path. + if (!this.lithiumLoaded) { + return true; + } + return !mixinClassName.endsWith(".LevelGetChunkMixin") + && !mixinClassName.endsWith(".ServerChunkCacheGetChunkFallbackMixin"); + } + + @Override + public String getRefMapperConfig() { + return null; + } + + @Override + public void acceptTargets(Set myTargets, Set otherTargets) { + } + + @Override + public List getMixins() { + return null; + } + + @Override + public void preApply(String targetClassName, ClassNode targetClass, String mixinClassName, IMixinInfo mixinInfo) { + } + + @Override + public void postApply(String targetClassName, ClassNode targetClass, String mixinClassName, IMixinInfo mixinInfo) { + } +} diff --git a/src/main/java/dev/modzuozhi/mixin/EntityGetEntityCollisionsMixin.java b/src/main/java/dev/modzuozhi/mixin/EntityGetEntityCollisionsMixin.java index ca57826..5b59ace 100644 --- a/src/main/java/dev/modzuozhi/mixin/EntityGetEntityCollisionsMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/EntityGetEntityCollisionsMixin.java @@ -32,11 +32,15 @@ import java.util.List; * 并发化实体存储后，碰撞开启也能并行跑，不再需要降频拐杖。本特性保留为可选开关（默认关闭 * 碰撞以获取最大 TPS；需要生物交互时开碰撞）。 */ -@Mixin(Entity.class) +// Give optimization mods such as Lithium first choice of this hot invocation. If another mixin +// redirects it, our optional redirect is skipped instead of making that mod's required injection +// fail and aborting game startup. The independent push cancellation below still applies. +@Mixin(value = Entity.class, priority = 900) public abstract class EntityGetEntityCollisionsMixin { @Redirect(method = "collide(Lnet/minecraft/world/phys/Vec3;)Lnet/minecraft/world/phys/Vec3;", at = @At(value = "INVOKE", - target = "Lnet/minecraft/world/level/Level;getEntityCollisions(Lnet/minecraft/world/entity/Entity;Lnet/minecraft/world/phys/AABB;)Ljava/util/List;")) + target = "Lnet/minecraft/world/level/Level;getEntityCollisions(Lnet/minecraft/world/entity/Entity;Lnet/minecraft/world/phys/AABB;)Ljava/util/List;"), + require = 0) private List modzuozhi_noMobCollision(Level level, @Nullable Entity entity, AABB aabb) { // 只受 nocollide 开关控制，不随全局开关失效：全局 /dimensionalripper off 后若恢复真实碰撞， // 实体密集场景会回归 O(n²) 碰撞热点，主线程被拖到几乎无响应（压测巨量实体时实测卡死）。 diff --git a/src/main/java/dev/modzuozhi/mixin/EntitySectionMixin.java b/src/main/java/dev/modzuozhi/mixin/EntitySectionMixin.java index ebed5d4..73f9a95 100644 --- a/src/main/java/dev/modzuozhi/mixin/EntitySectionMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/EntitySectionMixin.java @@ -6,6 +6,8 @@ import dev.modzuozhi.core.dimthread.EntityTickParallel; import dev.modzuozhi.core.dimthread.ISectionKeyAccessor; import java.util.stream.Collectors; import java.util.stream.Stream; +import java.util.ArrayList; +import java.util.List; import net.minecraft.util.AbortableIterationConsumer; import net.minecraft.world.level.entity.EntityAccess; import net.minecraft.world.level.entity.EntitySection; @@ -75,27 +77,46 @@ public abstract class EntitySectionMixin implements ISectionKeyAccessor { AABB aabb, AbortableIterationConsumer consumer, Operation original ) { - // 内容遍历/查询：分片读锁（共享，传感器/碰撞查询并发执行） + // Never invoke arbitrary mod callbacks under a read lock: a callback can move/remove + // an entity, attempting to upgrade this lock to a write lock and deadlocking itself. + List matches = new ArrayList<>(); + AbortableIterationConsumer.Continuation result; EntityTickParallel.lockSectionRead(this.modzuozhi_sectionKey); try { - return original.call(aabb, consumer); + result = original.call(aabb, (AbortableIterationConsumer) entity -> { + matches.add(entity); + return AbortableIterationConsumer.Continuation.CONTINUE; + }); } finally { EntityTickParallel.unlockSectionRead(this.modzuozhi_sectionKey); } + for (EntityAccess entity : matches) { + if (consumer.accept(entity).shouldAbort()) return AbortableIterationConsumer.Continuation.ABORT; + } + return result; } @WrapMethod(method = "getEntities(Lnet/minecraft/world/level/entity/EntityTypeTest;Lnet/minecraft/world/phys/AABB;Lnet/minecraft/util/AbortableIterationConsumer;)Lnet/minecraft/util/AbortableIterationConsumer$Continuation;") - private AbortableIterationConsumer.Continuation modzuozhi_lockedGetEntitiesTyped( - EntityTypeTest typeTest, AABB aabb, - AbortableIterationConsumer consumer, + private AbortableIterationConsumer.Continuation modzuozhi_lockedGetEntitiesTyped( + EntityTypeTest typeTest, AABB aabb, + AbortableIterationConsumer consumer, Operation original ) { + List matches = new ArrayList<>(); + AbortableIterationConsumer.Continuation result; EntityTickParallel.lockSectionRead(this.modzuozhi_sectionKey); try { - return original.call(typeTest, aabb, consumer); + result = original.call(typeTest, aabb, (AbortableIterationConsumer) entity -> { + matches.add(entity); + return AbortableIterationConsumer.Continuation.CONTINUE; + }); } finally { EntityTickParallel.unlockSectionRead(this.modzuozhi_sectionKey); } + for (U entity : matches) { + if (consumer.accept(entity).shouldAbort()) return AbortableIterationConsumer.Continuation.ABORT; + } + return result; } /** diff --git a/src/main/java/dev/modzuozhi/mixin/EntityTickParallelMixin.java b/src/main/java/dev/modzuozhi/mixin/EntityTickParallelMixin.java index b4f2356..89b4c2e 100644 --- a/src/main/java/dev/modzuozhi/mixin/EntityTickParallelMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/EntityTickParallelMixin.java @@ -40,15 +40,16 @@ public abstract class EntityTickParallelMixin { // 非并行窗口或非白名单实体（含玩家）：走原版串行（guardEntityTick 原文） return; } - // 首次启动 / 区块生成高峰：实体所在区块尚未 FULL 时跳过本 tick。 - // 原因：并行子任务线程对未生成区块做碰撞检测会拿到空气占位（见 ServerChunkCacheMixin - // 的非阻塞占位设计，为避免 join 生成死锁），村民会被持续下坠拉穿地面遁地；区块 FULL - // 后下一 tick 恢复正常。此处既不收集并行、也不走串行（串行在 worker 上 join 等生成 - // 会死锁），直接跳过是唯一安全选择。 + // Retain the existing centre-chunk readiness gate. Never invent terrain for an + // entity whose actual chunk is not ready. if (!modzuozhi_entityChunkReady((ServerLevel) (Object) this, entity)) { ci.cancel(); return; } + // The solid-placeholder workaround is gone. Keep entities whose swept collision + // bounds touch missing terrain on the owner, before any collected batch starts. + // The owner can drive chunk work; a subtask cannot while its owner awaits it. + if (!modzuozhi_collisionChunksReady((ServerLevel) (Object) this, entity)) return; MinecraftServer server = ((ServerLevel) (Object) this).getServer(); // 收集到本 tick 的实体批，循环结束统一按块提交并行（减少逐实体提交的调度/屏障开销） EntityTickParallel.collect(server, consumer, entity); @@ -61,7 +62,29 @@ public abstract class EntityTickParallelMixin { */ @Unique private static boolean modzuozhi_entityChunkReady(ServerLevel level, Entity entity) { - long key = ChunkPos.asLong(entity.getBlockX() >> 4, entity.getBlockZ() >> 4); + return modzuozhi_chunkReady(level, entity.getBlockX() >> 4, entity.getBlockZ() >> 4); + } + + @Unique + private static boolean modzuozhi_collisionChunksReady(ServerLevel level, Entity entity) { + var bounds = entity.getBoundingBox().expandTowards(entity.getDeltaMovement()).inflate(1.0); + int minX = net.minecraft.util.Mth.floor(bounds.minX) >> 4; + int maxX = net.minecraft.util.Mth.floor(bounds.maxX) >> 4; + int minZ = net.minecraft.util.Mth.floor(bounds.minZ) >> 4; + int maxZ = net.minecraft.util.Mth.floor(bounds.maxZ) >> 4; + // Avoid a huge readiness scan for unusual motion/size supplied by other mods. + if ((long) maxX - minX > 8 || (long) maxZ - minZ > 8) return false; + for (int x = minX; x <= maxX; x++) { + for (int z = minZ; z <= maxZ; z++) { + if (!modzuozhi_chunkReady(level, x, z)) return false; + } + } + return true; + } + + @Unique + private static boolean modzuozhi_chunkReady(ServerLevel level, int x, int z) { + long key = ChunkPos.asLong(x, z); try { ServerChunkCache cache = level.getChunkSource(); ChunkHolder holder = ((ServerChunkCacheAccessor) (Object) cache) @@ -75,4 +98,4 @@ public abstract class EntityTickParallelMixin { return false; // 保守：拿不到就跳过本 tick } } -} \ No newline at end of file +} diff --git a/src/main/java/dev/modzuozhi/mixin/IntegratedServerPauseMixin.java b/src/main/java/dev/modzuozhi/mixin/IntegratedServerPauseMixin.java new file mode 100644 index 0000000..8bcdbd3 --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/IntegratedServerPauseMixin.java @@ -0,0 +1,23 @@ +package dev.modzuozhi.mixin; + +import com.llamalad7.mixinextras.injector.wrapoperation.Operation; +import com.llamalad7.mixinextras.injector.wrapoperation.WrapOperation; +import dev.modzuozhi.core.compat.PauseOwnership; +import dev.modzuozhi.core.dimthread.DimThreadCore; +import net.minecraft.client.Minecraft; +import net.minecraft.client.server.IntegratedServer; +import org.spongepowered.asm.mixin.Mixin; +import org.spongepowered.asm.mixin.injection.At; + +/** Apply pause and ownership on the server, before vanilla's pause-triggered save. */ +@Mixin(IntegratedServer.class) +public abstract class IntegratedServerPauseMixin { + @WrapOperation(method = "tickServer", at = @At(value = "INVOKE", + target = "Lnet/minecraft/client/Minecraft;isPaused()Z")) + private boolean modzuozhi_syncPauseState(Minecraft client, Operation original) { + boolean paused = PauseOwnership.shouldPause(original.call(client), client.getConnection() != null, + client.level != null, client.player != null); + DimThreadCore.MANAGER.setPaused((IntegratedServer) (Object) this, paused); + return paused; + } +} diff --git a/src/main/java/dev/modzuozhi/mixin/LevelTicksMixin.java b/src/main/java/dev/modzuozhi/mixin/LevelTicksMixin.java index 662e3e1..3383ba4 100644 --- a/src/main/java/dev/modzuozhi/mixin/LevelTicksMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/LevelTicksMixin.java @@ -8,6 +8,19 @@ import net.minecraft.core.BlockPos; import net.minecraft.world.ticks.LevelTicks; import net.minecraft.world.ticks.ScheduledTick; import org.spongepowered.asm.mixin.Mixin; +import org.spongepowered.asm.mixin.Shadow; +import org.spongepowered.asm.mixin.Unique; +import org.spongepowered.asm.mixin.Final; +import org.spongepowered.asm.mixin.injection.At; +import org.spongepowered.asm.mixin.injection.Inject; +import org.spongepowered.asm.mixin.injection.callback.CallbackInfo; +import dev.modzuozhi.core.compat.ScheduledTickDiagnostics; +import it.unimi.dsi.fastutil.longs.Long2ObjectMap; +import it.unimi.dsi.fastutil.longs.Long2LongMap; +import net.minecraft.world.ticks.LevelChunkTicks; +import net.minecraft.world.level.ChunkPos; +import java.util.function.LongPredicate; +import java.util.function.BiConsumer; /** * 第6步：区块调度（{@code LevelTicks}）并发安全——并行子任务中的 tick 调度写入延迟到 @@ -31,7 +44,30 @@ import org.spongepowered.asm.mixin.Mixin; * 且读多写少、正确性可容忍偶发旧值，不做延迟（保持原样）。 */ @Mixin(LevelTicks.class) -public abstract class LevelTicksMixin { +public abstract class LevelTicksMixin implements ScheduledTickDiagnostics { + @Shadow @Final private Long2ObjectMap> allContainers; + @Shadow @Final private Long2LongMap nextTickForContainer; + @Shadow @Final private LongPredicate tickCheck; + @Unique private long dimensionalripper$lastTick = Long.MIN_VALUE; + + @Inject(method = "tick", at = @At("HEAD")) + private void dimensionalripper$recordTick(long time, int limit, BiConsumer consumer, CallbackInfo ci) { + dimensionalripper$lastTick = time; + } + + @Override + public String dimensionalripper$describeTicks(BlockPos pos, long gameTime) { + long key = ChunkPos.asLong(pos); + var container = allContainers.get(key); + var first = container == null ? null : container.peek(); + return "最近调度时间=" + dimensionalripper$lastTick + + "，完整执行条件=" + tickCheck.test(key) + + "，区块容器=" + (container != null) + + "，区块待执行=" + (container == null ? 0 : container.count()) + + "，索引时间=" + (nextTickForContainer.containsKey(key) ? nextTickForContainer.get(key) : "无") + + "，队首=" + (first == null ? "无" : first.type() + "@" + first.pos().toShortString() + + " 剩余刻=" + (first.triggerTick() - gameTime)); + } @WrapMethod(method = "schedule") private void modzuozhi_deferSchedule(ScheduledTick tick, Operation original) { diff --git a/src/main/java/dev/modzuozhi/mixin/LithiumEntityGroupsMixin.java b/src/main/java/dev/modzuozhi/mixin/LithiumEntityGroupsMixin.java new file mode 100644 index 0000000..0c6ae94 --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/LithiumEntityGroupsMixin.java @@ -0,0 +1,36 @@ +package dev.modzuozhi.mixin; + +import com.llamalad7.mixinextras.injector.wrapmethod.WrapMethod; +import com.llamalad7.mixinextras.injector.wrapoperation.Operation; +import net.minecraft.util.ClassInstanceMultiMap; +import org.spongepowered.asm.mixin.Mixin; +import org.spongepowered.asm.mixin.injection.Coerce; + +import java.util.ArrayList; +import java.util.Collection; + +/** Lithium collision queries bypass EntitySection locks and lazily mutate group caches. */ +@Mixin(value = ClassInstanceMultiMap.class, priority = 900) +public abstract class LithiumEntityGroupsMixin { + @WrapMethod(method = "add") + private boolean modzuozhi_add(Object entity, Operation original) { + synchronized (this) { return original.call(entity); } + } + + @WrapMethod(method = "remove") + private boolean modzuozhi_remove(Object entity, Operation original) { + synchronized (this) { return original.call(entity); } + } + + @WrapMethod(method = "find") + private Collection modzuozhi_find(Class type, Operation> original) { + synchronized (this) { return new ArrayList<>(original.call(type)); } + } + + @WrapMethod(method = "lithium$getAllOfGroupType", remap = false) + private Collection modzuozhi_groupSnapshot(@Coerce Object group, Operation> original) { + // Share the exact same lock with add/remove. Never return the live linked hash set: + // callers iterate it after returning, concurrently with entity movement. + synchronized (this) { return new ArrayList<>(original.call(group)); } + } +} diff --git a/src/main/java/dev/modzuozhi/mixin/MinecraftServerSaveMixin.java b/src/main/java/dev/modzuozhi/mixin/MinecraftServerSaveMixin.java index 22adaf4..c5a4083 100644 --- a/src/main/java/dev/modzuozhi/mixin/MinecraftServerSaveMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/MinecraftServerSaveMixin.java @@ -56,7 +56,11 @@ public abstract class MinecraftServerSaveMixin { // 保存前先排空细粒度子任务，避免残留子任务与实体/区块保存并发访问（残留任务此刻 // 若卡死，Barrier 超时降级已在维度 loop 侧触发，这里再给一个短等待兜底）。 - FineGrainScheduler.awaitIdle(2000L); + if (!FineGrainScheduler.awaitIdle(2000L)) { + ModZuozhi.LOGGER.error("[DimThread] 子任务未结束，本次保存取消，避免与仍运行的任务并发写入"); + cir.setReturnValue(false); + return; + } // 1. 并行保存每个维度的区块/实体（独立写盘） List levels = new ArrayList<>(); @@ -107,4 +111,4 @@ public abstract class MinecraftServerSaveMixin { cir.setReturnValue(true); } -} \ No newline at end of file +} diff --git a/src/main/java/dev/modzuozhi/mixin/MixinMinecraftPauseDetector.java b/src/main/java/dev/modzuozhi/mixin/MixinMinecraftPauseDetector.java deleted file mode 100644 index b91f61d..0000000 --- a/src/main/java/dev/modzuozhi/mixin/MixinMinecraftPauseDetector.java +++ /dev/null @@ -1,30 +0,0 @@ -package dev.modzuozhi.mixin; - -import dev.modzuozhi.core.dimthread.DimThreadCore; -import net.minecraft.client.Minecraft; -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.CallbackInfo; - -/** - * 客户端侧：每帧检测单机内嵌服务器的暂停状态（ESC 菜单「暂停并保存」）。 - *

- * 修复历史 bug：单机暂停只冻结主线程，Dimensional Ripper 的维度 worker 不受影响仍在并行 - * 写入 {@code LevelChunk}，与主线程的存盘/区块数据包序列化并发 → 客户端收到残缺区块包解码 - * 越界（"网络协议错误"）掉线（ESC 暂停后 0.5s 内必现）。这里在暂停开始的帧内把维度 loop - * 一并冻结（见 {@link DimThreadCore#setPaused(boolean)}），恢复时自动重排。 - *

    - *
  • LAN 开放时 {@code Minecraft.isPaused()} 恒为 false，不会误暂停；
    • - *
  • dedicated server 无客户端，本 mixin 仅注册于 {@code client} 段，不会被加载；
    • - *
  • {@code setPaused} 内部同值短路，每帧一次布尔比较开销可忽略。
    • - *
- */ -@Mixin(Minecraft.class) -public abstract class MixinMinecraftPauseDetector { - - @Inject(method = "runTick", at = @At("HEAD")) - private void modzuozhi$syncPauseState(CallbackInfo ci) { - DimThreadCore.setPaused(((Minecraft) (Object) this).isPaused()); - } -} \ No newline at end of file diff --git a/src/main/java/dev/modzuozhi/mixin/SableSubLevelContainerMixin.java b/src/main/java/dev/modzuozhi/mixin/SableSubLevelContainerMixin.java new file mode 100644 index 0000000..da5ad37 --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/SableSubLevelContainerMixin.java @@ -0,0 +1,25 @@ +package dev.modzuozhi.mixin; + +import com.llamalad7.mixinextras.injector.wrapmethod.WrapMethod; +import com.llamalad7.mixinextras.injector.wrapoperation.Operation; +import dev.modzuozhi.core.compat.SableMainThreadBridge; +import org.spongepowered.asm.mixin.Mixin; +import org.spongepowered.asm.mixin.Pseudo; + +/** + * Optional Create: Aeronautics / Sable compatibility hook. + * + *

Sable calls {@code ServerSubLevelContainer.tick()} from the HEAD of + * {@code ServerLevel.tick()}. When the parent level is running on a Dimensional Ripper worker, only + * this main-thread-bound physics segment is routed to the server thread. The worker blocks until it + * completes, so Sable never races the rest of its parent level tick.

+ */ +@Pseudo +@Mixin(targets = "dev.ryanhcode.sable.api.sublevel.ServerSubLevelContainer", remap = false) +public abstract class SableSubLevelContainerMixin { + + @WrapMethod(method = "tick") + private void dimensionalripper$runSableTickOnServerThread(Operation original) { + SableMainThreadBridge.run(this, original::call); + } +} diff --git a/src/main/java/dev/modzuozhi/mixin/ServerChunkCacheMixin.java b/src/main/java/dev/modzuozhi/mixin/ServerChunkCacheMixin.java index 2877854..ae35d77 100644 --- a/src/main/java/dev/modzuozhi/mixin/ServerChunkCacheMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/ServerChunkCacheMixin.java @@ -12,21 +12,13 @@ import dev.modzuozhi.core.dimthread.FineGrainScheduler; import dev.modzuozhi.core.dimthread.IChunkEnvTick; import dev.modzuozhi.core.dimthread.IMutableMainThread; import dev.modzuozhi.core.dimthread.ShardGate; -import net.minecraft.core.BlockPos; -import net.minecraft.core.Holder; -import net.minecraft.core.registries.Registries; import net.minecraft.server.MinecraftServer; import net.minecraft.server.level.ChunkHolder; import net.minecraft.server.level.ServerChunkCache; import net.minecraft.server.level.ServerLevel; import net.minecraft.util.RandomSource; import net.minecraft.world.level.ChunkPos; -import net.minecraft.world.level.biome.Biome; -import net.minecraft.world.level.biome.Biomes; -import net.minecraft.world.level.block.Blocks; -import net.minecraft.world.level.block.state.BlockState; import net.minecraft.world.level.chunk.ChunkAccess; -import net.minecraft.world.level.chunk.EmptyLevelChunk; import net.minecraft.world.level.chunk.LevelChunk; import net.minecraft.world.level.chunk.status.ChunkStatus; import org.spongepowered.asm.mixin.Final; @@ -60,6 +52,12 @@ import java.util.function.Consumer; */ @Mixin(value = ServerChunkCache.class, priority = 1001) public abstract class ServerChunkCacheMixin implements IMutableMainThread { + /** + * The real server thread can request chunks while a dimension worker owns this cache. + * Vanilla queues that request to the chunk executor and joins without pumping server tasks. + * A worker awaiting a Sable/player main-thread phase then cannot reach the queued request. + * Pump only the server event loop here; chunk state remains owned by the dimension worker. + */ @Shadow public Thread mainThread; @Shadow @@ -129,62 +127,48 @@ public abstract class ServerChunkCacheMixin implements IMutableMainThread { /** * 子任务线程的 {@code getChunk}：绕过 lastChunk 缓存，直接读正确坐标的 FULL 区块。 *

- * 第10步（修复撕裂回退 + 修复死锁）： + * 非阻塞读取契约： *

    *
  • FULL 命中：走 {@link #modzuozhi_subtaskCacheLookup}，绝不用共享 4 槽 lastChunk 缓存 * （多线程并发写会撕裂，拿到其它坐标区块 → 碰撞把地面读成空气 → 村民遁地）。
    • - *
  • FULL 未命中（区块尚未 FULL，多为加载/生成中）：非阻塞占位返回，绝不能等待—— - * 区块生成管线由维度 worker 的 {@code chunkSource.tick} / 服务器主线程驱动，而这两者此刻 - * 分别卡在「实体屏障 await」和「维度 awaitCompletion」上；等待生成 future 会形成 - * 循环等待 → 整服死锁（日志：[FaultGuard] 维度 tick 超时 >30000ms，玩家侧表现为游戏冻结）。 - * 返回占位块后，区块生成完的下一 tick 命中缓存、碰撞恢复正常，短暂下坠自愈，绝不冻结。
    • + *
  • 未命中：required=false 返回 null；required=true 明确失败并关闭后续细粒度并行。 + * 禁止创建假区块，也不能等待正在屏障处等待本子任务的维度线程生成区块。
    • *
*/ @WrapMethod(method = "getChunk(IILnet/minecraft/world/level/chunk/status/ChunkStatus;Z)Lnet/minecraft/world/level/chunk/ChunkAccess;") private ChunkAccess modzuozhi_subtaskSafeGetChunk( int x, int z, ChunkStatus status, boolean required, Operation original) { - if (!FineGrainScheduler.inSubTask()) { - return original.call(x, z, status, required); + MinecraftServer server = this.level.getServer(); + // mainThread changes between ticks: checking it here and then calling vanilla is + // a TOCTOU race. Route ALL external server requests once the level loop is scheduled. + // Only a synchronous Sable/player handoff guarantees that this worker is paused. + if (server.isSameThread() && DimThreadCore.MANAGER.isLoopScheduled(this.level) + && !dev.modzuozhi.core.compat.SableMainThreadBridge.ownsHandoff(this.level)) { + ServerChunkCache self = (ServerChunkCache) (Object) this; + var executor = ((ChunkMapExecutorAccessor) self.chunkMap).modzuozhi_chunkExecutor(); + return dev.modzuozhi.core.compat.ManagedFutureAwait.callQueued( + () -> original.call(x, z, status, required), executor::tell, server::managedBlock); } - if (status != ChunkStatus.FULL) { + if (!FineGrainScheduler.inSubTask()) { return original.call(x, z, status, required); } LevelChunk full = modzuozhi_subtaskCacheLookup(x, z); if (full != null) { return full; } - // 未 FULL：统一返回实心占位（石头），不再区分 required。 - // 根因：并行子任务线程对未生成区块的碰撞检测（required=false → 原返回 null → 碰撞整体 - // 跳过）与寻路/方块读取（required=true → 原返回空气占位）都把未生成区块当作"空气/无地面"， - // 村民会持续下坠穿入未生成区块 → 加载高峰期遁地（已实测：fine off 串行时 getChunk 会 - // join 同步生成拿真实方块故不遁地，仅并行路径有此问题）。 - // 实心占位让未生成区块表现为"实心墙"：村民被挡在生成边界，区块 FULL 后恢复正常。 - // 仍保持非阻塞（不等待生成 future，避免维度 worker 被 barrier 卡住的循环等待死锁）。 - return modzuozhi_solidChunk(x, z); - } - - /** 构造一个所有方块读作石头的实心占位区块（非阻塞、线程安全，避免生成等待死锁）。 */ - @Unique - private LevelChunk modzuozhi_solidChunk(int x, int z) { - Holder plains = this.level.registryAccess() - .registryOrThrow(Registries.BIOME).getHolderOrThrow(Biomes.PLAINS); - return new ModZuozhiSolidChunk(this.level, new ChunkPos(x, z), plains); - } - - /** - * 未 FULL 区块的实心占位：重写 {@code getBlockState} 恒返回石头。 - * 并行子任务线程把未生成区块表现为实心墙，防止实体下坠穿入（遁地）。 - */ - @Unique - private static final class ModZuozhiSolidChunk extends EmptyLevelChunk { - private ModZuozhiSolidChunk(ServerLevel level, ChunkPos pos, Holder biome) { - super(level, pos, biome); - } - - @Override - public BlockState getBlockState(BlockPos pos) { - return Blocks.STONE.defaultBlockState(); + if (status != ChunkStatus.FULL) { + ChunkHolder holder = ((ServerChunkCacheAccessor) (Object) this) + .modzuozhi_getVisibleChunkIfPresent(ChunkPos.asLong(x, z)); + ChunkAccess present = holder == null ? null : holder.getChunkIfPresent(status); + if (present != null) return present; } + // Optional probes must report absence, not manufacture a "loaded" solid chunk. + // In particular Sable collision guards can probe the same missing chunk per block. + if (!required) return null; + // The owner is waiting at our barrier: synchronous generation here would deadlock. + String detail = this.level.dimension().location() + " chunk=" + x + "," + z + " status=" + status; + FaultGuard.onUnsafeChunkRequest(detail); + throw new IllegalStateException("Parallel task requested synchronous chunk generation: " + detail); } /** @@ -336,7 +320,7 @@ public abstract class ServerChunkCacheMixin implements IMutableMainThread { return; } MinecraftServer server = this.level.getServer(); - this.modzuozhi_chunkBarrier = FineGrainScheduler.beginBarrier(); + this.modzuozhi_chunkBarrier = FineGrainScheduler.beginBarrier("chunk-environment"); this.modzuozhi_chunkGate = ShardGate.chunkGate(this.level.dimension()); boolean sharded = EntityTickParallel.isSharded(); if (sharded) { diff --git a/src/main/java/dev/modzuozhi/mixin/SimulatedAssemblePacketMixin.java b/src/main/java/dev/modzuozhi/mixin/SimulatedAssemblePacketMixin.java new file mode 100644 index 0000000..b83c7f4 --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/SimulatedAssemblePacketMixin.java @@ -0,0 +1,28 @@ +package dev.modzuozhi.mixin; + +import com.llamalad7.mixinextras.injector.wrapmethod.WrapMethod; +import com.llamalad7.mixinextras.injector.wrapoperation.Operation; +import dev.modzuozhi.core.compat.SableAssemblyDispatch; +import net.minecraft.server.level.ServerPlayer; +import org.spongepowered.asm.mixin.Mixin; +import org.spongepowered.asm.mixin.Pseudo; +import org.spongepowered.asm.mixin.injection.Coerce; + +/** Keep Simulated/Veil optional; intercept before the packet's silent getBlockEntity miss. */ +@Pseudo +@Mixin(targets = "dev.simulated_team.simulated.network.packets.AssemblePacket", remap = false) +public abstract class SimulatedAssemblePacketMixin { + @WrapMethod(method = "handle(Lfoundry/veil/api/network/handler/ServerPacketContext;)V") + private void dimensionalripper$assembleWithOwnership(@Coerce Object context, Operation original) { + final ServerPlayer player; + try { + // Invoke the public API interface, not a potentially non-public context implementation. + Class api = Class.forName("foundry.veil.api.network.handler.ServerPacketContext", false, + context.getClass().getClassLoader()); + player = (ServerPlayer) api.getMethod("player").invoke(context); + } catch (ReflectiveOperationException failure) { + throw new IllegalStateException("Cannot resolve Simulated assembly player", failure); + } + SableAssemblyDispatch.dispatch(player, () -> original.call(context)); + } +} diff --git a/src/main/java/dev/modzuozhi/mixin/TickingBlockEntityParallelMixin.java b/src/main/java/dev/modzuozhi/mixin/TickingBlockEntityParallelMixin.java index b1fedfd..7c907ef 100644 --- a/src/main/java/dev/modzuozhi/mixin/TickingBlockEntityParallelMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/TickingBlockEntityParallelMixin.java @@ -86,7 +86,7 @@ public abstract class TickingBlockEntityParallelMixin { // 每 tick 重建分片数组（16 槽），防止上一 tick 残留数据串拍 this.modzuozhi_teShards = new List[ShardGate.SHARDS]; if (this.modzuozhi_teParallel) { - this.modzuozhi_teBarrier = FineGrainScheduler.beginBarrier(); + this.modzuozhi_teBarrier = FineGrainScheduler.beginBarrier("block-entity"); this.modzuozhi_teGate = ShardGate.teGate(((ServerLevel) (Object) this).dimension()); } } @@ -180,4 +180,4 @@ public abstract class TickingBlockEntityParallelMixin { ci.cancel(); } } -} \ No newline at end of file +} diff --git a/src/main/resources/dimensionalripper.mixins.json b/src/main/resources/dimensionalripper.mixins.json index 58a01e0..5be612f 100644 --- a/src/main/resources/dimensionalripper.mixins.json +++ b/src/main/resources/dimensionalripper.mixins.json @@ -2,6 +2,7 @@ "required": true, "minVersion": "0.8.5", "package": "dev.modzuozhi.mixin", + "plugin": "dev.modzuozhi.mixin.DimensionalRipperMixinPlugin", "compatibilityLevel": "JAVA_17", "mixins": [ "AttributeInstanceMixin", @@ -12,8 +13,10 @@ "BlockDropCaptureMixin", "BrainMixin", "ChunkMapEntityMapMixin", + "ChunkMapExecutorAccessor", "ChunkMapLoadOffloadMixin", "ClassInstanceMultiMapMixin", + "LithiumEntityGroupsMixin", "CollectingNeighborUpdaterMixin", "DistanceManagerMixin", "EntityChangeDimensionMixin", @@ -60,11 +63,14 @@ "ServerLevelPlayerPacketMixin", "ServerPlayerChangeDimensionMixin", "ServerPlayerPrefetchMixin", + "SableSubLevelContainerMixin", + "SimulatedAssemblePacketMixin", + "CreateConfigurationPacketMixin", "TickingBlockEntityParallelMixin", "YieldJobSiteGuardMixin" ], "client": [ - "MixinMinecraftPauseDetector" + "IntegratedServerPauseMixin" ], "server": [ ], diff --git a/tests/ManagedFutureAwaitRegression.java b/tests/ManagedFutureAwaitRegression.java new file mode 100644 index 0000000..4dc9ce6 --- /dev/null +++ b/tests/ManagedFutureAwaitRegression.java @@ -0,0 +1,83 @@ +import dev.modzuozhi.core.compat.ManagedFutureAwait; +import java.util.concurrent.*; + +/** Standalone regression: dimension work needs a callback on the waiting server thread. */ +public final class ManagedFutureAwaitRegression { + public static void main(String[] args) throws Exception { + Thread serverThread = Thread.currentThread(); + BlockingQueue serverTasks = new LinkedBlockingQueue<>(); + CompletableFuture chunk = new CompletableFuture<>(); + Thread worker = new Thread(() -> { + try { + CompletableFuture physics = new CompletableFuture<>(); + serverTasks.add(() -> { + if (Thread.currentThread() != serverThread) { + physics.completeExceptionally(new AssertionError("wrong callback thread")); + } else { + physics.complete(null); + } + }); + physics.get(3, TimeUnit.SECONDS); + chunk.complete(42); + } catch (Throwable failure) { + chunk.completeExceptionally(failure); + } + }, "test-dimension-worker"); + worker.start(); + int value = ManagedFutureAwait.join(chunk, done -> { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (!done.getAsBoolean()) { + if (System.nanoTime() > deadline) throw new AssertionError("circular wait"); + try { + Runnable task = serverTasks.poll(10, TimeUnit.MILLISECONDS); + if (task != null) task.run(); + } catch (InterruptedException failure) { + throw new AssertionError(failure); + } + } + }); + worker.join(1000); + if (value != 42 || worker.isAlive()) throw new AssertionError("request did not finish"); + if (ManagedFutureAwait.join(CompletableFuture.completedFuture(7), done -> { + throw new AssertionError("completed future pumped callbacks"); + }) != 7) throw new AssertionError("wrong result"); + RuntimeException failure = new RuntimeException("expected"); + try { + ManagedFutureAwait.join(CompletableFuture.failedFuture(failure), done -> {}); + throw new AssertionError("exception swallowed"); + } catch (CompletionException expected) { + if (expected.getCause() != failure) throw new AssertionError("exception changed"); + } + CompletableFuture cancelled = new CompletableFuture<>(); + cancelled.cancel(false); + try { + ManagedFutureAwait.join(cancelled, done -> {}); + throw new AssertionError("cancellation swallowed"); + } catch (CancellationException expected) {} + // An apparently server-owned cache can switch owner after the entry check. + // The request must remain queued regardless; an inline-capable execute() is unsafe. + BlockingQueue chunkTasks = new LinkedBlockingQueue<>(); + java.util.concurrent.atomic.AtomicReference owner = + new java.util.concurrent.atomic.AtomicReference<>(serverThread); + java.util.concurrent.atomic.AtomicBoolean executed = new java.util.concurrent.atomic.AtomicBoolean(); + int queued = ManagedFutureAwait.callQueued(() -> { + if (owner.get() != Thread.currentThread()) throw new AssertionError("wrong chunk owner"); + executed.set(true); + return 19; + }, task -> { + chunkTasks.add(task); + if (executed.get()) throw new AssertionError("request executed inline"); + }, done -> { + Thread dimension = new Thread(() -> { + owner.set(Thread.currentThread()); + chunkTasks.remove().run(); + }); + dimension.start(); + try { dimension.join(2000); } catch (InterruptedException e) { throw new AssertionError(e); } + if (!done.getAsBoolean()) throw new AssertionError("queued request stuck"); + }); + if (queued != 19 || !executed.get()) throw new AssertionError("lost queued result"); + System.out.println("PASS: callback handoff, completion, failure and cancellation"); + System.out.println("PASS: unconditional enqueue survives owner switch after entry"); + } +} diff --git a/tests/PauseOwnershipRegression.java b/tests/PauseOwnershipRegression.java new file mode 100644 index 0000000..517550c --- /dev/null +++ b/tests/PauseOwnershipRegression.java @@ -0,0 +1,16 @@ +import dev.modzuozhi.core.compat.PauseOwnership; + +public final class PauseOwnershipRegression { + public static void main(String[] args) { + for (int mask = 0; mask < 16; mask++) { + boolean paused = PauseOwnership.shouldPause((mask & 1) != 0, (mask & 2) != 0, + (mask & 4) != 0, (mask & 8) != 0); + if (paused != (mask == 15)) throw new AssertionError("Loading screen paused worker: " + mask); + } + if (PauseOwnership.workerOwns(false, false, false)) throw new AssertionError("Startup routed to absent worker"); + if (!PauseOwnership.workerOwns(true, false, false)) throw new AssertionError("Running loop lost ownership"); + if (!PauseOwnership.workerOwns(true, true, true)) throw new AssertionError("In-flight tick lost ownership"); + if (PauseOwnership.workerOwns(true, true, false)) throw new AssertionError("Idle paused worker kept requests"); + System.out.println("PASS: loading/login pause gate and drained-pause chunk ownership"); + } +} diff --git a/tests/TaskBarrierRegression.java b/tests/TaskBarrierRegression.java new file mode 100644 index 0000000..a925cce --- /dev/null +++ b/tests/TaskBarrierRegression.java @@ -0,0 +1,65 @@ +import dev.modzuozhi.core.compat.TaskBarrier; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; + +public final class TaskBarrierRegression { + public static void main(String[] args) throws Exception { + // The old one-shot latch is permanently open after the first submission completes. + CountDownLatch old = new CountDownLatch(1); + old.countDown(); + if (!old.await(0, TimeUnit.MILLISECONDS)) throw new AssertionError(); + + TaskBarrier barrier = new TaskBarrier(); + barrier.register(); + barrier.complete(); // Fast task finishes while producer is still submitting. + barrier.register(); // A later task MUST still be waited for. + CountDownLatch warned = new CountDownLatch(1); + AtomicBoolean returned = new AtomicBoolean(); + AtomicBoolean interruptRestored = new AtomicBoolean(); + Thread waiter = new Thread(() -> { + barrier.await(30, warned::countDown); + returned.set(true); + interruptRestored.set(Thread.currentThread().isInterrupted()); + }); + waiter.setDaemon(true); + waiter.start(); + waiter.interrupt(); + if (!warned.await(2, TimeUnit.SECONDS)) throw new AssertionError("No slow-task diagnostic"); + if (returned.get()) throw new AssertionError("Timeout/interruption abandoned a running task"); + barrier.complete(); + waiter.join(2000); + if (waiter.isAlive() || !returned.get() || !interruptRestored.get()) throw new AssertionError("Drain/interrupt restoration failed"); + try { + barrier.register(); + throw new AssertionError("Post-seal submission allowed"); + } catch (IllegalStateException expected) {} + + TaskBarrier empty = new TaskBarrier(); + empty.await(30, () -> { throw new AssertionError("Empty barrier waited"); }); + try { + empty.complete(); + throw new AssertionError("Unbalanced completion allowed"); + } catch (IllegalStateException expected) {} + + ExecutorService workers = Executors.newFixedThreadPool(4); + try { + for (int round = 0; round < 200; round++) { + TaskBarrier batch = new TaskBarrier(); + int[] writes = new int[100]; + for (int i = 0; i < writes.length; i++) { + int index = i; + batch.register(); + workers.execute(() -> { + try { writes[index] = 1; } finally { batch.complete(); } + }); + if ((i & 7) == 0) Thread.yield(); + } + batch.await(2000, () -> { throw new AssertionError("Stress test stalled"); }); + for (int value : writes) if (value != 1) throw new AssertionError("Early return or missing visibility"); + } + } finally { + workers.shutdownNow(); + } + System.out.println("PASS: submission gaps, timeout, interrupt, empty/sealed barrier, 20,000 concurrent completions"); + } +} diff --git a/tests/TickProgressRegression.java b/tests/TickProgressRegression.java new file mode 100644 index 0000000..ffcfa60 --- /dev/null +++ b/tests/TickProgressRegression.java @@ -0,0 +1,36 @@ +import dev.modzuozhi.core.compat.TickProgress; + +public final class TickProgressRegression { + public static void main(String[] args) { + TickProgress p = new TickProgress(); + check(Double.isNaN(p.snapshot(0).tps()), "unobserved must not claim 20 TPS"); + for (int i = 0; i < 200; i++) { + p.begin(i * 50_000_000L); + p.finish(i * 50_000_000L + 10_000_000L, 10_000_000L, true); + } + var normal = p.snapshot(10_000_000_000L); + check(normal.tps() >= 19.8 && normal.tps() <= 20.1, "real completion rate"); + check(normal.averageWorkMs() == 10 && normal.lastWorkMs() == 10, "work MSPT"); + p.begin(10_000_000_000L); + var stuck = p.snapshot(21_000_000_000L); + check(stuck.tps() == 0, "stale samples must decay to zero"); + check(stuck.runningMs() == 11000 && stuck.sinceCompletedMs() > 11000, "live blocked duration"); + p.finish(22_000_000_000L, 12_000_000_000L, false); + var failed = p.snapshot(22_000_000_000L); + check(failed.tps() == 0 && failed.failures() == 1, "failed ticks are not successful completions"); + check(Double.isNaN(failed.runningMs()), "idle has no running tick"); + p.begin(23_000_000_000L); + p.finish(23_010_000_000L, 10_000_000L, true); + check(p.snapshot(23_010_000_000L).sinceCompletedMs() == 0, "resume records fresh progress"); + check(Double.isNaN(new TickProgress().snapshot(24_000_000_000L).tps()), "new world has no stale data"); + TickProgress slow = new TickProgress(); + for (int i = 0; i < 50; i++) { + slow.begin(i * 200_000_000L); + slow.finish(i * 200_000_000L + 1_000_000L, 1_000_000L, true); + } + check(Math.abs(slow.snapshot(10_000_000_000L).tps() - 5) <= 0.1, + "1ms work must not hide 5 TPS scheduling cadence"); + System.out.println("PASS: actual TPS, no data, stalled tick, stale window, failure and resume"); + } + private static void check(boolean ok, String message) { if (!ok) throw new AssertionError(message); } +} diff --git a/tests/integration/ChunkHandoffRegression.java b/tests/integration/ChunkHandoffRegression.java new file mode 100644 index 0000000..98f1bcb --- /dev/null +++ b/tests/integration/ChunkHandoffRegression.java @@ -0,0 +1,42 @@ +package dev.modzuozhi; + +import net.neoforged.bus.api.SubscribeEvent; +import net.neoforged.fml.common.EventBusSubscriber; +import net.neoforged.neoforge.event.tick.ServerTickEvent; +import net.minecraft.world.level.chunk.status.ChunkStatus; +import dev.modzuozhi.core.dimthread.FineGrainScheduler; +import dev.modzuozhi.core.dimthread.DimThreadCore; +import net.minecraft.server.level.ServerPlayer; +import net.minecraft.server.level.ClientInformation; +import com.mojang.authlib.GameProfile; +import java.util.UUID; + +/** Runtime regression fixture; deliberately outside production source sets. */ +@EventBusSubscriber(modid = "dimensionalripper") +public final class ChunkHandoffRegression { + private static int rounds; + @SubscribeEvent + public static void tick(ServerTickEvent.Post event) { + var server = event.getServer(); + if (server.getTickCount() < 10) return; + if (rounds % 10 == 0) { + DimThreadCore.MANAGER.setPaused(server, true); + if (DimThreadCore.MANAGER.isLoopScheduled(server.overworld())) + throw new AssertionError("Paused idle loop still owns chunk queue"); + // Exercise the actual login constructor from the reported waiting stack. + new ServerPlayer(server, server.overworld(), new GameProfile(UUID.randomUUID(), "PauseRegression"), + ClientInformation.createDefault()); + server.overworld().getChunkSource().getChunk(20 + rounds, 0, ChunkStatus.FULL, true); + DimThreadCore.MANAGER.setPaused(server, false); + } + for (int i = 0; i < 100; i++) { + if (server.overworld().getChunkSource().getChunk(0, 0, ChunkStatus.FULL, true) == null) + throw new AssertionError("Missing requested chunk"); + } + if (++rounds == 100) { + if (!FineGrainScheduler.isEnabled()) throw new AssertionError("Test silently degraded"); + ModZuozhi.LOGGER.info("REGRESSION PASS: 10 paused login constructors, paused generation, resume, 10000 running chunk requests; fine enabled"); + server.halt(false); + } + } +} diff --git a/tests/integration/InteractionFlowRegression.java b/tests/integration/InteractionFlowRegression.java new file mode 100644 index 0000000..fd60398 --- /dev/null +++ b/tests/integration/InteractionFlowRegression.java @@ -0,0 +1,171 @@ +package dev.modzuozhi; + +import com.mojang.authlib.GameProfile; +import dev.modzuozhi.core.compat.SableAssemblyDispatch; +import dev.modzuozhi.core.compat.SableMainThreadBridge; +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; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import net.neoforged.bus.api.SubscribeEvent; +import net.neoforged.fml.common.EventBusSubscriber; +import net.neoforged.neoforge.event.tick.ServerTickEvent; +import java.util.UUID; + +/** Opt-in runtime fixture: no real user world or network connection. */ +@EventBusSubscriber(modid = "dimensionalripper") +public final class InteractionFlowRegression { + private static boolean started, completed; + private static long startTime; + private static final BlockPos SOURCE = new BlockPos(8, 100, 8); + @SubscribeEvent + public static void tick(ServerTickEvent.Post event) { + var server = event.getServer(); + var level = server.overworld(); + if (server.getTickCount() < 15) return; + if (!started) { + started = true; + DimThreadCore.MANAGER.setPaused(server, true); + level.setChunkForced(0, 0, true); + level.getChunk(0, 0); + checkMissingChunkProbes(server); + for (int x = 4; x <= 12; x++) for (int z = 4; z <= 12; z++) { + level.setBlockAndUpdate(new BlockPos(x, 99, z), Blocks.STONE.defaultBlockState()); + 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; } + }; + startTime = level.getGameTime(); + DimThreadCore.MANAGER.setPaused(server, false); + SableAssemblyDispatch.dispatch(player, () -> { + if (!server.isSameThread() || !SableMainThreadBridge.ownsHandoff(level)) + 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"); + level.setBlockAndUpdate(SOURCE, Blocks.WATER.defaultBlockState()); + completed = true; + }); + } + if (server.getTickCount() == 180) { + DimThreadCore.MANAGER.setPaused(server, true); + if (!completed) throw new AssertionError("Interaction was not dispatched"); + var progress = DimThreadCore.MANAGER.getLoop(level).progress(System.nanoTime()); + if (!(progress.tps() > 0) || Double.isNaN(progress.averageWorkMs()) + || Double.isNaN(progress.sinceCompletedMs()) || !Double.isNaN(progress.runningMs())) + throw new AssertionError("Invalid drained-loop progress: " + progress); + ModZuozhi.LOGGER.info("TICK PROGRESS TEST: {}", progress); + if (level.getGameTime() <= startTime) throw new AssertionError("World clock stopped"); + String diagnostic = ((dev.modzuozhi.core.compat.ScheduledTickDiagnostics) level.getFluidTicks()) + .dimensionalripper$describeTicks(SOURCE, level.getGameTime()); + if (diagnostic.contains(Long.toString(Long.MIN_VALUE))) throw new AssertionError("Missing tick instrumentation"); + ModZuozhi.LOGGER.info("FLOW DIAGNOSTIC TEST: {}", diagnostic); + if (!level.getFluidState(SOURCE.east()).is(Fluids.FLOWING_WATER)) + throw new AssertionError("Scheduled water tick did not spread: time=" + level.getGameTime() + + " pending=" + level.getFluidTicks().count()); + ModZuozhi.LOGGER.info("INTERACTION REGRESSION PASS: exclusive main-thread block entity lookup and scheduled water spread"); + server.halt(false); + } + } + + /** Run the actual transformed getChunk from two real scheduler subtasks. */ + private static void checkMissingChunkProbes(net.minecraft.server.MinecraftServer server) { + var level = server.overworld(); + var cache = level.getChunkSource(); + var loaded = level.getChunk(0, 0); + int loadedCount = cache.getLoadedChunksCount(); + var failure = new AtomicReference(); + var completed = new AtomicInteger(); + var barrier = FineGrainScheduler.beginBarrier("missing-chunk-regression"); + long began = System.nanoTime(); + for (int task = 0; task < 2; task++) { + barrier.submit(server, () -> { + try { + if (!FineGrainScheduler.inSubTask()) throw new AssertionError("Not a subtask"); + for (int i = 0; i < 100_000; i++) { + // Alternate coordinates: no fake positive or negative cache entries. + int x = 100_000 + (i & 1); + if (cache.getChunk(x, -100_000, ChunkStatus.FULL, false) != null) + throw new AssertionError("Missing FULL probe returned a fabricated chunk"); + if (cache.getChunk(x, -100_000, ChunkStatus.EMPTY, false) != null) + throw new AssertionError("Missing partial-status probe returned a chunk"); + if (cache.getChunkNow(x, -100_000) != null) + throw new AssertionError("Missing getChunkNow returned a chunk"); + if (cache.getChunk(0, 0, ChunkStatus.FULL, false) != loaded + || cache.getChunk(0, 0, ChunkStatus.FULL, true) != loaded) + throw new AssertionError("Loaded chunk identity lost"); + } + completed.incrementAndGet(); + } catch (Throwable error) { + failure.compareAndSet(null, error); + } + }); + } + barrier.await(); + barrier.drain(); + if (failure.get() != null) throw new AssertionError("Missing chunk regression", failure.get()); + if (completed.get() != 2 || !FineGrainScheduler.isEnabled()) + throw new AssertionError("Probes did not complete with parallelism enabled"); + if (cache.getLoadedChunksCount() != loadedCount) + throw new AssertionError("Optional probe loaded chunks"); + var rabbit = new net.minecraft.world.entity.animal.Rabbit(net.minecraft.world.entity.EntityType.RABBIT, level); + rabbit.setPos(8, 101, 8); + var routed = new AtomicInteger(); + dev.modzuozhi.core.dimthread.EntityTickParallel.begin(); + level.guardEntityTick(entity -> { + if (!FineGrainScheduler.inSubTask()) throw new AssertionError("Loaded interior lost parallelism"); + routed.incrementAndGet(); + }, rabbit); + if (routed.get() != 0) throw new AssertionError("Interior entity was not batched"); + dev.modzuozhi.core.dimthread.EntityTickParallel.end(); + if (routed.get() != 1) throw new AssertionError("Interior entity did not run exactly once"); + rabbit.setDeltaMovement(10_000, 0, 0); + dev.modzuozhi.core.dimthread.EntityTickParallel.begin(); + level.guardEntityTick(entity -> { + if (FineGrainScheduler.inSubTask()) throw new AssertionError("Unbounded collision scan on subtask"); + routed.incrementAndGet(); + }, rabbit); + if (routed.get() != 2) throw new AssertionError("Wide movement did not stay on owner"); + dev.modzuozhi.core.dimthread.EntityTickParallel.end(); + if (routed.get() != 2) throw new AssertionError("Owner entity tick duplicated"); + ModZuozhi.LOGGER.info("ENTITY ROUTING REGRESSION PASS: loaded interior parallel, wide movement on owner, exactly once"); + ModZuozhi.LOGGER.info("MISSING CHUNK REGRESSION PASS: 600000 absent reads + 400000 loaded reads, {} ms, fine enabled", + (System.nanoTime() - began) / 1_000_000); + + // A required miss must fail instead of allocating a fake chunk or blocking forever. + var requiredBarrier = FineGrainScheduler.beginBarrier("required-chunk-regression"); + requiredBarrier.submit(server, () -> { + try { + for (var status : new ChunkStatus[]{ChunkStatus.FULL, ChunkStatus.EMPTY}) { + try { + cache.getChunk(100_000, -100_000, status, true); + throw new AssertionError("Required missing chunk was fabricated"); + } catch (IllegalStateException expected) { + if (!expected.getMessage().startsWith("Parallel task requested synchronous chunk generation:")) + throw expected; + } + } + } catch (Throwable error) { + failure.compareAndSet(null, error); + } + }); + requiredBarrier.await(); + requiredBarrier.drain(); + if (failure.get() != null) throw new AssertionError("Required chunk regression", failure.get()); + if (FineGrainScheduler.isEnabled() || !FaultGuard.isFineDegraded() + || cache.getLoadedChunksCount() != loadedCount) + throw new AssertionError("Required miss did not fail safely"); + FaultGuard.reset(); + FineGrainScheduler.setEnabled(true); + ModZuozhi.LOGGER.info("REQUIRED CHUNK REGRESSION PASS: FULL/EMPTY fail fast, no generation; test-only fault reset"); + } +}