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.
This commit is contained in:
modzuozhi committed 2026-09-12 00:44:23 +08:00
1 parent 3b9320d1e0
commit 801deb701a
12 files changed
+432 -29

No files matched your search

+2 -2
View File
@@ -6,7 +6,7 @@ Dimensional Ripper 是一个基于 **NeoForge 1.21.1** 的多线程维度并行
- **维度并行 Tick**:每个维度拥有独立的 tick 循环与 worker 线程,维度之间互不拖累——单维度低 TPS 不会影响其他维度。 - **维度并行 Tick**:每个维度拥有独立的 tick 循环与 worker 线程,维度之间互不拖累——单维度低 TPS 不会影响其他维度。
- **细粒度并行**:在维度内部对实体、方块实体、区块环境 tick 进行并行调度,充分压榨多核性能。 - **细粒度并行**:在维度内部对实体、方块实体、区块环境 tick 进行并行调度,充分压榨多核性能。
- **实体分片并行 (ShardGate)**:实体按稳定 ID 分为与 CPU 核数对应的若干片,每片由跨 tick 的 CAS 门控独立提交。单个慢分片自动掉拍延后,不再阻塞整个实体 tick 屏障——把"单点慢"从拖垮全局变成只拖自己。 - **分片并行 (ShardGate)**:实体/方块实体/区块环境三阶段统一按稳定键分片(实体按 ID、TE 按方块坐标、区块环境按区块坐标),每片由跨 tick 的 CAS 门控独立提交。单个慢分片自动掉拍延后,不再阻塞对应阶段的并行屏障——把"单点慢"从拖垮全局变成只拖自己。
- **POI 并发锁优化**:将村民等实体高频读取的职业兴趣点(POI)存储从全互斥同步容器替换为无锁并发容器,消除并行 tick 下的全局锁瓶颈。约 1.9 万村民压测下 TPS 由 9 提升至 20(16 核)。 - **POI 并发锁优化**:将村民等实体高频读取的职业兴趣点(POI)存储从全互斥同步容器替换为无锁并发容器,消除并行 tick 下的全局锁瓶颈。约 1.9 万村民压测下 TPS 由 9 提升至 20(16 核)。
- **碰撞优化**:可禁用实体间碰撞,消除 O(n²) 碰撞开销,显著释放 TPS。 - **碰撞优化**:可禁用实体间碰撞,消除 O(n²) 碰撞开销,显著释放 TPS。
- **异步传送门传送**:支持原版下界/末地传送门与模组维度(如暮色森林、以太)传送门的异步预加载与渐进生成,避免传送到未加载区块导致的卡顿、冻结与虚空传送。 - **异步传送门传送**:支持原版下界/末地传送门与模组维度(如暮色森林、以太)传送门的异步预加载与渐进生成,避免传送到未加载区块导致的卡顿、冻结与虚空传送。
@@ -29,7 +29,7 @@ Dimensional Ripper 是一个基于 **NeoForge 1.21.1** 的多线程维度并行
|------|------| |------|------|
| `/dimensionalripper status` | 查看当前状态、TPS、各开关与掉拍分片统计 | | `/dimensionalripper status` | 查看当前状态、TPS、各开关与掉拍分片统计 |
| `/dimensionalripper fine on\|off` | 开关维度内细粒度并行(实体/方块实体/区块环境) | | `/dimensionalripper fine on\|off` | 开关维度内细粒度并行(实体/方块实体/区块环境) |
| `/dimensionalripper fine sharded on\|off` | 开关实体分片并行(ShardGate,默认开启) | | `/dimensionalripper fine sharded on\|off` | 开关分片并行(实体/TE/区块环境,ShardGate,默认开启) |
| `/dimensionalripper nocollide on\|off` | 开关实体碰撞优化(默认开启) | | `/dimensionalripper nocollide on\|off` | 开关实体碰撞优化(默认开启) |
模组**默认启用**多线程与分片并行,开箱即用,无需额外配置。 模组**默认启用**多线程与分片并行,开箱即用,无需额外配置。
@@ -7,8 +7,11 @@ import dev.modzuozhi.core.dimthread.DimThreadCore;
import dev.modzuozhi.core.dimthread.EntityTickParallel; import dev.modzuozhi.core.dimthread.EntityTickParallel;
import dev.modzuozhi.core.dimthread.FaultGuard; import dev.modzuozhi.core.dimthread.FaultGuard;
import dev.modzuozhi.core.dimthread.FineGrainScheduler; import dev.modzuozhi.core.dimthread.FineGrainScheduler;
import dev.modzuozhi.core.dimthread.ShardGate;
import dev.modzuozhi.core.dimthread.TpsTracker; import dev.modzuozhi.core.dimthread.TpsTracker;
import dev.modzuozhi.core.filter.SerDesFilter;
import net.minecraft.commands.CommandSourceStack; import net.minecraft.commands.CommandSourceStack;
import net.minecraft.commands.arguments.ResourceLocationArgument;
import net.minecraft.network.chat.Component; import net.minecraft.network.chat.Component;
import net.minecraft.resources.ResourceKey; import net.minecraft.resources.ResourceKey;
import net.minecraft.server.level.ServerLevel; import net.minecraft.server.level.ServerLevel;
@@ -43,7 +46,11 @@ public final class MasterCommand {
.then(literal("off").executes(ctx -> setSharded(ctx, false))))) .then(literal("off").executes(ctx -> setSharded(ctx, false)))))
.then(literal("nocollide") .then(literal("nocollide")
.then(literal("on").executes(ctx -> setNoCollide(ctx, true))) .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<CommandSourceStack> ctx, boolean enabled) { private static int setFine(CommandContext<CommandSourceStack> ctx, boolean enabled) {
@@ -74,7 +81,7 @@ public final class MasterCommand {
private static int setSharded(CommandContext<CommandSourceStack> ctx, boolean enabled) { private static int setSharded(CommandContext<CommandSourceStack> ctx, boolean enabled) {
EntityTickParallel.setSharded(enabled); EntityTickParallel.setSharded(enabled);
ctx.getSource().sendSuccess(() -> ctx.getSource().sendSuccess(() ->
Component.literal("实体分片并行已" + (enabled ? "开启" : "关闭")), false); Component.literal("分片并行(实体/TE/区块环境)已" + (enabled ? "开启" : "关闭")), false);
return 1; return 1;
} }
@@ -86,6 +93,16 @@ public final class MasterCommand {
return 1; return 1;
} }
/** 将指定 TE 注册名加入 modded 并行白名单(仅影响该 TE,不改全局 lockModded 保守策略)。 */
private static int teWhitelist(CommandContext<CommandSourceStack> 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<CommandSourceStack> ctx) { private static int status(CommandContext<CommandSourceStack> ctx) {
String fine = FineGrainScheduler.isEnabled() ? "开启" : "关闭"; String fine = FineGrainScheduler.isEnabled() ? "开启" : "关闭";
String noCollide = ModZuozhi.MOB_COLLISION_OFF ? "开启" : "关闭"; String noCollide = ModZuozhi.MOB_COLLISION_OFF ? "开启" : "关闭";
@@ -104,11 +121,12 @@ public final class MasterCommand {
String msg = String.format( String msg = String.format(
"modzuozhi 状态:细粒度=%s%s,分片并行=%s,禁碰撞=%s,故障原因:%s\n" "modzuozhi 状态:细粒度=%s%s,分片并行=%s,禁碰撞=%s,故障原因:%s\n"
+ "当前维度:%s,TPS:%.1f,掉拍分片累计:%d", + "当前维度:%s,TPS:%.1f,掉拍分片累计:实体 %d / TE %d / 区块环境 %d",
fine, fineMark, fine, fineMark,
EntityTickParallel.isSharded() ? "开启" : "关闭", EntityTickParallel.isSharded() ? "开启" : "关闭",
noCollide, reason, noCollide, reason,
dimensionName(dimension), tps, EntityTickParallel.missedShards()); dimensionName(dimension), tps,
EntityTickParallel.missedShards(), ShardGate.teMissed(), ShardGate.chunkMissed());
ctx.getSource().sendSuccess(() -> Component.literal(msg), false); ctx.getSource().sendSuccess(() -> Component.literal(msg), false);
return 1; return 1;
} }
@@ -60,4 +60,19 @@ public final class DimThreadCore {
public static boolean owns(Thread thread) { public static boolean owns(Thread thread) {
return thread.getName().startsWith(MOD_ID + "_server_"); return thread.getName().startsWith(MOD_ID + "_server_");
} }
/** 查询是否处于单机暂停状态。 */
public static boolean isPaused() {
return MANAGER.isPaused();
}
/**
* 单机内嵌服务器暂停/恢复信号(见 {@link ServerManager#setPaused})。
* <p>
* 由客户端 mixin {@code MixinMinecraftPauseDetector} 在每帧 {@code Minecraft.runTick} 检测
* {@code isPaused()} 变化后调用;dedicated server 无客户端,永不会触发。
*/
public static void setPaused(boolean paused) {
MANAGER.setPaused(paused);
}
} }
@@ -51,7 +51,14 @@ public final class EntityTickParallel {
*/ */
private static final ConcurrentHashMap<ResourceKey<?>, ShardGate> GATES = new ConcurrentHashMap<>(); private static final ConcurrentHashMap<ResourceKey<?>, ShardGate> GATES = new ConcurrentHashMap<>();
/** 累计掉拍分片数(诊断 / status 展示)。 */ /**
* 累计掉拍分片数(诊断 / status 展示)。
* <p>
* 观测型指标:{@link #modzuozhi_tickShard} 中分片任务实际耗时 &gt; {@link ShardGate#TICK_BUDGET_NS}
* (一 tick 预算)即计一次"该片本 tick 掉拍"——它没能在一拍时间内处理完本片负载(慢片)。
* claim 抢占失败路径因阶段收口 {@link FineGrainScheduler.Barrier#await()} 会在本 tick 内等齐
* 所有已提交分片,结构上不可达,仅保留作为兜底。
*/
private static final LongAdder MISSED_SHARDS = new LongAdder(); private static final LongAdder MISSED_SHARDS = new LongAdder();
/** 实体分片并行的独立开关(默认开启;关闭时回退旧的批式 submitBatch 路径)。 */ /** 实体分片并行的独立开关(默认开启;关闭时回退旧的批式 submitBatch 路径)。 */
@@ -164,10 +171,10 @@ public final class EntityTickParallel {
public static void setSharded(boolean value) { public static void setSharded(boolean value) {
sharded = 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() { public static long missedShards() {
return MISSED_SHARDS.sum(); return MISSED_SHARDS.sum();
} }
@@ -275,6 +282,7 @@ public final class EntityTickParallel {
/** 分片内逐实体 tick:单实体异常隔离(与批式路径语义一致),异常计入 FaultGuard。 */ /** 分片内逐实体 tick:单实体异常隔离(与批式路径语义一致),异常计入 FaultGuard。 */
private static void modzuozhi_tickShard(List<Entity> entities, Consumer<Entity> consumer) { private static void modzuozhi_tickShard(List<Entity> entities, Consumer<Entity> consumer) {
long t0 = System.nanoTime();
for (Entity entity : entities) { for (Entity entity : entities) {
try { try {
if (!entity.isRemoved()) { if (!entity.isRemoved()) {
@@ -285,5 +293,9 @@ public final class EntityTickParallel {
FaultGuard.onSubTaskFailure(); FaultGuard.onSubTaskFailure();
} }
} }
// 观测型掉拍:本片负载没能在单 tick 预算内处理完(慢片),累计一次。
if (System.nanoTime() - t0 > ShardGate.TICK_BUDGET_NS) {
MISSED_SHARDS.increment();
}
} }
} }
@@ -60,6 +60,8 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu
private volatile boolean scheduled = false; private volatile boolean scheduled = false;
/** 是否正在执行一次 tick(health 检查用)。 */ /** 是否正在执行一次 tick(health 检查用)。 */
private volatile boolean ticking = false; private volatile boolean ticking = false;
/** 单机内嵌服务器暂停(ESC 菜单)时置 true:取消下一拍并不再自排,杜绝暂停/保存期间并行写区块。 */
private volatile boolean paused = false;
/** 本循环独立的 tick 计数(用于每 20 tick 的时间同步)。 */ /** 本循环独立的 tick 计数(用于每 20 tick 的时间同步)。 */
private int tickCount; private int tickCount;
@@ -97,7 +99,11 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu
/** 首次启动(或外部主动重排)本循环。 */ /** 首次启动(或外部主动重排)本循环。 */
public void schedule() { 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 只启动一次自排程。 */ /** 每 tick 由主线程调用:保证本 loop 只启动一次自排程。 */
@@ -108,6 +114,37 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu
} }
} }
public boolean isPaused() {
return this.paused;
}
/**
* 暂停本维度循环(单机暂停场景调用):置暂停标志并取消尚未开始的下一拍。
* <p>
* 正在执行的当前拍会自然跑完(其 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) { public void addConnection(Connection connection) {
this.connections.add(connection); this.connections.add(connection);
} }
@@ -124,6 +161,12 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu
@Override @Override
public void run() { public void run() {
DimThreadCore.attach(Thread.currentThread(), this.level); DimThreadCore.attach(Thread.currentThread(), this.level);
// 暂停期间若仍有取消失败/竞态残留的任务被执行,直接放弃本拍(不进入维度 tick,不续排);
// 正常路径下 pause() 会 cancel 掉当前拍,这里只是兜底。
if (this.paused || !this.running) {
this.currentTickTask = null;
return;
}
this.ticking = true; this.ticking = true;
MinecraftServer server = this.level.getServer(); MinecraftServer server = this.level.getServer();
@@ -149,10 +192,14 @@ public final class LevelTickLoop implements Runnable, java.util.concurrent.Execu
FaultGuard.onDimTickFailure(t); FaultGuard.onDimTickFailure(t);
} finally { } finally {
this.ticking = false; this.ticking = false;
long remaining = Math.max(0L, deadline - System.nanoTime()); if (this.running && !this.paused) {
if (this.running) { long remaining = Math.max(0L, deadline - System.nanoTime());
this.currentTickTask = this.masterPool.schedule(this, remaining, TimeUnit.NANOSECONDS); try {
} else { 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()); ModZuozhi.LOGGER.info("[DimThread] 维度 {} 独立循环已停止", this.level.dimension().location());
} }
} }
@@ -52,6 +52,16 @@ public class ServerManager {
this.actives.put(server, value); 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) { public ThreadPool getThreadPool(MinecraftServer server) {
// 初始尺寸取 CPU 核数兜底,实际线程数由 syncThreadCount 每 tick 贴合维度数 // 初始尺寸取 CPU 核数兜底,实际线程数由 syncThreadCount 每 tick 贴合维度数
return this.threadPools.computeIfAbsent(server, s -> { return this.threadPools.computeIfAbsent(server, s -> {
@@ -194,6 +204,39 @@ public class ServerManager {
} }
} }
/**
* 单机内嵌服务器暂停/恢复(由客户端 {@code Minecraft.isPaused()} 驱动,见
* {@code MixinMinecraftPauseDetector})。
* <p>
* 暂停时:先取消所有维度 loop 的下一拍,再等待正在执行的那一拍完全收口(含其内部
* 并行子任务屏障),使所有维度 worker 彻底停摆——暂停/保存期间不再有任何线程并行写入
* {@code LevelChunk};否则维度 worker 与主线程「暂停并保存」并发,区块数据包序列化读到
* 中间态,客户端解码越界("网络协议错误")掉线(历史 bug:ESC 暂停后 0.5s 必掉线)。
* <p>
* 恢复时:重新排程所有维度 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(维度卸载时调用)。 */ /** 移除并停止某维度的 loop(维度卸载时调用)。 */
public void removeLoop(ServerLevel level) { public void removeLoop(ServerLevel level) {
LevelTickLoop loop = this.loops.remove(level); LevelTickLoop loop = this.loops.remove(level);
@@ -1,6 +1,12 @@
package dev.modzuozhi.core.dimthread; 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.AtomicBoolean;
import java.util.concurrent.atomic.LongAdder;
/** /**
* 分片 Claim 门(借鉴 Tessellate 的"每区域一个 inFlight CAS claim、慢区掉拍"思想独立实现)。 * 分片 Claim 门(借鉴 Tessellate 的"每区域一个 inFlight CAS claim、慢区掉拍"思想独立实现)。
@@ -16,11 +22,39 @@ import java.util.concurrent.atomic.AtomicBoolean;
*/ */
public final class ShardGate { 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)); public static final int SHARDS = Integer.highestOneBit(Math.max(8, Runtime.getRuntime().availableProcessors() * 2));
/**
* 一 tick 的预算(纳秒,50ms = 20 TPS 的单 tick 时长)。
* <p>
* 观测型掉拍阈值:分片任务实际耗时超过该值即视为"该片本 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<ResourceKey<?>, ShardGate> TE_GATES = new ConcurrentHashMap<>();
/** 区块环境(随机 tick / 流体)分片门表:按维度隔离。 */
private static final ConcurrentHashMap<ResourceKey<?>, 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 有任务在跑/已提交)。 */ /** 每片一个 claim 位(false=空闲,true=本 tick 有任务在跑/已提交)。 */
private final AtomicBoolean[] gates = new AtomicBoolean[SHARDS]; private final AtomicBoolean[] gates = new AtomicBoolean[SHARDS];
@@ -43,4 +77,51 @@ public final class ShardGate {
public void release(int shard) { public void release(int shard) {
gates[shard].set(false); gates[shard].set(false);
} }
// ===== 静态工厂与统计 =====
/** 取本维度的 TE 分片门(缺失即建;维度数量极少,长期缓存可接受)。 */
public static ShardGate teGate(ResourceKey<Level> dimension) {
return TE_GATES.computeIfAbsent(dimension, key -> new ShardGate());
}
/** 取本维度的区块环境分片门(缺失即建;维度数量极少,长期缓存可接受)。 */
public static ShardGate chunkGate(ResourceKey<Level> 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();
}
} }
@@ -3,6 +3,7 @@ package dev.modzuozhi.core.filter;
import net.minecraft.world.level.block.entity.TickingBlockEntity; import net.minecraft.world.level.block.entity.TickingBlockEntity;
import java.util.Set; import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/** /**
* 默认 SerDes 过滤器。 * 默认 SerDes 过滤器。
@@ -12,7 +13,8 @@ import java.util.Set;
* <li>黑名单(强制串行):已知线程不安全、会与主线程交互的方块实体 * <li>黑名单(强制串行):已知线程不安全、会与主线程交互的方块实体
* (如移动的活塞/活塞头,涉及跨区块移除与实体交互)。</li> * (如移动的活塞/活塞头,涉及跨区块移除与实体交互)。</li>
* <li>modded 方块实体(非 {@code minecraft:} 前缀)默认串行,保守兜底; * <li>modded 方块实体(非 {@code minecraft:} 前缀)默认串行,保守兜底;
* 可通过 {@link #setLockModded(boolean)} 关闭该保守行为。</li> * 可通过 {@link #setLockModded(boolean)} 关闭该保守行为,或通过
* {@link #whitelistModded(String)} 将确认线程安全的 modded TE 单独加入白名单。</li>
* <li>其余 vanilla 方块实体默认可并行。</li> * <li>其余 vanilla 方块实体默认可并行。</li>
* </ul> * </ul>
*/ */
@@ -40,6 +42,24 @@ public final class SerDesFilter implements ISerDesFilter {
/** 是否对非 vanilla 的 modded 方块实体强制串行(保守兜底,默认开启)。 */ /** 是否对非 vanilla 的 modded 方块实体强制串行(保守兜底,默认开启)。 */
private static volatile boolean lockModded = true; private static volatile boolean lockModded = true;
/** 确认线程安全、可绕开 lockModded 强制串行的 modded TE 注册名白名单(运行时通过命令维护)。 */
private static final Set<String> 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<String> moddedWhitelist() {
return MODDED_WHITELIST;
}
public static void setLockModded(boolean lockModded) { public static void setLockModded(boolean lockModded) {
SerDesFilter.lockModded = lockModded; SerDesFilter.lockModded = lockModded;
} }
@@ -55,7 +75,7 @@ public final class SerDesFilter implements ISerDesFilter {
return false; return false;
} }
if (lockModded && !type.startsWith("minecraft:")) { if (lockModded && !type.startsWith("minecraft:")) {
return false; return MODDED_WHITELIST.contains(type);
} }
return true; return true;
} }
@@ -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 菜单「暂停并保存」)。
* <p>
* 修复历史 bug:单机暂停只冻结主线程,Dimensional Ripper 的维度 worker 不受影响仍在并行
* 写入 {@code LevelChunk},与主线程的存盘/区块数据包序列化并发 → 客户端收到残缺区块包解码
* 越界("网络协议错误")掉线(ESC 暂停后 0.5s 内必现)。这里在暂停开始的帧内把维度 loop
* 一并冻结(见 {@link DimThreadCore#setPaused(boolean)}),恢复时自动重排。
* <ul>
* <li>LAN 开放时 {@code Minecraft.isPaused()} 恒为 false,不会误暂停;</li>
* <li>dedicated server 无客户端,本 mixin 仅注册于 {@code client} 段,不会被加载;</li>
* <li>{@code setPaused} 内部同值短路,每帧一次布尔比较开销可忽略。</li>
* </ul>
*/
@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());
}
}
@@ -3,10 +3,15 @@ package dev.modzuozhi.mixin;
import com.llamalad7.mixinextras.injector.wrapmethod.WrapMethod; import com.llamalad7.mixinextras.injector.wrapmethod.WrapMethod;
import com.llamalad7.mixinextras.injector.wrapoperation.Operation; import com.llamalad7.mixinextras.injector.wrapoperation.Operation;
import com.llamalad7.mixinextras.injector.wrapoperation.WrapOperation; 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.DimThreadCore;
import dev.modzuozhi.core.dimthread.EntityTickParallel;
import dev.modzuozhi.core.dimthread.FaultGuard;
import dev.modzuozhi.core.dimthread.FineGrainScheduler; import dev.modzuozhi.core.dimthread.FineGrainScheduler;
import dev.modzuozhi.core.dimthread.IChunkEnvTick; import dev.modzuozhi.core.dimthread.IChunkEnvTick;
import dev.modzuozhi.core.dimthread.IMutableMainThread; import dev.modzuozhi.core.dimthread.IMutableMainThread;
import dev.modzuozhi.core.dimthread.ShardGate;
import net.minecraft.core.BlockPos; import net.minecraft.core.BlockPos;
import net.minecraft.core.Holder; import net.minecraft.core.Holder;
import net.minecraft.core.registries.Registries; import net.minecraft.core.registries.Registries;
@@ -274,6 +279,9 @@ public abstract class ServerChunkCacheMixin implements IMutableMainThread {
/** 本次 chunk tick 的子任务屏障。 */ /** 本次 chunk tick 的子任务屏障。 */
@Unique @Unique
private FineGrainScheduler.Barrier modzuozhi_chunkBarrier; private FineGrainScheduler.Barrier modzuozhi_chunkBarrier;
/** 区块环境分片门(跨 tick 存续,慢片掉拍时本 tick 该区块延后)。 */
@Unique
private ShardGate modzuozhi_chunkGate;
@Unique @Unique
private boolean modzuozhi_chunkShouldParallel() { private boolean modzuozhi_chunkShouldParallel() {
@@ -329,19 +337,77 @@ public abstract class ServerChunkCacheMixin implements IMutableMainThread {
} }
MinecraftServer server = this.level.getServer(); MinecraftServer server = this.level.getServer();
this.modzuozhi_chunkBarrier = FineGrainScheduler.beginBarrier(); this.modzuozhi_chunkBarrier = FineGrainScheduler.beginBarrier();
for (Object[] entry : this.modzuozhi_pendingChunks) { this.modzuozhi_chunkGate = ShardGate.chunkGate(this.level.dimension());
ServerLevel level = (ServerLevel) entry[0]; boolean sharded = EntityTickParallel.isSharded();
LevelChunk chunk = (LevelChunk) entry[1]; if (sharded) {
int randomTickSpeed = (Integer) entry[2]; // 分片模式:先按区块坐标稳定散列聚合收集,收口时每片只 tryAcquire 一次。
this.modzuozhi_chunkBarrier.submit(server, () -> { // 逐区块 tryAcquire 会让同片第 2 个起的区块全部被同拍竞争误判为掉拍(永不随机 tick)。
// 线程本地随机源,避免并行子任务跨线程访问共享 Level.random List<Object[]>[] shards = new List[ShardGate.SHARDS];
((IChunkEnvTick) level).modzuozhi_tickChunk(chunk, randomTickSpeed, RandomSource.create()); for (Object[] entry : this.modzuozhi_pendingChunks) {
}, chunk.getPos().getWorldPosition()); LevelChunk chunk = (LevelChunk) entry[1];
ChunkPos cpos = chunk.getPos();
int s = ShardGate.shardOf(ChunkPos.asLong(cpos.x, cpos.z));
List<Object[]> list = shards[s];
if (list == null) {
list = new ArrayList<>();
shards[s] = list;
}
list.add(entry);
}
for (int s = 0; s < ShardGate.SHARDS; s++) {
List<Object[]> 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_pendingChunks.clear();
this.modzuozhi_chunkBarrier.await(); this.modzuozhi_chunkBarrier.await();
this.modzuozhi_chunkBarrier.drain(); this.modzuozhi_chunkBarrier.drain();
this.modzuozhi_chunkBarrier = null; this.modzuozhi_chunkBarrier = null;
this.modzuozhi_chunkGate = null;
this.modzuozhi_chunkParallel = false; this.modzuozhi_chunkParallel = false;
} }
/**
* 分片内逐个区块做环境 tick:保持原逐区块提交的区块锁语义(防止与 TE / 实体阶段
* 并发访问同一区块),每个区块独立线程本地随机源,单区块异常隔离。
*/
@Unique
private void modzuozhi_tickChunkShard(List<Object[]> 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();
}
}
} }
@@ -1,8 +1,13 @@
package dev.modzuozhi.mixin; 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.DimThreadCore;
import dev.modzuozhi.core.dimthread.EntityTickParallel;
import dev.modzuozhi.core.dimthread.FaultGuard;
import dev.modzuozhi.core.dimthread.FineGrainScheduler; import dev.modzuozhi.core.dimthread.FineGrainScheduler;
import dev.modzuozhi.core.dimthread.PostExecuteQueue; import dev.modzuozhi.core.dimthread.PostExecuteQueue;
import dev.modzuozhi.core.dimthread.ShardGate;
import dev.modzuozhi.core.filter.SerDesFilter; import dev.modzuozhi.core.filter.SerDesFilter;
import net.minecraft.server.MinecraftServer; import net.minecraft.server.MinecraftServer;
import net.minecraft.server.level.ServerLevel; 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.Redirect;
import org.spongepowered.asm.mixin.injection.callback.CallbackInfo; import org.spongepowered.asm.mixin.injection.callback.CallbackInfo;
import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.List;
/** /**
* 方块实体(TE)细粒度并行(借鉴 MCMT 思路,写我们自己的实现)。 * 方块实体(TE)细粒度并行(借鉴 MCMT 思路,写我们自己的实现)。
@@ -27,6 +34,11 @@ import java.util.Collection;
* 并行执行,其余(黑名单/非 vanilla)串行;循环结束后 {@code await} 收口,再 * 并行执行,其余(黑名单/非 vanilla)串行;循环结束后 {@code await} 收口,再
* {@code drain} 子任务投递的延迟回调。 * {@code drain} 子任务投递的延迟回调。
* <p> * <p>
* <b>分片模式</b>:与实体分片同语义——循环内只按方块坐标稳定散列<b>收集</b>到分片数组,
* 循环结束后<b>每片只 tryAcquire 一次</b>(整片打包为一个子任务,片内串行 tick),
* 把掉拍粒度从"逐 TE 竞争"(会导致同片第 2 个起的 TE 全部被拒、永不执行)修正为
* "整片掉拍延后",与实体 Batch 聚合语义一致。
* <p>
* 线程安全关键:并行子任务运行在其它 worker 线程,可能并发写 {@code pendingBlockEntityTickers} * 线程安全关键:并行子任务运行在其它 worker 线程,可能并发写 {@code pendingBlockEntityTickers}
* 等普通集合。这里拦截 {@code addBlockEntityTicker} / {@code addFreshBlockEntities}, * 等普通集合。这里拦截 {@code addBlockEntityTicker} / {@code addFreshBlockEntities},
* 在并行子任务上下文中把写入投递到 {@link PostExecuteQueue},由维度 worker 串行执行。 * 在并行子任务上下文中把写入投递到 {@link PostExecuteQueue},由维度 worker 串行执行。
@@ -47,7 +59,12 @@ public abstract class TickingBlockEntityParallelMixin {
@Unique @Unique
private FineGrainScheduler.Barrier modzuozhi_teBarrier; private FineGrainScheduler.Barrier modzuozhi_teBarrier;
@Unique @Unique
private ShardGate modzuozhi_teGate;
@Unique
private boolean modzuozhi_teParallel; private boolean modzuozhi_teParallel;
/** 本 tick 按分片收集的待并行 TE(索引 = {@code ShardGate.shardOf(pos)})。 */
@Unique
private List<TickingBlockEntity>[] modzuozhi_teShards;
@Unique @Unique
private boolean modzuozhi_shouldParallel() { private boolean modzuozhi_shouldParallel() {
@@ -66,8 +83,11 @@ public abstract class TickingBlockEntityParallelMixin {
@Inject(method = "tickBlockEntities", at = @At("HEAD")) @Inject(method = "tickBlockEntities", at = @At("HEAD"))
private void modzuozhi_teBegin(CallbackInfo ci) { private void modzuozhi_teBegin(CallbackInfo ci) {
this.modzuozhi_teParallel = modzuozhi_shouldParallel(); this.modzuozhi_teParallel = modzuozhi_shouldParallel();
// 每 tick 重建分片数组(16 槽),防止上一 tick 残留数据串拍
this.modzuozhi_teShards = new List[ShardGate.SHARDS];
if (this.modzuozhi_teParallel) { if (this.modzuozhi_teParallel) {
this.modzuozhi_teBarrier = FineGrainScheduler.beginBarrier(); 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) { private void modzuozhi_teTick(TickingBlockEntity ticker) {
if (this.modzuozhi_teParallel && SerDesFilter.INSTANCE.isParallel(ticker)) { if (this.modzuozhi_teParallel && SerDesFilter.INSTANCE.isParallel(ticker)) {
MinecraftServer server = ((ServerLevel) (Object) this).getServer(); 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<TickingBlockEntity> 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 { } else {
ticker.tick(); ticker.tick();
} }
@@ -86,11 +118,49 @@ public abstract class TickingBlockEntityParallelMixin {
@Inject(method = "tickBlockEntities", at = @At("RETURN")) @Inject(method = "tickBlockEntities", at = @At("RETURN"))
private void modzuozhi_teEnd(CallbackInfo ci) { private void modzuozhi_teEnd(CallbackInfo ci) {
if (this.modzuozhi_teParallel && this.modzuozhi_teBarrier != null) { if (this.modzuozhi_teParallel && this.modzuozhi_teBarrier != null) {
MinecraftServer server = ((ServerLevel) (Object) this).getServer();
for (int s = 0; s < ShardGate.SHARDS; s++) {
List<TickingBlockEntity> 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.await();
this.modzuozhi_teBarrier.drain(); this.modzuozhi_teBarrier.drain();
this.modzuozhi_teBarrier = null; this.modzuozhi_teBarrier = null;
this.modzuozhi_teGate = null;
this.modzuozhi_teParallel = false; this.modzuozhi_teParallel = false;
} }
this.modzuozhi_teShards = null;
}
/**
* 分片内逐个 TE tick:保持原逐 TE 提交的区块锁语义(同 chunk 的 TE 可能散列到不同片,
* 需锁所在区块防止并发改方块),单 TE 异常隔离。
*/
@Unique
private void modzuozhi_tickTeShard(List<TickingBlockEntity> 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) @Inject(method = "addBlockEntityTicker", at = @At("HEAD"), cancellable = true)
@@ -110,4 +180,4 @@ public abstract class TickingBlockEntityParallelMixin {
ci.cancel(); ci.cancel();
} }
} }
} }
@@ -64,6 +64,7 @@
"YieldJobSiteGuardMixin" "YieldJobSiteGuardMixin"
], ],
"client": [ "client": [
"MixinMinecraftPauseDetector"
], ],
"server": [ "server": [
], ],