From 801deb701a4c6c67b6abd6a0154c84d8494d4ac7 Mon Sep 17 00:00:00 2001 From: modzuozhi Date: Sat, 12 Sep 2026 00:44:23 +0800 Subject: [PATCH] Fix singleplayer disconnect on ESC pause: freeze dimension loops - New client mixin MixinMinecraftPauseDetector: on each Minecraft.runTick, detect integrated-server pause and freeze/unfreeze all dimension loops via DimThreadCore.setPaused, so no worker writes LevelChunk while the main thread is saving/pausing. Fixes reproducible disconnect (packet decode IndexOutOfBounds on ClientboundLevelChunkWithLight) ~0.5s after opening the pause menu. - LevelTickLoop: add pause()/resume() (cancel next tick, stop self-rescheduling, re-schedule on resume) plus a paused guard at run() entry. Resilient to RejectedExecutionException when the master pool is already shut down. - ServerManager.setPaused: pause all loops, then wait up to 5s for the in-flight tick to fully settle before returning. - Keep existing work: TE whitelist command (/dimensionalripper tewhitelist), shard drop counters (entity/TE/chunk-env), POI lock optimization. --- README.md | 4 +- .../dev/modzuozhi/command/MasterCommand.java | 26 +++++- .../core/dimthread/DimThreadCore.java | 15 ++++ .../core/dimthread/EntityTickParallel.java | 18 +++- .../core/dimthread/LevelTickLoop.java | 57 ++++++++++-- .../core/dimthread/ServerManager.java | 43 +++++++++ .../modzuozhi/core/dimthread/ShardGate.java | 87 ++++++++++++++++++- .../modzuozhi/core/filter/SerDesFilter.java | 24 ++++- .../mixin/MixinMinecraftPauseDetector.java | 30 +++++++ .../mixin/ServerChunkCacheMixin.java | 82 +++++++++++++++-- .../TickingBlockEntityParallelMixin.java | 74 +++++++++++++++- .../resources/dimensionalripper.mixins.json | 1 + 12 files changed, 432 insertions(+), 29 deletions(-) create mode 100644 src/main/java/dev/modzuozhi/mixin/MixinMinecraftPauseDetector.java diff --git a/README.md b/README.md index c73fb9b..95ae587 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ Dimensional Ripper 是一个基于 **NeoForge 1.21.1** 的多线程维度并行 - **维度并行 Tick**：每个维度拥有独立的 tick 循环与 worker 线程，维度之间互不拖累——单维度低 TPS 不会影响其他维度。 - **细粒度并行**：在维度内部对实体、方块实体、区块环境 tick 进行并行调度，充分压榨多核性能。 -- **实体分片并行 (ShardGate)**：实体按稳定 ID 分为与 CPU 核数对应的若干片，每片由跨 tick 的 CAS 门控独立提交。单个慢分片自动掉拍延后，不再阻塞整个实体 tick 屏障——把"单点慢"从拖垮全局变成只拖自己。 +- **分片并行 (ShardGate)**：实体/方块实体/区块环境三阶段统一按稳定键分片（实体按 ID、TE 按方块坐标、区块环境按区块坐标），每片由跨 tick 的 CAS 门控独立提交。单个慢分片自动掉拍延后，不再阻塞对应阶段的并行屏障——把"单点慢"从拖垮全局变成只拖自己。 - **POI 并发锁优化**：将村民等实体高频读取的职业兴趣点（POI）存储从全互斥同步容器替换为无锁并发容器，消除并行 tick 下的全局锁瓶颈。约 1.9 万村民压测下 TPS 由 9 提升至 20（16 核）。 - **碰撞优化**：可禁用实体间碰撞，消除 O(n²) 碰撞开销，显著释放 TPS。 - **异步传送门传送**：支持原版下界/末地传送门与模组维度（如暮色森林、以太）传送门的异步预加载与渐进生成，避免传送到未加载区块导致的卡顿、冻结与虚空传送。 @@ -29,7 +29,7 @@ Dimensional Ripper 是一个基于 **NeoForge 1.21.1** 的多线程维度并行 |------|------| | `/dimensionalripper status` | 查看当前状态、TPS、各开关与掉拍分片统计 | | `/dimensionalripper fine on\|off` | 开关维度内细粒度并行（实体/方块实体/区块环境） | -| `/dimensionalripper fine sharded on\|off` | 开关实体分片并行（ShardGate，默认开启） | +| `/dimensionalripper fine sharded on\|off` | 开关分片并行（实体/TE/区块环境，ShardGate，默认开启） | | `/dimensionalripper nocollide on\|off` | 开关实体碰撞优化（默认开启） | 模组**默认启用**多线程与分片并行，开箱即用，无需额外配置。 diff --git a/src/main/java/dev/modzuozhi/command/MasterCommand.java b/src/main/java/dev/modzuozhi/command/MasterCommand.java index 49b1db3..6aa49b3 100644 --- a/src/main/java/dev/modzuozhi/command/MasterCommand.java +++ b/src/main/java/dev/modzuozhi/command/MasterCommand.java @@ -7,8 +7,11 @@ 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; import net.minecraft.network.chat.Component; import net.minecraft.resources.ResourceKey; import net.minecraft.server.level.ServerLevel; @@ -43,7 +46,11 @@ public final class MasterCommand { .then(literal("off").executes(ctx -> setSharded(ctx, false))))) .then(literal("nocollide") .then(literal("on").executes(ctx -> setNoCollide(ctx, true))) - .then(literal("off").executes(ctx -> setNoCollide(ctx, false))))); + .then(literal("off").executes(ctx -> setNoCollide(ctx, false)))) + .then(literal("tewhitelist") + .then(net.minecraft.commands.Commands.argument("type", net.minecraft.commands.arguments.ResourceLocationArgument.id()) + .executes(MasterCommand::teWhitelist))) + .then(literal("status").executes(MasterCommand::status))); } private static int setFine(CommandContext ctx, boolean enabled) { @@ -74,7 +81,7 @@ public final class MasterCommand { private static int setSharded(CommandContext ctx, boolean enabled) { EntityTickParallel.setSharded(enabled); ctx.getSource().sendSuccess(() -> - Component.literal("实体分片并行已" + (enabled ? "开启" : "关闭")), false); + Component.literal("分片并行（实体/TE/区块环境）已" + (enabled ? "开启" : "关闭")), false); return 1; } @@ -86,6 +93,16 @@ public final class MasterCommand { return 1; } + /** 将指定 TE 注册名加入 modded 并行白名单（仅影响该 TE，不改全局 lockModded 保守策略）。 */ + private static int teWhitelist(CommandContext ctx) { + String type = ResourceLocationArgument.getId(ctx, "type").toString(); + SerDesFilter.whitelistModded(type); + ctx.getSource().sendSuccess(() -> + Component.literal("TE 白名单已添加: " + type + "（清单: " + SerDesFilter.moddedWhitelist() + "）"), false); + ModZuozhi.LOGGER.info("[ModZuozhi] TE 白名单新增 {}", type); + return 1; + } + private static int status(CommandContext ctx) { String fine = FineGrainScheduler.isEnabled() ? "开启" : "关闭"; String noCollide = ModZuozhi.MOB_COLLISION_OFF ? "开启" : "关闭"; @@ -104,11 +121,12 @@ public final class MasterCommand { String msg = String.format( "modzuozhi 状态：细粒度=%s%s，分片并行=%s，禁碰撞=%s，故障原因：%s\n" - + "当前维度：%s，TPS：%.1f，掉拍分片累计：%d", + + "当前维度：%s，TPS：%.1f，掉拍分片累计：实体 %d / TE %d / 区块环境 %d", fine, fineMark, EntityTickParallel.isSharded() ? "开启" : "关闭", noCollide, reason, - dimensionName(dimension), tps, EntityTickParallel.missedShards()); + dimensionName(dimension), tps, + EntityTickParallel.missedShards(), ShardGate.teMissed(), ShardGate.chunkMissed()); ctx.getSource().sendSuccess(() -> Component.literal(msg), false); return 1; } diff --git a/src/main/java/dev/modzuozhi/core/dimthread/DimThreadCore.java b/src/main/java/dev/modzuozhi/core/dimthread/DimThreadCore.java index ae91740..86881eb 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/DimThreadCore.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/DimThreadCore.java @@ -60,4 +60,19 @@ public final class DimThreadCore { public static boolean owns(Thread thread) { return thread.getName().startsWith(MOD_ID + "_server_"); } + + /** 查询是否处于单机暂停状态。 */ + public static boolean isPaused() { + return MANAGER.isPaused(); + } + + /** + * 单机内嵌服务器暂停/恢复信号（见 {@link ServerManager#setPaused}）。 + *

+ * 由客户端 mixin {@code MixinMinecraftPauseDetector} 在每帧 {@code Minecraft.runTick} 检测 + * {@code isPaused()} 变化后调用；dedicated server 无客户端，永不会触发。 + */ + public static void setPaused(boolean paused) { + MANAGER.setPaused(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 dfb92ac..4038204 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/EntityTickParallel.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/EntityTickParallel.java @@ -51,7 +51,14 @@ public final class EntityTickParallel { */ private static final ConcurrentHashMap, ShardGate> GATES = new ConcurrentHashMap<>(); - /** 累计掉拍分片数（诊断 / status 展示）。 */ + /** + * 累计掉拍分片数（诊断 / status 展示）。 + *

+ * 观测型指标：{@link #modzuozhi_tickShard} 中分片任务实际耗时 > {@link ShardGate#TICK_BUDGET_NS} + * （一 tick 预算）即计一次"该片本 tick 掉拍"——它没能在一拍时间内处理完本片负载（慢片）。 + * claim 抢占失败路径因阶段收口 {@link FineGrainScheduler.Barrier#await()} 会在本 tick 内等齐 + * 所有已提交分片，结构上不可达，仅保留作为兜底。 + */ private static final LongAdder MISSED_SHARDS = new LongAdder(); /** 实体分片并行的独立开关（默认开启；关闭时回退旧的批式 submitBatch 路径）。 */ @@ -164,10 +171,10 @@ public final class EntityTickParallel { public static void setSharded(boolean value) { sharded = value; - ModZuozhi.LOGGER.info("[EntityTick] 实体分片并行已切换为 {}", value ? "开启" : "关闭"); + ModZuozhi.LOGGER.info("[EntityTick] 分片并行（实体/TE/区块环境）已切换为 {}", value ? "开启" : "关闭"); } - /** 累计掉拍分片数（诊断 / {@code status} 展示）。 */ + /** 累计掉拍分片数（诊断 / {@code status} 展示；观测型，见 {@link #MISSED_SHARDS}）。 */ public static long missedShards() { return MISSED_SHARDS.sum(); } @@ -275,6 +282,7 @@ public final class EntityTickParallel { /** 分片内逐实体 tick：单实体异常隔离（与批式路径语义一致），异常计入 FaultGuard。 */ private static void modzuozhi_tickShard(List entities, Consumer consumer) { + long t0 = System.nanoTime(); for (Entity entity : entities) { try { if (!entity.isRemoved()) { @@ -285,5 +293,9 @@ public final class EntityTickParallel { FaultGuard.onSubTaskFailure(); } } + // 观测型掉拍：本片负载没能在单 tick 预算内处理完（慢片），累计一次。 + if (System.nanoTime() - t0 > ShardGate.TICK_BUDGET_NS) { + MISSED_SHARDS.increment(); + } } } diff --git a/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java b/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java index 2269df3..ba89087 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/LevelTickLoop.java @@ -60,6 +60,8 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu private volatile boolean scheduled = false; /** 是否正在执行一次 tick（health 检查用）。 */ private volatile boolean ticking = false; + /** 单机内嵌服务器暂停（ESC 菜单）时置 true：取消下一拍并不再自排，杜绝暂停/保存期间并行写区块。 */ + private volatile boolean paused = false; /** 本循环独立的 tick 计数（用于每 20 tick 的时间同步）。 */ private int tickCount; @@ -97,7 +99,11 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu /** 首次启动（或外部主动重排）本循环。 */ public void schedule() { - this.currentTickTask = this.masterPool.submit(this); + try { + this.currentTickTask = this.masterPool.submit(this); + } catch (java.util.concurrent.RejectedExecutionException ignored) { + // masterPool 已被关闭（停机/降级/重启用）：放弃排程，run 内的 running 检查兜底 + } } /** 每 tick 由主线程调用：保证本 loop 只启动一次自排程。 */ @@ -108,6 +114,37 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu } } + public boolean isPaused() { + return this.paused; + } + + /** + * 暂停本维度循环（单机暂停场景调用）：置暂停标志并取消尚未开始的下一拍。 + *

+ * 正在执行的当前拍会自然跑完（其 finally 检查暂停后不再续排），随后维度 worker 完全停摆， + * 不再有任何并行写入 LevelChunk 的随机tick/TE/实体，消灭「暂停/保存期间区块数据包序列化 + * 读到中间态 → 客户端解码越界 → 网络协议错误掉线」的竞态。 + */ + public void pause() { + if (this.paused) { + return; + } + this.paused = true; + Future task = this.currentTickTask; + if (task != null && !task.isCancelled()) { + task.cancel(false); + } + } + + /** 恢复本维度循环：立即重排下一拍。 */ + public void resume() { + if (!this.paused) { + return; + } + this.paused = false; + this.schedule(); + } + public void addConnection(Connection connection) { this.connections.add(connection); } @@ -124,6 +161,12 @@ 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; MinecraftServer server = this.level.getServer(); @@ -149,10 +192,14 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu FaultGuard.onDimTickFailure(t); } finally { this.ticking = false; - long remaining = Math.max(0L, deadline - System.nanoTime()); - if (this.running) { - this.currentTickTask = this.masterPool.schedule(this, remaining, TimeUnit.NANOSECONDS); - } else { + 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 已关闭：放弃续排 + } + } else if (!this.running) { ModZuozhi.LOGGER.info("[DimThread] 维度 {} 独立循环已停止", this.level.dimension().location()); } } diff --git a/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java b/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java index 7428b09..255120b 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/ServerManager.java @@ -52,6 +52,16 @@ public class ServerManager { this.actives.put(server, value); } + /** 是否处于单机暂停（ESC 菜单）状态。 */ + public boolean isPaused() { + for (LevelTickLoop loop : this.loops.values()) { + if (loop.isPaused()) { + return true; + } + } + return false; + } + public ThreadPool getThreadPool(MinecraftServer server) { // 初始尺寸取 CPU 核数兜底，实际线程数由 syncThreadCount 每 tick 贴合维度数 return this.threadPools.computeIfAbsent(server, s -> { @@ -194,6 +204,39 @@ public class ServerManager { } } + /** + * 单机内嵌服务器暂停/恢复（由客户端 {@code Minecraft.isPaused()} 驱动，见 + * {@code MixinMinecraftPauseDetector}）。 + *

+ * 暂停时：先取消所有维度 loop 的下一拍，再等待正在执行的那一拍完全收口（含其内部 + * 并行子任务屏障），使所有维度 worker 彻底停摆——暂停/保存期间不再有任何线程并行写入 + * {@code LevelChunk}；否则维度 worker 与主线程「暂停并保存」并发，区块数据包序列化读到 + * 中间态，客户端解码越界（"网络协议错误"）掉线（历史 bug：ESC 暂停后 0.5s 必掉线）。 + *

+ * 恢复时：重新排程所有维度 loop。LAN 开放时客户端 {@code isPaused()} 恒 false，不会误暂停。 + */ + public void setPaused(boolean paused) { + 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 退避 + } + } + } else { + this.loops.values().forEach(LevelTickLoop::resume); + } + } + /** 移除并停止某维度的 loop（维度卸载时调用）。 */ public void removeLoop(ServerLevel level) { LevelTickLoop loop = this.loops.remove(level); diff --git a/src/main/java/dev/modzuozhi/core/dimthread/ShardGate.java b/src/main/java/dev/modzuozhi/core/dimthread/ShardGate.java index d5e8b94..cc88fe6 100644 --- a/src/main/java/dev/modzuozhi/core/dimthread/ShardGate.java +++ b/src/main/java/dev/modzuozhi/core/dimthread/ShardGate.java @@ -1,6 +1,12 @@ package dev.modzuozhi.core.dimthread; +import net.minecraft.core.BlockPos; +import net.minecraft.resources.ResourceKey; +import net.minecraft.world.level.Level; + +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.LongAdder; /** * 分片 Claim 门（借鉴 Tessellate 的"每区域一个 inFlight CAS claim、慢区掉拍"思想独立实现）。 @@ -16,11 +22,39 @@ import java.util.concurrent.atomic.AtomicBoolean; */ public final class ShardGate { - /** - * 分片数：2 的幂（供 & 掩码），最低 8，按 CPU 核数线性扩展（≈ 2×核数）。 - */ + /** 分片数：2 的幂（供 & 掩码），最低 8，按 CPU 核数线性扩展（≈ 2×核数）。 */ public static final int SHARDS = Integer.highestOneBit(Math.max(8, Runtime.getRuntime().availableProcessors() * 2)); + /** + * 一 tick 的预算（纳秒，50ms = 20 TPS 的单 tick 时长）。 + *

+ * 观测型掉拍阈值：分片任务实际耗时超过该值即视为"该片本 tick 掉拍"——它没能在 + * 一拍时间内处理完本片负载（慢片）。由各阶段分片任务完成后的耗时统计计入累计 + * （claim 抢占失败路径因阶段收口会在本 tick 内等齐所有分片，结构上不可达）。 + */ + public static final long TICK_BUDGET_NS = 50_000_000L; + + /** 黄金比例乘散列常数（Fibonacci hashing），保证相邻坐标映射到不同分片。 */ + private static final long GOLDEN = 0x9E3779B97F4A7C15L; + + // ===== 各阶段 per-dimension 门表 ===== + // + // 实体 / 方块实体 / 区块环境是维度 tick 内三个独立阶段，负载特征互不相同， + // 必须各自独立 claim（共用同一门表会让某阶段慢片拖累其它阶段的同片）。 + // 实体阶段的门表在 EntityTickParallel.GATES（历史实现，保持不动）。 + + /** 方块实体（TE）分片门表：按维度隔离。 */ + private static final ConcurrentHashMap, ShardGate> TE_GATES = new ConcurrentHashMap<>(); + + /** 区块环境（随机 tick / 流体）分片门表：按维度隔离。 */ + private static final ConcurrentHashMap, ShardGate> CHUNK_GATES = new ConcurrentHashMap<>(); + + /** 累计 TE 掉拍分片数（诊断 / status 展示；观测型，超 {@link #TICK_BUDGET_NS} 计入）。 */ + private static final LongAdder TE_MISSED = new LongAdder(); + + /** 累计区块环境掉拍分片数（诊断 / status 展示；观测型，超 {@link #TICK_BUDGET_NS} 计入）。 */ + private static final LongAdder CHUNK_MISSED = new LongAdder(); + /** 每片一个 claim 位（false=空闲，true=本 tick 有任务在跑/已提交）。 */ private final AtomicBoolean[] gates = new AtomicBoolean[SHARDS]; @@ -43,4 +77,51 @@ public final class ShardGate { public void release(int shard) { gates[shard].set(false); } + + // ===== 静态工厂与统计 ===== + + /** 取本维度的 TE 分片门（缺失即建；维度数量极少，长期缓存可接受）。 */ + public static ShardGate teGate(ResourceKey dimension) { + return TE_GATES.computeIfAbsent(dimension, key -> new ShardGate()); + } + + /** 取本维度的区块环境分片门（缺失即建；维度数量极少，长期缓存可接受）。 */ + public static ShardGate chunkGate(ResourceKey dimension) { + return CHUNK_GATES.computeIfAbsent(dimension, key -> new ShardGate()); + } + + /** 按方块坐标稳定散列到分片索引（TE 分片用；黄金比例乘散列取高位）。 */ + public static int shardOf(BlockPos pos) { + long hash = GOLDEN; + hash = (hash ^ pos.getX()) * GOLDEN; + hash = (hash ^ pos.getY()) * GOLDEN; + hash = (hash ^ pos.getZ()) * GOLDEN; + return (int) (hash >>> 32) & (SHARDS - 1); + } + + /** 按 {@code ChunkPos.asLong(x,z)} 稳定散列到分片索引（区块环境分片用）。 */ + public static int shardOf(long chunkPos) { + long hash = chunkPos * GOLDEN; + return (int) (hash >>> 32) & (SHARDS - 1); + } + + /** TE 掉拍分片累计（诊断 / status 展示）。 */ + public static long teMissed() { + return TE_MISSED.sum(); + } + + /** 区块环境掉拍分片累计（诊断 / status 展示）。 */ + public static long chunkMissed() { + return CHUNK_MISSED.sum(); + } + + /** TE 分片掉拍计数（观测型：分片任务耗时超预算、或 claim 抢占失败）。 */ + public static void bumpTeMissed() { + TE_MISSED.increment(); + } + + /** 区块环境分片掉拍计数（观测型：分片任务耗时超预算、或 claim 抢占失败）。 */ + public static void bumpChunkMissed() { + CHUNK_MISSED.increment(); + } } \ No newline at end of file diff --git a/src/main/java/dev/modzuozhi/core/filter/SerDesFilter.java b/src/main/java/dev/modzuozhi/core/filter/SerDesFilter.java index 0cfaa95..7146e5c 100644 --- a/src/main/java/dev/modzuozhi/core/filter/SerDesFilter.java +++ b/src/main/java/dev/modzuozhi/core/filter/SerDesFilter.java @@ -3,6 +3,7 @@ package dev.modzuozhi.core.filter; import net.minecraft.world.level.block.entity.TickingBlockEntity; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; /** * 默认 SerDes 过滤器。 @@ -12,7 +13,8 @@ import java.util.Set; *

  • 黑名单（强制串行）：已知线程不安全、会与主线程交互的方块实体 * （如移动的活塞/活塞头，涉及跨区块移除与实体交互）。
    • *
  • modded 方块实体（非 {@code minecraft:} 前缀）默认串行，保守兜底； - * 可通过 {@link #setLockModded(boolean)} 关闭该保守行为。
    • + * 可通过 {@link #setLockModded(boolean)} 关闭该保守行为，或通过 + * {@link #whitelistModded(String)} 将确认线程安全的 modded TE 单独加入白名单。 *
  • 其余 vanilla 方块实体默认可并行。
    • * */ @@ -40,6 +42,24 @@ public final class SerDesFilter implements ISerDesFilter { /** 是否对非 vanilla 的 modded 方块实体强制串行（保守兜底，默认开启）。 */ private static volatile boolean lockModded = true; + /** 确认线程安全、可绕开 lockModded 强制串行的 modded TE 注册名白名单（运行时通过命令维护）。 */ + private static final Set MODDED_WHITELIST = ConcurrentHashMap.newKeySet(); + + /** 将某个 modded TE 注册名加入白名单，使其进入并行分片管线（本模组不保证其线程安全，加入前需自行确认）。 */ + public static void whitelistModded(String type) { + MODDED_WHITELIST.add(type); + } + + /** 移除白名单项（恢复默认的 modded 强制串行）。 */ + public static void unwhitelistModded(String type) { + MODDED_WHITELIST.remove(type); + } + + /** 当前白名单内容（诊断展示用）。 */ + public static Set moddedWhitelist() { + return MODDED_WHITELIST; + } + public static void setLockModded(boolean lockModded) { SerDesFilter.lockModded = lockModded; } @@ -55,7 +75,7 @@ public final class SerDesFilter implements ISerDesFilter { return false; } if (lockModded && !type.startsWith("minecraft:")) { - return false; + return MODDED_WHITELIST.contains(type); } return true; } diff --git a/src/main/java/dev/modzuozhi/mixin/MixinMinecraftPauseDetector.java b/src/main/java/dev/modzuozhi/mixin/MixinMinecraftPauseDetector.java new file mode 100644 index 0000000..b91f61d --- /dev/null +++ b/src/main/java/dev/modzuozhi/mixin/MixinMinecraftPauseDetector.java @@ -0,0 +1,30 @@ +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/ServerChunkCacheMixin.java b/src/main/java/dev/modzuozhi/mixin/ServerChunkCacheMixin.java index 6d859c4..2877854 100644 --- a/src/main/java/dev/modzuozhi/mixin/ServerChunkCacheMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/ServerChunkCacheMixin.java @@ -3,10 +3,15 @@ package dev.modzuozhi.mixin; import com.llamalad7.mixinextras.injector.wrapmethod.WrapMethod; import com.llamalad7.mixinextras.injector.wrapoperation.Operation; import com.llamalad7.mixinextras.injector.wrapoperation.WrapOperation; +import dev.modzuozhi.ModZuozhi; +import dev.modzuozhi.core.dimthread.ChunkLock; 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.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; @@ -274,6 +279,9 @@ public abstract class ServerChunkCacheMixin implements IMutableMainThread { /** 本次 chunk tick 的子任务屏障。 */ @Unique private FineGrainScheduler.Barrier modzuozhi_chunkBarrier; + /** 区块环境分片门（跨 tick 存续，慢片掉拍时本 tick 该区块延后）。 */ + @Unique + private ShardGate modzuozhi_chunkGate; @Unique private boolean modzuozhi_chunkShouldParallel() { @@ -329,19 +337,77 @@ public abstract class ServerChunkCacheMixin implements IMutableMainThread { } MinecraftServer server = this.level.getServer(); this.modzuozhi_chunkBarrier = FineGrainScheduler.beginBarrier(); - for (Object[] entry : this.modzuozhi_pendingChunks) { - ServerLevel level = (ServerLevel) entry[0]; - LevelChunk chunk = (LevelChunk) entry[1]; - int randomTickSpeed = (Integer) entry[2]; - this.modzuozhi_chunkBarrier.submit(server, () -> { - // 线程本地随机源，避免并行子任务跨线程访问共享 Level.random - ((IChunkEnvTick) level).modzuozhi_tickChunk(chunk, randomTickSpeed, RandomSource.create()); - }, chunk.getPos().getWorldPosition()); + this.modzuozhi_chunkGate = ShardGate.chunkGate(this.level.dimension()); + boolean sharded = EntityTickParallel.isSharded(); + if (sharded) { + // 分片模式：先按区块坐标稳定散列聚合收集，收口时每片只 tryAcquire 一次。 + // 逐区块 tryAcquire 会让同片第 2 个起的区块全部被同拍竞争误判为掉拍（永不随机 tick）。 + List[] shards = new List[ShardGate.SHARDS]; + for (Object[] entry : this.modzuozhi_pendingChunks) { + LevelChunk chunk = (LevelChunk) entry[1]; + ChunkPos cpos = chunk.getPos(); + int s = ShardGate.shardOf(ChunkPos.asLong(cpos.x, cpos.z)); + List list = shards[s]; + if (list == null) { + list = new ArrayList<>(); + shards[s] = list; + } + list.add(entry); + } + for (int s = 0; s < ShardGate.SHARDS; s++) { + List list = shards[s]; + if (list == null || list.isEmpty()) { + continue; + } + if (this.modzuozhi_chunkGate.tryAcquire(s)) { + this.modzuozhi_chunkBarrier.submitShard(server, this.modzuozhi_chunkGate, s, + () -> modzuozhi_tickChunkShard(list)); + } else { + ShardGate.bumpChunkMissed(); + ModZuozhi.LOGGER.debug("[ChunkEnv] 分片 {} 掉拍(上 tick 未完成)，整片延后 1 tick", s); + } + } + } else { + for (Object[] entry : this.modzuozhi_pendingChunks) { + ServerLevel level = (ServerLevel) entry[0]; + LevelChunk chunk = (LevelChunk) entry[1]; + int randomTickSpeed = (Integer) entry[2]; + Runnable task = () -> { + // 线程本地随机源，避免并行子任务跨线程访问共享 Level.random + ((IChunkEnvTick) level).modzuozhi_tickChunk(chunk, randomTickSpeed, RandomSource.create()); + }; + this.modzuozhi_chunkBarrier.submit(server, task, chunk.getPos().getWorldPosition()); + } } this.modzuozhi_pendingChunks.clear(); this.modzuozhi_chunkBarrier.await(); this.modzuozhi_chunkBarrier.drain(); this.modzuozhi_chunkBarrier = null; + this.modzuozhi_chunkGate = null; this.modzuozhi_chunkParallel = false; } + + /** + * 分片内逐个区块做环境 tick：保持原逐区块提交的区块锁语义（防止与 TE / 实体阶段 + * 并发访问同一区块），每个区块独立线程本地随机源，单区块异常隔离。 + */ + @Unique + private void modzuozhi_tickChunkShard(List list) { + long t0 = System.nanoTime(); + for (Object[] entry : list) { + ServerLevel level = (ServerLevel) entry[0]; + LevelChunk chunk = (LevelChunk) entry[1]; + int randomTickSpeed = (Integer) entry[2]; + try (AutoCloseable lock = ChunkLock.lock(chunk.getPos().getWorldPosition())) { + ((IChunkEnvTick) level).modzuozhi_tickChunk(chunk, randomTickSpeed, RandomSource.create()); + } catch (Throwable t) { + ModZuozhi.LOGGER.error("[ChunkEnv] 分片子任务异常", t); + FaultGuard.onSubTaskFailure(); + } + } + // 观测型掉拍：本片负载没能在单 tick 预算内处理完（慢片），累计一次。 + if (System.nanoTime() - t0 > ShardGate.TICK_BUDGET_NS) { + ShardGate.bumpChunkMissed(); + } + } } diff --git a/src/main/java/dev/modzuozhi/mixin/TickingBlockEntityParallelMixin.java b/src/main/java/dev/modzuozhi/mixin/TickingBlockEntityParallelMixin.java index 72b0e0b..b1fedfd 100644 --- a/src/main/java/dev/modzuozhi/mixin/TickingBlockEntityParallelMixin.java +++ b/src/main/java/dev/modzuozhi/mixin/TickingBlockEntityParallelMixin.java @@ -1,8 +1,13 @@ package dev.modzuozhi.mixin; +import dev.modzuozhi.ModZuozhi; +import dev.modzuozhi.core.dimthread.ChunkLock; 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.PostExecuteQueue; +import dev.modzuozhi.core.dimthread.ShardGate; import dev.modzuozhi.core.filter.SerDesFilter; import net.minecraft.server.MinecraftServer; import net.minecraft.server.level.ServerLevel; @@ -17,7 +22,9 @@ import org.spongepowered.asm.mixin.injection.Inject; import org.spongepowered.asm.mixin.injection.Redirect; import org.spongepowered.asm.mixin.injection.callback.CallbackInfo; +import java.util.ArrayList; import java.util.Collection; +import java.util.List; /** * 方块实体（TE）细粒度并行（借鉴 MCMT 思路，写我们自己的实现）。 @@ -27,6 +34,11 @@ import java.util.Collection; * 并行执行，其余（黑名单/非 vanilla）串行；循环结束后 {@code await} 收口，再 * {@code drain} 子任务投递的延迟回调。 *

    + * 分片模式：与实体分片同语义——循环内只按方块坐标稳定散列收集到分片数组， + * 循环结束后每片只 tryAcquire 一次（整片打包为一个子任务，片内串行 tick）， + * 把掉拍粒度从"逐 TE 竞争"（会导致同片第 2 个起的 TE 全部被拒、永不执行）修正为 + * "整片掉拍延后"，与实体 Batch 聚合语义一致。 + *

    * 线程安全关键：并行子任务运行在其它 worker 线程，可能并发写 {@code pendingBlockEntityTickers} * 等普通集合。这里拦截 {@code addBlockEntityTicker} / {@code addFreshBlockEntities}， * 在并行子任务上下文中把写入投递到 {@link PostExecuteQueue}，由维度 worker 串行执行。 @@ -47,7 +59,12 @@ public abstract class TickingBlockEntityParallelMixin { @Unique private FineGrainScheduler.Barrier modzuozhi_teBarrier; @Unique + private ShardGate modzuozhi_teGate; + @Unique private boolean modzuozhi_teParallel; + /** 本 tick 按分片收集的待并行 TE（索引 = {@code ShardGate.shardOf(pos)}）。 */ + @Unique + private List[] modzuozhi_teShards; @Unique private boolean modzuozhi_shouldParallel() { @@ -66,8 +83,11 @@ public abstract class TickingBlockEntityParallelMixin { @Inject(method = "tickBlockEntities", at = @At("HEAD")) private void modzuozhi_teBegin(CallbackInfo ci) { this.modzuozhi_teParallel = modzuozhi_shouldParallel(); + // 每 tick 重建分片数组（16 槽），防止上一 tick 残留数据串拍 + this.modzuozhi_teShards = new List[ShardGate.SHARDS]; if (this.modzuozhi_teParallel) { this.modzuozhi_teBarrier = FineGrainScheduler.beginBarrier(); + this.modzuozhi_teGate = ShardGate.teGate(((ServerLevel) (Object) this).dimension()); } } @@ -77,7 +97,19 @@ public abstract class TickingBlockEntityParallelMixin { private void modzuozhi_teTick(TickingBlockEntity ticker) { if (this.modzuozhi_teParallel && SerDesFilter.INSTANCE.isParallel(ticker)) { MinecraftServer server = ((ServerLevel) (Object) this).getServer(); - this.modzuozhi_teBarrier.submit(server, ticker::tick, ticker.getPos()); + if (EntityTickParallel.isSharded()) { + // 分片模式：只收集，不在此处 tryAcquire（同拍冲突会让同片第 2 个起的 TE 全部误判掉拍）。 + int shard = ShardGate.shardOf(ticker.getPos()); + List list = this.modzuozhi_teShards[shard]; + if (list == null) { + list = new ArrayList<>(); + this.modzuozhi_teShards[shard] = list; + } + list.add(ticker); + } else { + // 非分片回退：逐 TE 独立子任务并行（原语义） + this.modzuozhi_teBarrier.submit(server, ticker::tick, ticker.getPos()); + } } else { ticker.tick(); } @@ -86,11 +118,49 @@ public abstract class TickingBlockEntityParallelMixin { @Inject(method = "tickBlockEntities", at = @At("RETURN")) private void modzuozhi_teEnd(CallbackInfo ci) { if (this.modzuozhi_teParallel && this.modzuozhi_teBarrier != null) { + MinecraftServer server = ((ServerLevel) (Object) this).getServer(); + for (int s = 0; s < ShardGate.SHARDS; s++) { + List list = this.modzuozhi_teShards[s]; + if (list == null || list.isEmpty()) { + continue; + } + // 每片一次 tryAcquire：掉拍 = 上一 tick 该片未完成（真实慢片），整片延后。 + if (this.modzuozhi_teGate.tryAcquire(s)) { + this.modzuozhi_teBarrier.submitShard(server, this.modzuozhi_teGate, s, + () -> modzuozhi_tickTeShard(list)); + } else { + ShardGate.bumpTeMissed(); + ModZuozhi.LOGGER.debug("[TE] 分片 {} 掉拍(上 tick 未完成)，整片延后 1 tick", s); + } + } this.modzuozhi_teBarrier.await(); this.modzuozhi_teBarrier.drain(); this.modzuozhi_teBarrier = null; + this.modzuozhi_teGate = null; this.modzuozhi_teParallel = false; } + this.modzuozhi_teShards = null; + } + + /** + * 分片内逐个 TE tick：保持原逐 TE 提交的区块锁语义（同 chunk 的 TE 可能散列到不同片， + * 需锁所在区块防止并发改方块），单 TE 异常隔离。 + */ + @Unique + private void modzuozhi_tickTeShard(List list) { + long t0 = System.nanoTime(); + for (TickingBlockEntity ticker : list) { + try (AutoCloseable lock = ChunkLock.lock(ticker.getPos())) { + ticker.tick(); + } catch (Throwable t) { + ModZuozhi.LOGGER.error("[TE] 分片子任务异常", t); + FaultGuard.onSubTaskFailure(); + } + } + // 观测型掉拍：本片负载没能在单 tick 预算内处理完（慢片），累计一次。 + if (System.nanoTime() - t0 > ShardGate.TICK_BUDGET_NS) { + ShardGate.bumpTeMissed(); + } } @Inject(method = "addBlockEntityTicker", at = @At("HEAD"), cancellable = true) @@ -110,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 7b8335f..58a01e0 100644 --- a/src/main/resources/dimensionalripper.mixins.json +++ b/src/main/resources/dimensionalripper.mixins.json @@ -64,6 +64,7 @@ "YieldJobSiteGuardMixin" ], "client": [ + "MixinMinecraftPauseDetector" ], "server": [ ],