Add entity sharding (ShardGate) and POI lock optimization

- ShardGate: shard entity ticking into per-shard CAS gates so slow shards
  skip a tick instead of blocking the whole barrier
- EntityTickParallel: collect entities by stable id into shards, submit via
  ShardGate, count missed shards
- FineGrainScheduler: add submitShard path with gate release and barrier
  counting
- SectionStorageMixin: replace global exclusive synchronized map with
  ConcurrentLong2ObjectMap to remove the POI read lock bottleneck
  (~19k villagers: TPS 9 -> 20)
- MasterCommand: add 'fine sharded on|off' toggle and status display
- Add environment setup/launch scripts
This commit is contained in:
modzuozhi committed 2026-09-09 22:13:02 +08:00
1 parent 1607a87213
commit 3c4710490c
11 files changed
+391 -31

No files matched your search

@@ -4,6 +4,7 @@ import com.mojang.brigadier.CommandDispatcher;
import com.mojang.brigadier.context.CommandContext;
import dev.modzuozhi.ModZuozhi;
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.TpsTracker;
@@ -36,7 +37,10 @@ public final class MasterCommand {
.then(literal("status").executes(MasterCommand::status))
.then(literal("fine")
.then(literal("on").executes(ctx -> setFine(ctx, true)))
.then(literal("off").executes(ctx -> setFine(ctx, false))))
.then(literal("off").executes(ctx -> setFine(ctx, false)))
.then(literal("sharded")
.then(literal("on").executes(ctx -> setSharded(ctx, true)))
.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)))));
@@ -67,6 +71,13 @@ public final class MasterCommand {
return 1;
}
private static int setSharded(CommandContext<CommandSourceStack> ctx, boolean enabled) {
EntityTickParallel.setSharded(enabled);
ctx.getSource().sendSuccess(() ->
Component.literal("实体分片并行已" + (enabled ? "开启" : "关闭")), false);
return 1;
}
private static int setNoCollide(CommandContext<CommandSourceStack> ctx, boolean enabled) {
ModZuozhi.MOB_COLLISION_OFF = enabled;
ctx.getSource().sendSuccess(() ->
@@ -92,9 +103,12 @@ public final class MasterCommand {
: TpsTracker.tps();
String msg = String.format(
"modzuozhi 状态:细粒度=%s%s,禁碰撞=%s,故障原因:%s\n当前维度:%s,TPS:%.1f",
fine, fineMark, noCollide, reason,
dimensionName(dimension), tps);
"modzuozhi 状态:细粒度=%s%s,分片并行=%s,禁碰撞=%s,故障原因:%s\n"
+ "当前维度:%s,TPS:%.1f,掉拍分片累计:%d",
fine, fineMark,
EntityTickParallel.isSharded() ? "开启" : "关闭",
noCollide, reason,
dimensionName(dimension), tps, EntityTickParallel.missedShards());
ctx.getSource().sendSuccess(() -> Component.literal(msg), false);
return 1;
}
@@ -1,12 +1,17 @@
package dev.modzuozhi.core.dimthread;
import dev.modzuozhi.ModZuozhi;
import net.minecraft.resources.ResourceKey;
import net.minecraft.server.MinecraftServer;
import net.minecraft.world.entity.Entity;
import net.minecraft.world.level.Level;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Consumer;
@@ -37,6 +42,21 @@ public final class EntityTickParallel {
/** 分片读写锁数组:保护单个 {@code EntitySection} 的内容读写({@code ClassInstanceMultiMap})。 */
private static final ReentrantReadWriteLock[] SECTION_LOCKS = new ReentrantReadWriteLock[STRIPE_COUNT];
/**
* 实体分片 Claim 门表:按维度隔离(不同维度的实体循环各自独立 claim,互不掉拍)。
* <p>
* 门状态跨 tick 存续:本 tick 提交的某片未完成前,下 tick 对该片 tryAcquire 失败 →
* 整片掉拍延后(等价"该片本 tick 视为空")。阶段收口 {@link FineGrainScheduler.Barrier#await()}
* 只等待"实际提交的片",慢片不再阻塞维度 worker(借鉴 Tessellate 慢区域掉拍思想)。
*/
private static final ConcurrentHashMap<ResourceKey<?>, ShardGate> GATES = new ConcurrentHashMap<>();
/** 累计掉拍分片数(诊断 / status 展示)。 */
private static final LongAdder MISSED_SHARDS = new LongAdder();
/** 实体分片并行的独立开关(默认开启;关闭时回退旧的批式 submitBatch 路径)。 */
private static volatile boolean sharded = true;
/**
* 当前线程的锁配对栈:记录每次 lock 调用时<b>是否真正加了锁</b>,unlock 时弹出对应决定。
* <p>
@@ -137,19 +157,64 @@ public final class EntityTickParallel {
return CURRENT.get();
}
/** 实体分片并行开关(默认开启;关闭时回退旧的批式 {@code submitBatch} 路径)。 */
public static boolean isSharded() {
return sharded;
}
public static void setSharded(boolean value) {
sharded = value;
ModZuozhi.LOGGER.info("[EntityTick] 实体分片并行已切换为 {}", value ? "开启" : "关闭");
}
/** 累计掉拍分片数(诊断 / {@code status} 展示)。 */
public static long missedShards() {
return MISSED_SHARDS.sum();
}
/**
* 本 tick 收集的一批待并行实体(同一 consumer,即 {@code tickNonPassenger})。
* 收集代替逐实体提交:循环结束后统一按线程池大小切成块提交,把每实体一次的
* 调度/屏障计数开销从 O(实体数) 降到 O(线程数)。
* 收集代替逐实体提交:循环结束后按稳定 id 分片,每片独立提交(分片 Claim 模型),
* 或按线程池大小切成块提交(非分片回退),把每实体一次的调度/屏障计数开销降到 O(线程数)。
*/
public static final class Batch {
public final MinecraftServer server;
public final Consumer<Entity> consumer;
public final List<Entity> entities = new ArrayList<>();
public final ResourceKey<Level> dimension;
/** 按分片存放的实体列表(索引 = entity.getId() & (SHARDS-1))。 */
private final List<Entity>[] shards = new List[ShardGate.SHARDS];
public Batch(MinecraftServer server, Consumer<Entity> consumer) {
public Batch(MinecraftServer server, Consumer<Entity> consumer, ResourceKey<Level> dimension) {
this.server = server;
this.consumer = consumer;
this.dimension = dimension;
}
/** 收集一个实体:按稳定 id 分流到对应分片(id 自增、分布均匀)。 */
public void add(Entity entity) {
int shard = entity.getId() & (ShardGate.SHARDS - 1);
List<Entity> list = this.shards[shard];
if (list == null) {
list = new ArrayList<>();
this.shards[shard] = list;
}
list.add(entity);
}
/** 全部分片数组(end 遍历用)。 */
public List<Entity>[] allShards() {
return this.shards;
}
/** 汇总全量实体列表(仅非分片回退路径使用)。 */
public List<Entity> merged() {
List<Entity> all = new ArrayList<>();
for (List<Entity> shard : this.shards) {
if (shard != null) {
all.addAll(shard);
}
}
return all;
}
}
@@ -157,10 +222,10 @@ public final class EntityTickParallel {
public static void collect(MinecraftServer server, Consumer<Entity> consumer, Entity entity) {
Batch batch = BATCH.get();
if (batch == null) {
batch = new Batch(server, consumer);
batch = new Batch(server, consumer, entity.level().dimension());
BATCH.set(batch);
}
batch.entities.add(entity);
batch.add(entity);
}
/** 实体循环开始:为本维度 tick 建立实体子任务屏障(仅 worker 线程调用)。 */
@@ -169,8 +234,8 @@ public final class EntityTickParallel {
}
/**
* 实体循环结束(finally 中调用):先把收集的实体批按块提交并行,再等待全部子任务完成
* 并执行延迟回调,清除上下文。
* 实体循环结束(finally 中调用):分片模式先按片 Claim 提交(慢片掉拍延后),
* 再等待已提交的子任务全部完成并执行延迟回调,清除上下文。
*/
public static void end() {
FineGrainScheduler.Barrier barrier = CURRENT.get();
@@ -180,10 +245,45 @@ public final class EntityTickParallel {
CURRENT.remove();
Batch batch = BATCH.get();
BATCH.remove();
if (batch != null && !batch.entities.isEmpty()) {
barrier.submitBatch(batch.server, batch.entities, batch.consumer);
if (batch != null) {
if (sharded) {
ShardGate gate = GATES.computeIfAbsent(batch.dimension, key -> new ShardGate());
List<Entity>[] shards = batch.allShards();
for (int s = 0; s < ShardGate.SHARDS; s++) {
List<Entity> entities = shards[s];
if (entities == null || entities.isEmpty()) {
continue;
}
if (gate.tryAcquire(s)) {
barrier.submitShard(batch.server, gate, s,
() -> modzuozhi_tickShard(entities, batch.consumer));
} else {
MISSED_SHARDS.increment();
ModZuozhi.LOGGER.debug("[EntityTick] 实体分片 {} 掉拍(上 tick 未完成),整体延后 1 tick", s);
}
}
} else {
List<Entity> all = batch.merged();
if (!all.isEmpty()) {
barrier.submitBatch(batch.server, all, batch.consumer);
}
}
}
barrier.await();
barrier.drain();
}
/** 分片内逐实体 tick:单实体异常隔离(与批式路径语义一致),异常计入 FaultGuard。 */
private static void modzuozhi_tickShard(List<Entity> entities, Consumer<Entity> consumer) {
for (Entity entity : entities) {
try {
if (!entity.isRemoved()) {
consumer.accept(entity);
}
} catch (Throwable t) {
ModZuozhi.LOGGER.error("[EntityTick] 实体分片子任务异常", t);
FaultGuard.onSubTaskFailure();
}
}
}
}
@@ -247,6 +247,52 @@ public final class FineGrainScheduler {
}
}
/**
* 按分片提交一个子任务(分片 Claim 模型,借鉴 Tessellate 的"每区域 inFlight CAS claim")。
* <p>
* 调用方必须先通过 {@link ShardGate#tryAcquire} 取得本片许可;本方法在任务完成后
* {@link ShardGate#release} 释放 claim。已提交该片的 {@code pending} 计入 barrier,
* 阶段末 {@link #await()} 只等待"实际提交的片"——掉拍(未抢到 claim)的片不在
* pending 内,维度 worker 不会为其阻塞。
* <p>
* 关闭({@code enabled=false})时同步串行执行并释放 claim,行为与原串行路径一致。
*
* @param gate 分片门(跨 tick 存续)
* @param shard 分片索引
* @param task 子任务(如某一片实体的 tick)
*/
public void submitShard(MinecraftServer server, ShardGate gate, int shard, Runnable task) {
if (!FineGrainScheduler.enabled) {
gate.release(shard);
task.run();
return;
}
synchronized (this) {
pending++;
}
ACTIVE_SUBTASKS.incrementAndGet();
SUB_POOL.execute(() -> {
PostExecuteQueue prev = CURRENT_QUEUE.get();
CURRENT_QUEUE.set(postQueue);
try {
task.run();
} catch (Throwable t) {
ModZuozhi.LOGGER.error("[FineGrain] 子任务异常", t);
FaultGuard.onSubTaskFailure();
} finally {
CURRENT_QUEUE.set(prev);
gate.release(shard);
synchronized (this) {
pending--;
if (pending == 0) {
latch.countDown();
}
}
ACTIVE_SUBTASKS.decrementAndGet();
}
});
}
/** 块内逐实体 tick:单实体异常隔离(等价原逐实体提交语义),异常计入 FaultGuard。 */
private void modzuozhi_runChunk(List<Entity> chunk, Consumer<Entity> consumer) {
PostExecuteQueue prev = CURRENT_QUEUE.get();
@@ -0,0 +1,46 @@
package dev.modzuozhi.core.dimthread;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* 分片 Claim 门(借鉴 Tessellate 的"每区域一个 inFlight CAS claim、慢区掉拍"思想独立实现)。
* <p>
* 一片在同一时刻至多允许一个任务运行,claim 状态跨 tick 存续:本 tick 提交的任务在
* {@link #tryAcquire} 之后、任务完成 {@link #release} 之前,下 tick 对该片的抢占必然失败 →
* 整片掉拍延后(等价"该片本 tick 视为空",与 MC 实体密度限制同性质,无感知)。
* <p>
* 收益:阶段收口从"等全部子任务"(慢片拖累全体)变成"只等已提交且上一 tick 已完成过的片",
* 慢片不再阻塞维度 worker。
* <p>
* 线程安全:仅使用 CAS 置位,无锁、无等待。
*/
public final class ShardGate {
/**
* 分片数:2 的幂(供 & 掩码),最低 8,按 CPU 核数线性扩展(≈ 2×核数)。
*/
public static final int SHARDS = Integer.highestOneBit(Math.max(8, Runtime.getRuntime().availableProcessors() * 2));
/** 每片一个 claim 位(false=空闲,true=本 tick 有任务在跑/已提交)。 */
private final AtomicBoolean[] gates = new AtomicBoolean[SHARDS];
public ShardGate() {
for (int i = 0; i < SHARDS; i++) {
gates[i] = new AtomicBoolean(false);
}
}
/**
* CAS false→true 抢占本片。
*
* @return true=允许提交(本片空闲,已占用);false=上一 tick 该片任务尚未完成(掉拍)。
*/
public boolean tryAcquire(int shard) {
return gates[shard].compareAndSet(false, true);
}
/** 本片任务完成时释放 claim,允许下 tick 重新提交。 */
public void release(int shard) {
gates[shard].set(false);
}
}
@@ -1,8 +1,7 @@
package dev.modzuozhi.mixin;
import dev.modzuozhi.core.collection.ConcurrentLong2ObjectMap;
import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
import it.unimi.dsi.fastutil.longs.Long2ObjectMaps;
import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
import net.minecraft.world.level.chunk.storage.SectionStorage;
import org.spongepowered.asm.mixin.Mixin;
import org.spongepowered.asm.mixin.Shadow;
@@ -19,13 +18,19 @@ import java.util.Optional;
* {@code ArrayIndexOutOfBoundsException: Index -1 out of bounds ... rehash}),
* 进而引发 chunk 加载失败、连锁异常。
* <p>
* 修复:把 {@code storage} 替换为 {@link Long2ObjectMaps#synchronize 同步包装},所有
* get/put/remove/values 访问线程安全,从根上消除该数据竞争。
* 旧修复:把 {@code storage} 替换为 {@code Long2ObjectMaps.synchronize} 同步包装,数据安全了,
* 但<b>全表互斥</b>把村民 POI 查询(读多写少、原本可并行)串行化:19000 只村民压测下
* 19 个子任务线程 16 个 BLOCKED 在同一把锁上,成为并行 tick 的串行瓶颈(TPS 9 / CPU 半闲)。
* <p>
* 新修复:换成 {@link ConcurrentLong2ObjectMap}({@code ConcurrentHashMap} 底座,
* {@code SectionStorage} 对 storage 的全部调用 get/put/remove/size 均已覆盖)——
* 读路径无锁并行,写路径(区块加载 put、方块变化)分段安全,弱一致迭代兼容
* {@code writeColumn} 的读取。POI 查询恢复并行,消除该锁地板。
*/
@Mixin(SectionStorage.class)
public abstract class SectionStorageMixin<R> {
@Shadow
private final Long2ObjectMap<Optional<R>> storage =
Long2ObjectMaps.synchronize(new Long2ObjectOpenHashMap<>());
new ConcurrentLong2ObjectMap<>();
}