fix: improve threaded world compatibility and chunk reads in 1.1.15

Add regression tests and per-dimension progress diagnostics. Document the unresolved C2ME random-initialization redirect conflict.
This commit is contained in:
Disy committed 2026-10-03 21:24:31 +08:00
1 parent 74cae0b581
commit 32b11aee87
42 files changed
+1506 -381

No files matched your search

+83
View File
@@ -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<Runnable> serverTasks = new LinkedBlockingQueue<>();
CompletableFuture<Integer> chunk = new CompletableFuture<>();
Thread worker = new Thread(() -> {
try {
CompletableFuture<Void> 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<Object> 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<Runnable> chunkTasks = new LinkedBlockingQueue<>();
java.util.concurrent.atomic.AtomicReference<Thread> 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");
}
}
+16
View File
@@ -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");
}
}
+65
View File
@@ -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");
}
}
+36
View File
@@ -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); }
}
@@ -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);
}
}
}
@@ -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<Throwable>();
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");
}
}