Drained pregen before shutdown

- Server shutdown advances native chunk tasks while waiting for pregeneration
- Native world close owns chunk saves during server shutdown
- Pending plugin flushes stop waiting when native shutdown begins
- Late eviction completion cannot schedule another shutdown flush
- Regression tests cover owner waits and shutdown save ownership
This commit is contained in:
Brian Neumann-Fopiano
2026-09-06 18:20:26 -04:00
parent 319f8aed47
commit 083e689864
6 changed files with 420 additions and 4 deletions
@@ -1037,7 +1037,16 @@ public class Iris extends VolmitPlugin implements Listener, ReloadAware {
}
try {
PregeneratorJob.shutdownAndWait(SERVER_STOP_PREGEN_TIMEOUT_MILLIS);
if (J.isFolia()) {
PregeneratorJob.shutdownAndWait(SERVER_STOP_PREGEN_TIMEOUT_MILLIS);
} else {
List<World> worlds = Bukkit.getWorlds();
PregeneratorJob.shutdownAndWait(SERVER_STOP_PREGEN_TIMEOUT_MILLIS, () -> {
for (World world : worlds) {
INMS.get().pollChunkTask(world);
}
});
}
} catch (Throwable e) {
Iris.reportError("Failed to quiesce the Iris pregenerator before server shutdown.", e);
}
@@ -52,7 +52,10 @@ public class BukkitEngineLifecycleContractTest {
String waitedShutdown = method(jobSource,
"private static boolean shutdownAndWait(PregeneratorJob inst, long timeoutMs)");
assertBefore(waitedShutdown, "inst.requestStop()", "inst.worker.join(");
assertTrue(waitedShutdown.contains("inst.worker.isAlive()"));
assertBefore(waitedShutdown, "inst.worker.join(", "finishShutdown(inst, timeoutMs)");
String finishedShutdown = method(jobSource,
"private static boolean finishShutdown(PregeneratorJob inst, long timeoutMs)");
assertBefore(finishedShutdown, "inst.worker.isAlive()", "instance.compareAndSet(inst, null)");
String hooksSource = Files.readString(Path.of(System.getProperty("iris.bukkitEnginePlatformHooksSource")));
String hookShutdown = method(hooksSource, "public void shutdownPregenerator(Engine engine)");
@@ -208,6 +208,28 @@ public class PregeneratorJob implements PregenListener, PregenRenderSource {
return shutdownAndWait(inst, timeoutMs);
}
public static boolean shutdownAndWait(long timeoutMs, Runnable processPendingTasks) {
Objects.requireNonNull(processPendingTasks, "processPendingTasks");
PregeneratorJob inst = instance.get();
if (inst == null) {
return false;
}
inst.requestStop();
long started = System.nanoTime();
long timeoutNanos = TimeUnit.MILLISECONDS.toNanos(Math.max(1L, timeoutMs));
try {
while (inst.worker.isAlive() && System.nanoTime() - started < timeoutNanos) {
processPendingTasks.run();
inst.worker.join(1L);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException("Interrupted while stopping the Iris pregenerator.", e);
}
return finishShutdown(inst, timeoutMs);
}
private static boolean shutdownAndWait(PregeneratorJob inst, long timeoutMs) {
inst.requestStop();
try {
@@ -216,6 +238,10 @@ public class PregeneratorJob implements PregenListener, PregenRenderSource {
Thread.currentThread().interrupt();
throw new IllegalStateException("Interrupted while stopping the Iris pregenerator.", e);
}
return finishShutdown(inst, timeoutMs);
}
private static boolean finishShutdown(PregeneratorJob inst, long timeoutMs) {
if (inst.worker.isAlive()) {
throw new IllegalStateException("Timed out while stopping the Iris pregenerator after "
+ Math.max(1L, timeoutMs) + "ms.");
@@ -357,6 +357,9 @@ public class AsyncPregenMethod implements PregeneratorMethod {
}
private CompletableFuture<Void> evictRegion(long c) {
if (IrisToolbelt.isServerStopping()) {
return CompletableFuture.completedFuture(null);
}
if (!evictedRegions.add(c)) {
return CompletableFuture.completedFuture(null);
}
@@ -441,6 +444,9 @@ public class AsyncPregenMethod implements PregeneratorMethod {
}
private void flushAllRemainingChunks() {
if (IrisToolbelt.isServerStopping()) {
return;
}
List<Long> keys = new ArrayList<>(regionChunks.keySet());
List<CompletableFuture<Void>> evictions = new ArrayList<>(keys.size() + pendingEvictions.size());
for (Long regionKey : keys) {
@@ -453,10 +459,27 @@ public class AsyncPregenMethod implements PregeneratorMethod {
CompletableFuture<Void> settledEvictions = evictionsComplete.handle((ignored, failure) -> null);
CompletableFuture<Void> flush = foliaRuntime
? settledEvictions
: settledEvictions.thenCompose(ignored -> J.sfut(() -> INMS.get().flushChunkIO(world)));
: settledEvictions.thenCompose(ignored -> IrisToolbelt.isServerStopping()
? CompletableFuture.completedFuture(null)
: J.sfut(() -> INMS.get().flushChunkIO(world)));
long started = System.nanoTime();
long timeoutNanos = TimeUnit.SECONDS.toNanos(FLUSH_TIMEOUT_SECONDS);
try {
flush.get(FLUSH_TIMEOUT_SECONDS, TimeUnit.SECONDS);
while (!IrisToolbelt.isServerStopping()) {
long remaining = timeoutNanos - (System.nanoTime() - started);
if (remaining <= 0L) {
throw new TimeoutException();
}
try {
flush.get(Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(100L)), TimeUnit.NANOSECONDS);
return;
} catch (TimeoutException e) {
if (System.nanoTime() - started >= timeoutNanos) {
throw e;
}
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
IrisLogging.warn("Interrupted while flushing pregen chunks for " + world.getName() + ".");
@@ -0,0 +1,134 @@
package art.arcane.iris.core.gui;
import art.arcane.iris.core.pregenerator.IrisPregenerator;
import art.arcane.volmlib.util.format.MemoryMonitor;
import org.junit.Test;
import java.lang.reflect.Field;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
public class PregeneratorJobOwnerDrainTest {
@Test
public void ownerPollingCompletesTheRequestThatKeepsTheWorkerAlive() throws Exception {
try (Fixture fixture = new Fixture()) {
Thread owner = Thread.currentThread();
AtomicInteger polls = new AtomicInteger();
assertTrue(PregeneratorJob.shutdownAndWait(1_000L, () -> {
assertSame(owner, Thread.currentThread());
polls.incrementAndGet();
fixture.completion.complete(null);
}));
assertTrue(polls.get() > 0);
assertFalse(fixture.worker.isAlive());
assertNull(PregeneratorJob.getInstance());
verify(fixture.pregenerator).close();
}
}
@Test
public void timeoutKeepsThePendingJobAndDoesNotCompleteItsRequest() throws Exception {
try (Fixture fixture = new Fixture()) {
AtomicInteger polls = new AtomicInteger();
IllegalStateException failure = assertThrows(IllegalStateException.class,
() -> PregeneratorJob.shutdownAndWait(20L, polls::incrementAndGet));
assertEquals("Timed out while stopping the Iris pregenerator after 20ms.", failure.getMessage());
assertTrue(polls.get() > 0);
assertTrue(fixture.worker.isAlive());
assertFalse(fixture.completion.isDone());
assertSame(fixture.job, PregeneratorJob.getInstance());
}
}
@Test
public void interruptedOwnerKeepsTheJobAndOriginalInterruption() throws Exception {
try (Fixture fixture = new Fixture()) {
try {
IllegalStateException failure = assertThrows(IllegalStateException.class,
() -> PregeneratorJob.shutdownAndWait(1_000L, () -> Thread.currentThread().interrupt()));
assertTrue(failure.getCause() instanceof InterruptedException);
assertTrue(Thread.currentThread().isInterrupted());
assertSame(fixture.job, PregeneratorJob.getInstance());
assertFalse(fixture.completion.isDone());
} finally {
Thread.interrupted();
}
}
}
@Test
public void ownerTaskFailureEscapesWithItsCauseAndRetainsTheJob() throws Exception {
try (Fixture fixture = new Fixture()) {
IllegalStateException expected = new IllegalStateException("Native task failed", new Exception("Chunk cause"));
IllegalStateException actual = assertThrows(IllegalStateException.class,
() -> PregeneratorJob.shutdownAndWait(1_000L, () -> {
throw expected;
}));
assertSame(expected, actual);
assertSame(expected.getCause(), actual.getCause());
assertSame(fixture.job, PregeneratorJob.getInstance());
assertFalse(fixture.completion.isDone());
}
}
private static final class Fixture implements AutoCloseable {
private final PregeneratorJob job = mock(PregeneratorJob.class, CALLS_REAL_METHODS);
private final IrisPregenerator pregenerator = mock(IrisPregenerator.class);
private final CompletableFuture<Void> completion = new CompletableFuture<>();
private final AtomicReference<PregeneratorJob> instance;
private final PregeneratorJob previous;
private final Thread worker;
private Fixture() throws Exception {
CountDownLatch started = new CountDownLatch(1);
worker = new Thread(() -> {
started.countDown();
completion.join();
}, "Pregen owner drain test");
set("worker", worker);
set("pregenerator", pregenerator);
set("monitor", mock(MemoryMonitor.class));
set("stopRequested", new AtomicBoolean());
set("bounds", new PregenRenderSnapshot.Bounds(0, 0, 0, 0));
instance = instance();
previous = instance.getAndSet(job);
worker.start();
assertTrue(started.await(5L, TimeUnit.SECONDS));
}
@Override
public void close() throws Exception {
completion.complete(null);
worker.join(5_000L);
instance.set(previous);
assertFalse(worker.isAlive());
}
private void set(String name, Object value) throws Exception {
Field field = PregeneratorJob.class.getDeclaredField(name);
field.setAccessible(true);
field.set(job, value);
}
@SuppressWarnings("unchecked")
private static AtomicReference<PregeneratorJob> instance() throws Exception {
Field field = PregeneratorJob.class.getDeclaredField("instance");
field.setAccessible(true);
return (AtomicReference<PregeneratorJob>) field.get(null);
}
}
}
@@ -0,0 +1,221 @@
package art.arcane.iris.core.pregenerator.methods;
import art.arcane.iris.core.nms.INMS;
import art.arcane.iris.core.nms.INMSBinding;
import art.arcane.iris.core.tools.IrisToolbelt;
import art.arcane.iris.platform.bukkit.BukkitPlatform;
import art.arcane.iris.util.common.scheduling.J;
import art.arcane.volmlib.util.collection.KSet;
import org.bukkit.Chunk;
import org.bukkit.World;
import org.bukkit.plugin.Plugin;
import org.junit.Test;
import org.mockito.MockedStatic;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
public class AsyncPregenServerSaveOwnershipTest {
@Test
public void serverStopRetainsChunksForNativeSaveWithoutSchedulingEviction() throws Exception {
Fixture fixture = new Fixture();
try (MockedStatic<IrisToolbelt> toolbelt = mockStatic(IrisToolbelt.class);
MockedStatic<J> scheduling = mockStatic(J.class)) {
toolbelt.when(IrisToolbelt::isServerStopping).thenReturn(true);
CompletableFuture<?> eviction = (CompletableFuture<?>) invoke(fixture.method, "evictRegion", 0L);
assertTrue(eviction.isDone());
assertFalse(eviction.isCompletedExceptionally());
assertSame(fixture.chunk, fixture.chunks.get(0L).peek());
scheduling.verifyNoInteractions();
verifyNoInteractions(fixture.world, fixture.binding);
}
}
@Test
public void serverStopDoesNotWaitForAnUnresolvedPluginEviction() throws Exception {
Fixture fixture = new Fixture();
ExecutorService executor = Executors.newSingleThreadExecutor();
try {
Future<?> flushed = executor.submit(() -> {
try (MockedStatic<IrisToolbelt> toolbelt = mockStatic(IrisToolbelt.class);
MockedStatic<J> scheduling = mockStatic(J.class)) {
toolbelt.when(IrisToolbelt::isServerStopping).thenReturn(true);
scheduling.when(() -> J.sfut(any(Runnable.class)))
.thenReturn(CompletableFuture.completedFuture(null));
invoke(fixture.method, "flushAllRemainingChunks");
scheduling.verifyNoInteractions();
}
return null;
});
flushed.get(1L, TimeUnit.SECONDS);
assertFalse(fixture.pending.isDone());
assertSame(fixture.chunk, fixture.chunks.get(0L).peek());
verifyNoInteractions(fixture.world, fixture.binding);
} finally {
fixture.pending.complete(null);
executor.shutdownNow();
assertTrue(executor.awaitTermination(5L, TimeUnit.SECONDS));
}
}
@Test
public void serverStopDuringFlushWaitReturnsWithoutSchedulingALateFlush() throws Exception {
Fixture fixture = new Fixture();
AtomicBoolean stopping = new AtomicBoolean();
CountDownLatch firstFlush = new CountDownLatch(1);
doAnswer(invocation -> {
firstFlush.countDown();
return null;
}).when(fixture.binding).flushChunkIO(fixture.world);
ExecutorService executor = Executors.newSingleThreadExecutor();
try (SchedulingContext owner = new SchedulingContext(fixture)) {
owner.toolbelt.when(IrisToolbelt::isServerStopping).thenAnswer(invocation -> stopping.get());
Future<?> flushed = executor.submit(() -> {
try (SchedulingContext context = new SchedulingContext(fixture)) {
context.toolbelt.when(IrisToolbelt::isServerStopping).thenAnswer(invocation -> stopping.get());
invoke(fixture.method, "flushAllRemainingChunks");
}
return null;
});
assertTrue(firstFlush.await(5L, TimeUnit.SECONDS));
assertThrows(TimeoutException.class, () -> flushed.get(30L, TimeUnit.MILLISECONDS));
stopping.set(true);
flushed.get(1L, TimeUnit.SECONDS);
assertFalse(fixture.pending.isDone());
verify(fixture.binding).flushChunkIO(fixture.world);
fixture.pending.complete(null);
owner.scheduling.verifyNoInteractions();
verify(fixture.binding).flushChunkIO(fixture.world);
} finally {
executor.shutdownNow();
assertTrue(executor.awaitTermination(5L, TimeUnit.SECONDS));
}
}
@Test
public void ordinaryFlushWaitsForExistingEvictionsAndFlushesNewChunks() throws Exception {
Fixture fixture = new Fixture();
CountDownLatch firstFlush = new CountDownLatch(1);
doAnswer(invocation -> {
firstFlush.countDown();
return null;
}).when(fixture.binding).flushChunkIO(fixture.world);
ExecutorService executor = Executors.newSingleThreadExecutor();
try (SchedulingContext ignored = new SchedulingContext(fixture)) {
Future<?> flushed = executor.submit(() -> {
try (SchedulingContext context = new SchedulingContext(fixture)) {
invoke(fixture.method, "flushAllRemainingChunks");
}
return null;
});
assertTrue(firstFlush.await(5L, TimeUnit.SECONDS));
assertFalse(flushed.isDone());
assertFalse(fixture.pending.isDone());
verify(fixture.binding).saveAndUnloadChunk(fixture.world, 2, 3);
verify(fixture.binding).flushChunkIO(fixture.world);
fixture.pending.complete(null);
flushed.get(5L, TimeUnit.SECONDS);
verify(fixture.binding, times(2)).flushChunkIO(fixture.world);
assertTrue(fixture.chunks.isEmpty());
} finally {
fixture.pending.complete(null);
executor.shutdownNow();
assertTrue(executor.awaitTermination(5L, TimeUnit.SECONDS));
}
}
private static Object invoke(AsyncPregenMethod target, String name, Object... arguments) throws Exception {
Method method = arguments.length == 0
? AsyncPregenMethod.class.getDeclaredMethod(name)
: AsyncPregenMethod.class.getDeclaredMethod(name, long.class);
method.setAccessible(true);
try {
return method.invoke(target, arguments);
} catch (InvocationTargetException failure) {
throw new IllegalStateException(failure.getCause());
}
}
private static final class Fixture {
private final AsyncPregenMethod method = mock(AsyncPregenMethod.class, CALLS_REAL_METHODS);
private final World world = mock(World.class);
private final Chunk chunk = mock(Chunk.class);
private final INMSBinding binding = mock(INMSBinding.class);
private final ConcurrentHashMap<Long, Queue<Chunk>> chunks = new ConcurrentHashMap<>();
private final CompletableFuture<Void> pending = new CompletableFuture<>();
private Fixture() throws Exception {
Queue<Chunk> region = new ConcurrentLinkedQueue<>();
region.add(chunk);
chunks.put(0L, region);
Queue<CompletableFuture<Void>> evictions = new ConcurrentLinkedQueue<>();
evictions.add(pending);
set("world", world);
set("regionChunks", chunks);
set("regionPending", new ConcurrentHashMap<>());
set("pendingEvictions", evictions);
set("evictedRegions", new KSet<>());
when(chunk.getX()).thenReturn(2);
when(chunk.getZ()).thenReturn(3);
when(binding.saveAndUnloadChunk(world, 2, 3)).thenReturn(true);
}
private void set(String name, Object value) throws Exception {
Field field = AsyncPregenMethod.class.getDeclaredField(name);
field.setAccessible(true);
field.set(method, value);
}
}
private static final class SchedulingContext implements AutoCloseable {
private final MockedStatic<IrisToolbelt> toolbelt = mockStatic(IrisToolbelt.class);
private final MockedStatic<J> scheduling = mockStatic(J.class);
private final MockedStatic<INMS> nativeAccess = mockStatic(INMS.class);
private final MockedStatic<BukkitPlatform> platform = mockStatic(BukkitPlatform.class);
private SchedulingContext(Fixture fixture) {
nativeAccess.when(INMS::get).thenReturn(fixture.binding);
platform.when(BukkitPlatform::plugin).thenReturn(mock(Plugin.class));
scheduling.when(() -> J.sfut(any(Runnable.class))).thenAnswer(invocation -> {
Runnable task = invocation.getArgument(0);
task.run();
return CompletableFuture.completedFuture(null);
});
}
@Override
public void close() {
platform.close();
nativeAccess.close();
scheduling.close();
toolbelt.close();
}
}
}