perf(payload): scope payload ownership to affected dimensions only
问题:多玩家服务器上,每个 mod payload(NeoForge ServerPayloadContext.enqueueWork) 处理都走 MainThreadPhaseCoordinator 的全局独占——挂起并等待【全部】维度 tick 停下。 跑图时维度 tick 是长任务,主线程 managedBlock 自旋等待达数秒(日志 oldestMs=5917), 全服掉到 10 TPS。而 payload 实际只碰它所属玩家的那个维度。 修法: - PayloadWorkQueue.Entry 增 dimension 字段(submit 时取 player.serverLevel) - 新增 pendingDimensions() / drainBatchByDimension():按维度分组处理 - MainThreadPhaseCoordinator.runScoped():只挂起/等待指定维度,不设全局 EXCLUSIVE (其它维度 blocksWorkerStart 仍 false,照常自排程) - PayloadMainThreadDispatch.drain 改走 runScoped(涉及维度),只暂停受影响维度 效果:多玩家时玩家 A(overworld) 的 payload 只停 overworld,nether/end 等继续 tick。 注:这是 payload 路径的优化,未改变维度 tick 本身不可中断的性质。
This commit is contained in:
1 parent
d76d6593bd
commit
490be29f57
3 files changed
+118
-14
No files matched your search
@@ -2,7 +2,12 @@ package dev.modzuozhi.core.compat;
|
||||
|
||||
import dev.modzuozhi.core.dimthread.DimThreadCore;
|
||||
import dev.modzuozhi.core.dimthread.LevelTickLoop;
|
||||
import net.minecraft.resources.ResourceKey;
|
||||
import net.minecraft.server.MinecraftServer;
|
||||
import net.minecraft.world.level.Level;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
@@ -38,6 +43,7 @@ public final class MainThreadPhaseCoordinator {
|
||||
public static void updatePause(MinecraftServer server, Runnable change) {
|
||||
synchronized(state(server)){change.run();}
|
||||
}
|
||||
|
||||
public static void run(MinecraftServer server, String reason, Runnable action) {
|
||||
if (!server.isSameThread()) throw new IllegalStateException("Compatibility phase requires server thread");
|
||||
State s = state(server);
|
||||
@@ -73,7 +79,40 @@ public final class MainThreadPhaseCoordinator {
|
||||
}
|
||||
}
|
||||
}
|
||||
private static java.util.List<LevelTickLoop> loops(MinecraftServer server) {
|
||||
|
||||
/**
|
||||
* 按维度作用域的独占:只挂起并等待 {@code scopeDimensions} 中的维度,其余维度继续并行 tick。
|
||||
* <p>
|
||||
* <b>不设置全局 EXCLUSIVE 相位</b>——因此其它维度不会被 {@link #blocksWorkerStart} 波及,
|
||||
* 照常自排程。适用于「只碰某几个维度」的场景(如某玩家的 payload)。
|
||||
* <p>
|
||||
* 若 {@code scopeDimensions} 为 null 或空,退化为 {@link #run(MinecraftServer, String, Runnable)}(全维度独占)。
|
||||
*/
|
||||
public static void runScoped(MinecraftServer server, String reason,
|
||||
Set<ResourceKey<Level>> scopeDimensions, Runnable action) {
|
||||
if (!server.isSameThread()) throw new IllegalStateException("Compatibility phase requires server thread");
|
||||
if (scopeDimensions == null || scopeDimensions.isEmpty()) { run(server, reason, action); return; }
|
||||
if (owns(server)) { action.run(); return; }
|
||||
if (SableMainThreadBridge.inHandoff(server))
|
||||
throw new IllegalStateException("Scoped phase inside synchronous Sable handoff would deadlock: " + reason);
|
||||
List<LevelTickLoop> affected = loops(server).stream()
|
||||
.filter(l -> scopeDimensions.contains(l.getLevel().dimension())).toList();
|
||||
affected.forEach(loop -> loop.holdForMainThread(true));
|
||||
long began = System.nanoTime();
|
||||
try {
|
||||
server.managedBlock(() -> {
|
||||
boolean idle = affected.stream().noneMatch(LevelTickLoop::isTicking);
|
||||
if (!idle) ThreadOwnershipDiagnostics.slowPhase(server, reason, began);
|
||||
return idle;
|
||||
});
|
||||
action.run();
|
||||
} finally {
|
||||
ThreadOwnershipDiagnostics.phaseTiming(reason, began, System.nanoTime());
|
||||
affected.forEach(loop -> loop.holdForMainThread(false));
|
||||
}
|
||||
}
|
||||
|
||||
private static List<LevelTickLoop> loops(MinecraftServer server) {
|
||||
return DimThreadCore.MANAGER.ownershipLoopsSnapshot().stream()
|
||||
.filter(loop -> loop.getLevel().getServer() == server).toList();
|
||||
}
|
||||
@@ -82,4 +121,4 @@ public final class MainThreadPhaseCoordinator {
|
||||
synchronized (s) { s.phase = Phase.STOPPING; loops(server).forEach(loop -> loop.holdForMainThread(true)); }
|
||||
}
|
||||
public static void clear(MinecraftServer server) { STATES.remove(server); }
|
||||
}
|
||||
}
|
||||
@@ -1,9 +1,13 @@
|
||||
package dev.modzuozhi.core.compat;
|
||||
|
||||
import net.minecraft.network.Connection;
|
||||
import net.minecraft.resources.ResourceKey;
|
||||
import net.minecraft.resources.ResourceLocation;
|
||||
import net.minecraft.server.MinecraftServer;
|
||||
import net.minecraft.server.level.ServerPlayer;
|
||||
import net.minecraft.world.level.Level;
|
||||
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
@@ -36,8 +40,9 @@ public final class PayloadMainThreadDispatch {
|
||||
if(closed(server)) return CompletableFuture.failedFuture(new StalePayload("Server stopping"));
|
||||
queue = QUEUES.computeIfAbsent(server, s -> new PayloadWorkQueue<>(System::nanoTime));
|
||||
}
|
||||
ResourceKey<Level> dimension = level.dimension();
|
||||
var ticket=PayloadDiagnostics.begin(server,payloadId.toString());
|
||||
CompletableFuture<T> result = queue.submit(listener.getConnection(), () -> {
|
||||
CompletableFuture<T> result = queue.submit(listener.getConnection(), dimension, () -> {
|
||||
if (player.connection != listener || listener.player != player || player.hasDisconnected()
|
||||
|| player.isRemoved() || player.serverLevel() != level)
|
||||
throw new StalePayload("Payload player/world changed before execution");
|
||||
@@ -50,8 +55,6 @@ public final class PayloadMainThreadDispatch {
|
||||
if(!closed(server)) OVERLOADED.computeIfAbsent(server,s -> new ConcurrentHashMap<>()).putIfAbsent(listener.getConnection(),player);
|
||||
}
|
||||
});
|
||||
// Preserve synchronous main-thread enqueueWork semantics and observed FIFO order.
|
||||
// Each acquisition still obeys the per-batch budget. Never drain recursively while pumping callbacks.
|
||||
if (server.isSameThread() && !MainThreadPhaseCoordinator.blocksWorkerStart(server)) {
|
||||
while (!result.isDone()) drain(server);
|
||||
try { result.join(); }
|
||||
@@ -63,6 +66,7 @@ public final class PayloadMainThreadDispatch {
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public static void drain(MinecraftServer server) {
|
||||
if (!server.isSameThread()) throw new IllegalStateException("Payload drain requires server thread");
|
||||
var queue = QUEUES.get(server);
|
||||
@@ -71,11 +75,12 @@ public final class PayloadMainThreadDispatch {
|
||||
if (MainThreadPhaseCoordinator.blocksWorkerStart(server)
|
||||
|| SableMainThreadBridge.inHandoff(server)) return;
|
||||
long start=System.nanoTime();
|
||||
MainThreadPhaseCoordinator.run(server, "custom payloads", () -> {
|
||||
// 精细化独占:只暂停本批 payload 涉及的维度,其余维度继续并行 tick。
|
||||
// 多玩家时玩家 A(overworld) 的 payload 不再停 nether/end 等,反之亦然。
|
||||
Set<ResourceKey<Level>> dims = (queue != null) ? queue.pendingDimensions() : Set.of();
|
||||
MainThreadPhaseCoordinator.runScoped(server, "custom payloads", dims, () -> {
|
||||
long acquired=System.nanoTime();
|
||||
try {
|
||||
// Cleanup goes through the original disconnect lifecycle with world ownership.
|
||||
// A cleanup-only batch avoids bypassing the payload batch count/time budget.
|
||||
if(overloaded!=null && !overloaded.isEmpty()) {
|
||||
int count=0;
|
||||
for(var entry:overloaded.entrySet()) {
|
||||
@@ -86,7 +91,7 @@ public final class PayloadMainThreadDispatch {
|
||||
player.connection.disconnect(net.minecraft.network.chat.Component.literal("Too many pending mod interactions; please reconnect."));
|
||||
}
|
||||
}
|
||||
} else if(queue!=null) queue.drainBatch();
|
||||
} else if(queue!=null) queue.drainBatchByDimension(dims);
|
||||
} finally {PayloadDiagnostics.batch(server,start,acquired);}
|
||||
});
|
||||
}
|
||||
@@ -103,7 +108,6 @@ public final class PayloadMainThreadDispatch {
|
||||
/** NeoForge guard's null-on-handler-error contract, with bounded diagnostics already attached to raw result. */
|
||||
public static <T> CompletableFuture<T> guard(CompletableFuture<T> result) {
|
||||
CompletableFuture<T> guarded=result.handle((value,error) -> {
|
||||
// Preserve NeoForge's handler-error result, not false success for queue policy failures.
|
||||
if(error instanceof StalePayload || error instanceof PayloadWorkQueue.CapacityExceeded || result.isCancelled())
|
||||
throw new CompletionException(error);
|
||||
return error==null ? value : null;
|
||||
@@ -111,4 +115,4 @@ public final class PayloadMainThreadDispatch {
|
||||
guarded.whenComplete((value,error) -> {if(guarded.isCancelled())result.cancel(false);});
|
||||
return guarded;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,8 @@
|
||||
package dev.modzuozhi.core.compat;
|
||||
|
||||
import net.minecraft.resources.ResourceKey;
|
||||
import net.minecraft.world.level.Level;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.RejectedExecutionException;
|
||||
@@ -20,9 +23,19 @@ public final class PayloadWorkQueue<K> {
|
||||
public PayloadWorkQueue(LongSupplier nanoTime) { clock = Objects.requireNonNull(nanoTime); }
|
||||
|
||||
public <T> CompletableFuture<T> submit(K connection, Supplier<T> action) {
|
||||
return submit(connection, null, action);
|
||||
}
|
||||
|
||||
/**
|
||||
* 提交一个绑定到具体维度的 payload 任务。
|
||||
* <p>
|
||||
* {@code dimension} 为该 payload 所属玩家当前维度,用于 {@link #drainBatchByDimension}
|
||||
* 按维度精细化独占——多玩家时只暂停受影响维度,而非全部维度。
|
||||
*/
|
||||
public <T> CompletableFuture<T> submit(K connection, ResourceKey<Level> dimension, Supplier<T> action) {
|
||||
Objects.requireNonNull(connection);
|
||||
Objects.requireNonNull(action);
|
||||
Entry<T> entry = new Entry<>(connection, action);
|
||||
Entry<T> entry = new Entry<>(connection, dimension, action);
|
||||
Throwable rejected;
|
||||
synchronized (this) {
|
||||
rejected = closed;
|
||||
@@ -67,6 +80,51 @@ public final class PayloadWorkQueue<K> {
|
||||
} finally { synchronized (this) { draining = false; } }
|
||||
}
|
||||
|
||||
/** 当前待处理任务涉及的维度集合快照(用于按维度精细化独占)。 */
|
||||
public synchronized Set<ResourceKey<Level>> pendingDimensions() {
|
||||
Set<ResourceKey<Level>> dims = new HashSet<>();
|
||||
for (Entry<?> entry : pending) {
|
||||
if (entry.dimension != null) dims.add(entry.dimension);
|
||||
}
|
||||
return dims;
|
||||
}
|
||||
|
||||
/**
|
||||
* 只处理属于指定维度子集的待办任务(其余维度任务留在队列,由各自批次处理)。
|
||||
* <p>
|
||||
* 供 {@code PayloadMainThreadDispatch} 在「仅暂停受影响维度」后调用:锁定世界所有权时
|
||||
* 只暂停了这些维度,故只能执行这些维度的 payload,避免在其它维度仍在并行 tick 时并发访问。
|
||||
*/
|
||||
public int drainBatchByDimension(Set<ResourceKey<Level>> allowedDimensions) {
|
||||
if (allowedDimensions.isEmpty()) return drainBatch();
|
||||
synchronized (this) {
|
||||
if (draining || closed != null) return 0;
|
||||
draining = true;
|
||||
}
|
||||
int ran = 0;
|
||||
long start = clock.getAsLong();
|
||||
try {
|
||||
while (ran < 64 && clock.getAsLong() - start < 2_000_000L) {
|
||||
Entry<?> entry;
|
||||
synchronized (this) {
|
||||
if (closed != null || pending.isEmpty()) break;
|
||||
entry = null;
|
||||
for (Entry<?> candidate : pending) {
|
||||
if (candidate.dimension == null || allowedDimensions.contains(candidate.dimension)) {
|
||||
entry = candidate;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (entry == null) break; // 没有可处理的(其余维度任务留待下批)
|
||||
remove(entry);
|
||||
}
|
||||
entry.run();
|
||||
ran++;
|
||||
}
|
||||
return ran;
|
||||
} finally { synchronized (this) { draining = false; } }
|
||||
}
|
||||
|
||||
public void close(Throwable reason) {
|
||||
Objects.requireNonNull(reason);
|
||||
List<Entry<?>> abandoned;
|
||||
@@ -83,13 +141,16 @@ public final class PayloadWorkQueue<K> {
|
||||
|
||||
private final class Entry<T> {
|
||||
final K connection;
|
||||
final ResourceKey<Level> dimension;
|
||||
final Supplier<T> action;
|
||||
final CompletableFuture<T> result = new CompletableFuture<>();
|
||||
Entry(K connection, Supplier<T> action) { this.connection = connection; this.action = action; }
|
||||
Entry(K connection, ResourceKey<Level> dimension, Supplier<T> action) {
|
||||
this.connection = connection; this.dimension = dimension; this.action = action;
|
||||
}
|
||||
void run() {
|
||||
if (result.isCancelled()) return;
|
||||
try { result.complete(action.get()); }
|
||||
catch (Throwable error) { result.completeExceptionally(error); }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user