diff --git a/adapters/bukkit/nms/v26_2_R1/src/main/java/art/arcane/iris/core/nms/v26_2_R1/NMSBinding.java b/adapters/bukkit/nms/v26_2_R1/src/main/java/art/arcane/iris/core/nms/v26_2_R1/NMSBinding.java index 58d017a62..70cc3fa7b 100644 --- a/adapters/bukkit/nms/v26_2_R1/src/main/java/art/arcane/iris/core/nms/v26_2_R1/NMSBinding.java +++ b/adapters/bukkit/nms/v26_2_R1/src/main/java/art/arcane/iris/core/nms/v26_2_R1/NMSBinding.java @@ -38,6 +38,7 @@ import art.arcane.volmlib.util.json.JSONObject; import art.arcane.volmlib.util.mantle.runtime.Mantle; import art.arcane.volmlib.util.matter.Matter; import art.arcane.volmlib.util.math.Vector3d; +import art.arcane.volmlib.util.scheduling.FoliaScheduler; import art.arcane.volmlib.util.matter.MatterBiomeInject; import art.arcane.iris.spi.PlatformBiome; import art.arcane.iris.util.nbt.common.mca.NBTWorld; @@ -1140,6 +1141,18 @@ public class NMSBinding implements INMSBinding { } } + @Override + public boolean pollChunkTask(World world) { + if (FoliaScheduler.isRegionizedRuntime(Bukkit.getServer())) { + return false; + } + ServerLevel level = ((CraftWorld) world).getHandle(); + if (Thread.currentThread() != level.getServer().getRunningThread()) { + throw new IllegalStateException("Native chunk tasks require the server lifecycle thread"); + } + return level.getChunkSource().pollTask(); + } + @Override public void flushChunkIO(World world) { try { diff --git a/adapters/bukkit/nms/v26_2_R1/src/test/java/art/arcane/iris/core/nms/v26_2_R1/NmsChunkTaskPollingTest.java b/adapters/bukkit/nms/v26_2_R1/src/test/java/art/arcane/iris/core/nms/v26_2_R1/NmsChunkTaskPollingTest.java new file mode 100644 index 000000000..1f61cfd04 --- /dev/null +++ b/adapters/bukkit/nms/v26_2_R1/src/test/java/art/arcane/iris/core/nms/v26_2_R1/NmsChunkTaskPollingTest.java @@ -0,0 +1,95 @@ +package art.arcane.iris.core.nms.v26_2_R1; + +import art.arcane.volmlib.util.scheduling.FoliaScheduler; +import net.minecraft.SharedConstants; +import net.minecraft.server.Bootstrap; +import net.minecraft.server.MinecraftServer; +import net.minecraft.server.level.ServerChunkCache; +import net.minecraft.server.level.ServerLevel; +import org.bukkit.Bukkit; +import org.bukkit.Server; +import org.bukkit.craftbukkit.CraftWorld; +import org.junit.BeforeClass; +import org.junit.Test; +import org.mockito.MockedStatic; + +import java.util.concurrent.CompletableFuture; + +import static org.junit.Assert.assertFalse; +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.mockStatic; +import static org.mockito.Mockito.never; +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 NmsChunkTaskPollingTest { + @BeforeClass + public static void bootstrapMinecraftRegistries() { + SharedConstants.tryDetectVersion(); + Bootstrap.bootStrap(); + } + + @Test + public void serverOwnerPumpsOnlyTheTargetWorldQueue() { + Server bukkitServer = mock(Server.class); + CraftWorld world = mock(CraftWorld.class); + ServerLevel level = mock(ServerLevel.class); + MinecraftServer server = mock(MinecraftServer.class); + ServerChunkCache chunks = mock(ServerChunkCache.class); + NMSBinding binding = mock(NMSBinding.class, CALLS_REAL_METHODS); + CompletableFuture completed = new CompletableFuture<>(); + when(world.getHandle()).thenReturn(level); + when(level.getServer()).thenReturn(server); + when(server.getRunningThread()).thenReturn(Thread.currentThread()); + when(level.getChunkSource()).thenReturn(chunks); + when(chunks.pollTask()).thenAnswer(invocation -> completed.complete(null)).thenReturn(false); + try (MockedStatic bukkit = mockStatic(Bukkit.class); + MockedStatic folia = mockStatic(FoliaScheduler.class)) { + bukkit.when(Bukkit::getServer).thenReturn(bukkitServer); + folia.when(() -> FoliaScheduler.isRegionizedRuntime(bukkitServer)).thenReturn(false); + assertTrue(binding.pollChunkTask(world)); + assertTrue(completed.isDone()); + assertFalse(binding.pollChunkTask(world)); + } + verify(chunks, times(2)).pollTask(); + verify(server, never()).executeAllRecentInternalTasks(); + } + + @Test + public void nonOwnerIsRejectedBeforeAccessingTheChunkQueue() { + Server bukkitServer = mock(Server.class); + CraftWorld world = mock(CraftWorld.class); + ServerLevel level = mock(ServerLevel.class); + MinecraftServer server = mock(MinecraftServer.class); + NMSBinding binding = mock(NMSBinding.class, CALLS_REAL_METHODS); + when(world.getHandle()).thenReturn(level); + when(level.getServer()).thenReturn(server); + when(server.getRunningThread()).thenReturn(new Thread("Other server owner")); + try (MockedStatic bukkit = mockStatic(Bukkit.class); + MockedStatic folia = mockStatic(FoliaScheduler.class)) { + bukkit.when(Bukkit::getServer).thenReturn(bukkitServer); + folia.when(() -> FoliaScheduler.isRegionizedRuntime(bukkitServer)).thenReturn(false); + assertThrows(IllegalStateException.class, () -> binding.pollChunkTask(world)); + } + verify(level, never()).getChunkSource(); + } + + @Test + public void foliaNeverUsesTheWorldWideTaskPump() { + Server bukkitServer = mock(Server.class); + CraftWorld world = mock(CraftWorld.class); + NMSBinding binding = mock(NMSBinding.class, CALLS_REAL_METHODS); + try (MockedStatic bukkit = mockStatic(Bukkit.class); + MockedStatic folia = mockStatic(FoliaScheduler.class)) { + bukkit.when(Bukkit::getServer).thenReturn(bukkitServer); + folia.when(() -> FoliaScheduler.isRegionizedRuntime(bukkitServer)).thenReturn(true); + assertFalse(binding.pollChunkTask(world)); + } + verifyNoInteractions(world); + } +} diff --git a/adapters/bukkit/nms/v26_2_R1/src/test/java/art/arcane/iris/nativegen/NativeStructureVolumeIndexTest.java b/adapters/bukkit/nms/v26_2_R1/src/test/java/art/arcane/iris/nativegen/NativeStructureVolumeIndexTest.java index 05021ee9b..509b5b5f5 100644 --- a/adapters/bukkit/nms/v26_2_R1/src/test/java/art/arcane/iris/nativegen/NativeStructureVolumeIndexTest.java +++ b/adapters/bukkit/nms/v26_2_R1/src/test/java/art/arcane/iris/nativegen/NativeStructureVolumeIndexTest.java @@ -1,5 +1,6 @@ package art.arcane.iris.nativegen; +import art.arcane.iris.engine.IrisEngine; import art.arcane.iris.engine.framework.Engine; import art.arcane.iris.engine.framework.NativeStructureVolume; import art.arcane.volmlib.util.collection.KList; @@ -20,18 +21,26 @@ import org.junit.Test; import java.nio.file.Files; import java.nio.file.Path; import java.lang.reflect.Proxy; +import java.lang.reflect.Field; import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.IntConsumer; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; public class NativeStructureVolumeIndexTest { private static final String STRUCTURE_KEY = "minecraft:swamp_hut"; @@ -105,6 +114,86 @@ public class NativeStructureVolumeIndexTest { assertEquals(867, resolver.resolutions()); } + @Test + public void invalidatedIndexReceivesRuntimeRetirementAndDetachesOnUninstall() throws Exception { + IrisEngine engine = mock(IrisEngine.class); + Set listeners = new CopyOnWriteArraySet<>(); + doAnswer(invocation -> { + listeners.add(invocation.getArgument(0)); + return null; + }).when(engine).addGenerationRuntimeRetirementListener(any()); + doAnswer(invocation -> { + listeners.remove(invocation.getArgument(0)); + return null; + }).when(engine).removeGenerationRuntimeRetirementListener(any()); + NativeStructureVolumeIndex.Context context = new NativeStructureVolumeIndex.Context( + null, null, null, null, () -> null, () -> null, () -> null); + NativeStructureVolumeIndex.install(engine, context); + try { + assertEquals(1, listeners.size()); + IntConsumer original = listeners.iterator().next(); + NativeStructureVolumeIndex.invalidate(engine); + assertEquals(1, listeners.size()); + assertFalse(listeners.contains(original)); + + NativeStructureVolumeIndex.volumes(engine, 0, 0, 15, 15); + NativeStructureVolumeIndex current = installedIndexes().get(engine); + assertEquals(289, cache(current, "originCache").size()); + assertEquals(1, cache(current, "queryCache").size()); + for (IntConsumer listener : listeners) { + listener.accept(0); + } + assertTrue(cache(current, "originCache").isEmpty()); + assertTrue(cache(current, "queryCache").isEmpty()); + } finally { + NativeStructureVolumeIndex.uninstall(engine); + } + assertTrue(listeners.isEmpty()); + } + + @Test + public void retiringOriginBuildCannotRepublishWhileQueryEvictionWaits() throws Exception { + BlockingResolver resolver = new BlockingResolver(); + NativeStructureVolumeIndex index = NativeStructureVolumeIndex.forTesting(resolver); + ExecutorService executor = Executors.newFixedThreadPool(2); + Map origins = cache(index, "originCache"); + Map queries = cache(index, "queryCache"); + try { + Future> builder = executor.submit(() -> index.originVolumes(null, 0, 0)); + assertTrue(resolver.awaitFirstEntry()); + index.originVolumes(null, 1, 0); + assertEquals(1, origins.size()); + Future retirement; + synchronized (queries) { + retirement = executor.submit(() -> index.evictRuntime(0)); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + boolean cleared = false; + while (System.nanoTime() < deadline) { + synchronized (origins) { + cleared = origins.isEmpty(); + } + if (cleared) { + break; + } + Thread.sleep(1); + } + assertTrue("Origin eviction did not complete before the query cache lock", cleared); + resolver.release(); + assertTrue(builder.get(5, TimeUnit.SECONDS).isEmpty()); + synchronized (origins) { + assertTrue("The retired in-flight origin was cached again", origins.isEmpty()); + } + } + retirement.get(5, TimeUnit.SECONDS); + index.originVolumes(null, 0, 0); + assertEquals(3, resolver.resolutions()); + } finally { + resolver.release(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + @Test public void independentIndexesResolveIdenticalVolumesFromTheSameSeed() { NativeStructureVolumeIndex first = NativeStructureVolumeIndex.forTesting(new CountingResolver(0, 0)); @@ -224,6 +313,19 @@ public class NativeStructureVolumeIndexTest { assertTrue(source.contains("queryCache.keySet().removeIf(key -> key.runtimeId() == runtimeId)")); } + @SuppressWarnings("unchecked") + private static Map installedIndexes() throws Exception { + Field indexes = NativeStructureVolumeIndex.class.getDeclaredField("INDEXES"); + indexes.setAccessible(true); + return (Map) indexes.get(null); + } + + private static Map cache(NativeStructureVolumeIndex index, String name) throws Exception { + Field field = NativeStructureVolumeIndex.class.getDeclaredField(name); + field.setAccessible(true); + return (Map) field.get(index); + } + private static StructureStart swampHut(int chunkX, int chunkZ, int x, int z) { Structure source = new SwampHutStructure(new Structure.StructureSettings(HolderSet.empty())); SwampHutPiece piece = new SwampHutPiece(RandomSource.create(17L), x, z); diff --git a/adapters/bukkit/plugin/src/main/java/art/arcane/iris/Iris.java b/adapters/bukkit/plugin/src/main/java/art/arcane/iris/Iris.java index d55fa2124..19568ca4a 100644 --- a/adapters/bukkit/plugin/src/main/java/art/arcane/iris/Iris.java +++ b/adapters/bukkit/plugin/src/main/java/art/arcane/iris/Iris.java @@ -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 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); } diff --git a/adapters/bukkit/plugin/src/test/java/art/arcane/iris/core/runtime/BukkitEngineLifecycleContractTest.java b/adapters/bukkit/plugin/src/test/java/art/arcane/iris/core/runtime/BukkitEngineLifecycleContractTest.java index c715cf9a0..4cef35422 100644 --- a/adapters/bukkit/plugin/src/test/java/art/arcane/iris/core/runtime/BukkitEngineLifecycleContractTest.java +++ b/adapters/bukkit/plugin/src/test/java/art/arcane/iris/core/runtime/BukkitEngineLifecycleContractTest.java @@ -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)"); diff --git a/adapters/minecraft-common/src/main/java/art/arcane/iris/nativegen/NativeStructureVolumeIndex.java b/adapters/minecraft-common/src/main/java/art/arcane/iris/nativegen/NativeStructureVolumeIndex.java index 37a71a29d..0b7b8d2de 100644 --- a/adapters/minecraft-common/src/main/java/art/arcane/iris/nativegen/NativeStructureVolumeIndex.java +++ b/adapters/minecraft-common/src/main/java/art/arcane/iris/nativegen/NativeStructureVolumeIndex.java @@ -113,7 +113,10 @@ public final class NativeStructureVolumeIndex { synchronized (INDEXES) { NativeStructureVolumeIndex current = INDEXES.get(engine); if (current != null) { - INDEXES.put(engine, current.fresh()); + NativeStructureVolumeIndex replacement = current.fresh(); + current.detachRetirementListener(engine); + INDEXES.put(engine, replacement); + replacement.attachRetirementListener(engine); } } } @@ -395,13 +398,13 @@ public final class NativeStructureVolumeIndex { void evictRuntime(int runtimeId) { synchronized (originCache) { + originBuilds.keySet().removeIf(key -> key.runtimeId() == runtimeId); originCache.keySet().removeIf(key -> key.runtimeId() == runtimeId); } synchronized (queryCache) { + queryBuilds.keySet().removeIf(key -> key.runtimeId() == runtimeId); queryCache.keySet().removeIf(key -> key.runtimeId() == runtimeId); } - originBuilds.keySet().removeIf(key -> key.runtimeId() == runtimeId); - queryBuilds.keySet().removeIf(key -> key.runtimeId() == runtimeId); } private KList cached( diff --git a/core/src/main/java/art/arcane/iris/core/gui/PregeneratorJob.java b/core/src/main/java/art/arcane/iris/core/gui/PregeneratorJob.java index e45b809cb..e226a202d 100644 --- a/core/src/main/java/art/arcane/iris/core/gui/PregeneratorJob.java +++ b/core/src/main/java/art/arcane/iris/core/gui/PregeneratorJob.java @@ -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."); diff --git a/core/src/main/java/art/arcane/iris/core/nms/INMSBinding.java b/core/src/main/java/art/arcane/iris/core/nms/INMSBinding.java index 440f61ba5..9ccf45001 100644 --- a/core/src/main/java/art/arcane/iris/core/nms/INMSBinding.java +++ b/core/src/main/java/art/arcane/iris/core/nms/INMSBinding.java @@ -242,6 +242,10 @@ public interface INMSBinding { return false; } + default boolean pollChunkTask(World world) { + return false; + } + default void flushChunkIO(World world) { } diff --git a/core/src/main/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenMethod.java b/core/src/main/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenMethod.java index 9800983b7..399eee3d6 100644 --- a/core/src/main/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenMethod.java +++ b/core/src/main/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenMethod.java @@ -357,6 +357,9 @@ public class AsyncPregenMethod implements PregeneratorMethod { } private CompletableFuture 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 keys = new ArrayList<>(regionChunks.keySet()); List> evictions = new ArrayList<>(keys.size() + pendingEvictions.size()); for (Long regionKey : keys) { @@ -453,10 +459,27 @@ public class AsyncPregenMethod implements PregeneratorMethod { CompletableFuture settledEvictions = evictionsComplete.handle((ignored, failure) -> null); CompletableFuture 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() + "."); diff --git a/core/src/main/java/art/arcane/iris/engine/history/NativeTerrainReceipt.java b/core/src/main/java/art/arcane/iris/engine/history/NativeTerrainReceipt.java index fc518e4c3..943b87299 100644 --- a/core/src/main/java/art/arcane/iris/engine/history/NativeTerrainReceipt.java +++ b/core/src/main/java/art/arcane/iris/engine/history/NativeTerrainReceipt.java @@ -1,5 +1,6 @@ package art.arcane.iris.engine.history; +import java.io.BufferedOutputStream; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.DataInputStream; @@ -23,7 +24,7 @@ public final class NativeTerrainReceipt { public static byte[] encode(SavedTerrainChunk chunk, long activationId, String epochId) throws IOException { ByteArrayOutputStream bytes = new ByteArrayOutputStream(); - try (DataOutputStream output = new DataOutputStream(new GZIPOutputStream(bytes))) { + try (DataOutputStream output = new DataOutputStream(new BufferedOutputStream(new GZIPOutputStream(bytes)))) { output.writeInt(MAGIC); output.writeInt(VERSION); output.writeInt(chunk.chunkX()); diff --git a/core/src/main/java/art/arcane/iris/engine/history/SavedBiomeStore.java b/core/src/main/java/art/arcane/iris/engine/history/SavedBiomeStore.java index ee8776fbb..4abcd342d 100644 --- a/core/src/main/java/art/arcane/iris/engine/history/SavedBiomeStore.java +++ b/core/src/main/java/art/arcane/iris/engine/history/SavedBiomeStore.java @@ -99,28 +99,30 @@ public final class SavedBiomeStore { public boolean claimAndPersist(SavedBiomeChunk chunk) throws IOException { SavedBiomeChunk required = Objects.requireNonNull(chunk, "chunk"); - synchronized (regionLock(required.chunkX(), required.chunkZ())) { - Optional existing = get(required.chunkX(), required.chunkZ()); - if (existing.isPresent()) { - if (!existing.get().equals(required)) { - throw new IOException("Conflicting saved biome claim at chunk " + required.chunkX() + ", " + required.chunkZ()); - } + Object lock = regionLock(required.chunkX(), required.chunkZ()); + synchronized (lock) { + if (alreadyClaimed(required)) { + return false; + } + } + byte[] body = encode(required); + byte[] record = ByteBuffer.allocate(body.length + 8) + .putInt(body.length).put(body).putInt(checksum(body)).array(); + synchronized (lock) { + if (alreadyClaimed(required)) { return false; } - byte[] body = encode(required); RegionIndex region = region(required.chunkX(), required.chunkZ()); ensureRegionFile(region); long offset; try (RandomAccessFile output = new RandomAccessFile(region.path.toFile(), "rw")) { offset = output.length(); - if (offset + body.length + 8L > MAXIMUM_REGION_BYTES) { + if (offset + record.length > MAXIMUM_REGION_BYTES) { throw new IOException("Saved biome region exceeds its storage limit: " + region.path); } try { output.seek(offset); - output.writeInt(body.length); - output.write(body); - output.writeInt(checksum(body)); + output.write(record); output.getChannel().force(true); } catch (IOException failure) { try { @@ -202,6 +204,17 @@ public final class SavedBiomeStore { } } + private boolean alreadyClaimed(SavedBiomeChunk chunk) throws IOException { + Optional existing = get(chunk.chunkX(), chunk.chunkZ()); + if (existing.isEmpty()) { + return false; + } + if (!existing.get().equals(chunk)) { + throw new IOException("Conflicting saved biome claim at chunk " + chunk.chunkX() + ", " + chunk.chunkZ()); + } + return true; + } + private RegionIndex region(int chunkX, int chunkZ) throws IOException { int regionX = chunkX >> 5; int regionZ = chunkZ >> 5; diff --git a/core/src/main/java/art/arcane/iris/util/common/scheduling/J.java b/core/src/main/java/art/arcane/iris/util/common/scheduling/J.java index dd2cee405..cccef549a 100644 --- a/core/src/main/java/art/arcane/iris/util/common/scheduling/J.java +++ b/core/src/main/java/art/arcane/iris/util/common/scheduling/J.java @@ -772,7 +772,7 @@ public class J { } private static boolean isPluginEnabled() { - return usesBukkitScheduler() && BukkitPlatform.hasPlugin() && Bukkit.getPluginManager().isPluginEnabled(BukkitPlatform.plugin()); + return usesBukkitScheduler() && BukkitPlatform.hasPlugin() && BukkitPlatform.plugin().isEnabled(); } private static boolean canSchedule() { diff --git a/core/src/test/java/art/arcane/iris/core/gui/PregeneratorJobOwnerDrainTest.java b/core/src/test/java/art/arcane/iris/core/gui/PregeneratorJobOwnerDrainTest.java new file mode 100644 index 000000000..68e858f64 --- /dev/null +++ b/core/src/test/java/art/arcane/iris/core/gui/PregeneratorJobOwnerDrainTest.java @@ -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 completion = new CompletableFuture<>(); + private final AtomicReference 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 instance() throws Exception { + Field field = PregeneratorJob.class.getDeclaredField("instance"); + field.setAccessible(true); + return (AtomicReference) field.get(null); + } + } +} diff --git a/core/src/test/java/art/arcane/iris/core/gui/PregeneratorJobShutdownDispatchTest.java b/core/src/test/java/art/arcane/iris/core/gui/PregeneratorJobShutdownDispatchTest.java new file mode 100644 index 000000000..9db13b6f1 --- /dev/null +++ b/core/src/test/java/art/arcane/iris/core/gui/PregeneratorJobShutdownDispatchTest.java @@ -0,0 +1,223 @@ +package art.arcane.iris.core.gui; + +import art.arcane.iris.core.pregenerator.IrisPregenerator; +import art.arcane.iris.core.pregenerator.PregenApiPhase; +import art.arcane.iris.core.pregenerator.PregenApiSink; +import art.arcane.iris.core.pregenerator.PregenPhaseTracker; +import art.arcane.iris.platform.bukkit.BukkitPlatform; +import art.arcane.iris.spi.IrisPlatform; +import art.arcane.iris.spi.IrisPlatforms; +import art.arcane.iris.spi.IrisServices; +import art.arcane.iris.util.common.scheduling.J; +import art.arcane.volmlib.util.format.MemoryMonitor; +import art.arcane.volmlib.util.scheduling.FoliaScheduler; +import org.bukkit.Bukkit; +import org.bukkit.Server; +import org.bukkit.plugin.IllegalPluginAccessException; +import org.bukkit.plugin.Plugin; +import org.bukkit.plugin.PluginManager; +import org.bukkit.scheduler.BukkitScheduler; +import org.junit.Test; +import org.mockito.MockedStatic; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +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.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.contains; +import static org.mockito.ArgumentMatchers.eq; +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.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class PregeneratorJobShutdownDispatchTest { + @Test + public void pluginDisableCanJoinAWorkerAlreadyDispatchingSaving() throws Exception { + try (Fixture fixture = new Fixture()) { + CountDownLatch stop = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + Thread worker = new Thread(() -> { + try { + fixture.withBukkit(() -> { + fixture.job.onSaving(); + try { + stop.await(); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } + fixture.job.onClose(); + }); + } catch (Throwable error) { + failure.set(error); + } + }); + fixture.set("worker", worker); + doAnswer(invocation -> { + stop.countDown(); + return null; + }).when(fixture.pregenerator).close(); + AtomicReference instance = instance(); + PregeneratorJob previous = instance.getAndSet(fixture.job); + try { + synchronized (fixture.manager) { + worker.start(); + assertTrue(fixture.dispatchEntered.await(5, TimeUnit.SECONDS)); + fixture.enabled.set(false); + assertTrue(PregeneratorJob.shutdownAndWait(500)); + } + assertFalse(worker.isAlive()); + assertNull(failure.get()); + verify(fixture.pregenerator).close(); + verify(fixture.monitor).close(); + verify(fixture.service).shutdownNow(); + verify(fixture.manager, never()).isPluginEnabled(fixture.plugin); + } finally { + stop.countDown(); + worker.interrupt(); + worker.join(5_000); + assertFalse(worker.isAlive()); + instance.set(previous); + } + } + } + + @Test + public void enabledJobsKeepScheduledSavingAndTerminalEvents() throws Exception { + try (Fixture fixture = new Fixture()) { + fixture.withBukkit(() -> { + fixture.job.onSaving(); + fixture.job.onClose(); + }); + assertTrue(fixture.delivered.isEmpty()); + assertEquals(2, fixture.queued.size()); + fixture.queued.forEach(Runnable::run); + assertEquals(List.of(PregenApiPhase.SAVING, PregenApiPhase.CANCELLED), fixture.delivered); + verify(fixture.monitor).close(); + verify(fixture.service).shutdownNow(); + } + } + + @Test + public void disableAfterEnabledCheckReportsRejectionAndStillCleansUp() throws Exception { + try (Fixture fixture = new Fixture()) { + IllegalPluginAccessException rejected = new IllegalPluginAccessException("Plugin disabled during scheduling"); + when(fixture.scheduler.scheduleSyncDelayedTask(eq(fixture.plugin), any(Runnable.class))) + .thenAnswer(invocation -> { + fixture.enabled.set(false); + throw rejected; + }); + fixture.withBukkit(() -> { + fixture.job.onSaving(); + fixture.job.onClose(); + }); + verify(fixture.platform).reportError(contains("SAVING"), eq(rejected)); + verify(fixture.monitor).close(); + verify(fixture.service).shutdownNow(); + assertTrue(fixture.queued.isEmpty()); + } + } + + @Test + public void unexpectedSchedulingFailureRetainsItsFullCause() throws Exception { + try (Fixture fixture = new Fixture()) { + IllegalStateException failure = new IllegalStateException("Scheduler unavailable"); + when(fixture.scheduler.scheduleSyncDelayedTask(eq(fixture.plugin), any(Runnable.class))).thenThrow(failure); + fixture.withBukkit(fixture.job::onSaving); + verify(fixture.platform).reportError(contains("SAVING"), eq(failure)); + } + } + + @SuppressWarnings("unchecked") + private static AtomicReference instance() throws Exception { + Field field = PregeneratorJob.class.getDeclaredField("instance"); + field.setAccessible(true); + return (AtomicReference) field.get(null); + } + + 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 MemoryMonitor monitor = mock(MemoryMonitor.class); + private final ExecutorService service = mock(ExecutorService.class); + private final IrisPlatform platform = mock(IrisPlatform.class); + private final Plugin plugin = mock(Plugin.class); + private final PluginManager manager = mock(PluginManager.class); + private final BukkitScheduler scheduler = mock(BukkitScheduler.class); + private final Server server = mock(Server.class); + private final AtomicBoolean enabled = new AtomicBoolean(true); + private final CountDownLatch dispatchEntered = new CountDownLatch(1); + private final List queued = new ArrayList<>(); + private final List delivered = new ArrayList<>(); + private final PregenApiSink previousSink = IrisServices.getOrNull(PregenApiSink.class); + + private Fixture() throws Exception { + when(platform.platformName()).thenReturn("bukkit"); + when(plugin.isEnabled()).thenAnswer(invocation -> enabled.get()); + when(manager.isPluginEnabled(plugin)).thenAnswer(invocation -> { + synchronized (manager) { + return plugin.isEnabled(); + } + }); + when(scheduler.scheduleSyncDelayedTask(eq(plugin), any(Runnable.class))).thenAnswer(invocation -> { + queued.add(invocation.getArgument(1)); + return queued.size(); + }); + IrisPlatforms.bind(platform); + BukkitPlatform.hostPlugin(plugin); + PregenApiSink sink = (phase, progress) -> { + dispatchEntered.countDown(); + J.s(() -> delivered.add(phase)); + }; + IrisServices.register(PregenApiSink.class, sink); + set("bounds", new PregenRenderSnapshot.Bounds(0, 0, 0, 0)); + set("pregenerator", pregenerator); + set("monitor", monitor); + set("service", service); + set("saving", new AtomicBoolean()); + set("stopRequested", new AtomicBoolean()); + set("apiPhases", new PregenPhaseTracker()); + set("whenDone", List.of()); + } + + @Override + public void close() { + if (previousSink == null) { + IrisServices.remove(PregenApiSink.class); + } else { + IrisServices.register(PregenApiSink.class, previousSink); + } + BukkitPlatform.hostPlugin(null); + IrisPlatforms.unbind(); + } + + private void withBukkit(Runnable action) { + try (MockedStatic bukkit = mockStatic(Bukkit.class); + MockedStatic folia = mockStatic(FoliaScheduler.class)) { + bukkit.when(Bukkit::getPluginManager).thenReturn(manager); + bukkit.when(Bukkit::getScheduler).thenReturn(scheduler); + bukkit.when(Bukkit::getServer).thenReturn(server); + action.run(); + } + } + + private void set(String name, Object value) throws Exception { + Field field = PregeneratorJob.class.getDeclaredField(name); + field.setAccessible(true); + field.set(job, value); + } + } +} diff --git a/core/src/test/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenServerSaveOwnershipTest.java b/core/src/test/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenServerSaveOwnershipTest.java new file mode 100644 index 000000000..1e7dba2a1 --- /dev/null +++ b/core/src/test/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenServerSaveOwnershipTest.java @@ -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 toolbelt = mockStatic(IrisToolbelt.class); + MockedStatic 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 toolbelt = mockStatic(IrisToolbelt.class); + MockedStatic 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> chunks = new ConcurrentHashMap<>(); + private final CompletableFuture pending = new CompletableFuture<>(); + + private Fixture() throws Exception { + Queue region = new ConcurrentLinkedQueue<>(); + region.add(chunk); + chunks.put(0L, region); + Queue> 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 toolbelt = mockStatic(IrisToolbelt.class); + private final MockedStatic scheduling = mockStatic(J.class); + private final MockedStatic nativeAccess = mockStatic(INMS.class); + private final MockedStatic 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(); + } + } +} diff --git a/core/src/test/java/art/arcane/iris/engine/history/NativeTerrainReceiptTest.java b/core/src/test/java/art/arcane/iris/engine/history/NativeTerrainReceiptTest.java new file mode 100644 index 000000000..2794cfdc0 --- /dev/null +++ b/core/src/test/java/art/arcane/iris/engine/history/NativeTerrainReceiptTest.java @@ -0,0 +1,84 @@ +package art.arcane.iris.engine.history; + +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.HexFormat; +import java.util.List; +import java.util.OptionalInt; +import java.util.zip.GZIPInputStream; + +import static org.junit.Assert.assertEquals; + +public class NativeTerrainReceiptTest { + @Test + public void preservesCompleteAndBoundaryReceiptBytes() throws Exception { + for (boolean boundary : new boolean[]{false, true}) { + SavedTerrainChunk chunk = chunk(boundary); + String epoch = "epoch-\u0000-é-水-\ud83c\udf32"; + byte[] encoded = NativeTerrainReceipt.encode(chunk, 7, epoch); + byte[] payload; + try (GZIPInputStream input = new GZIPInputStream(new ByteArrayInputStream(encoded))) { + payload = input.readAllBytes(); + } + assertEquals(boundary + ? "42317aaeb436fa2ab174eaae05777dc46f4a3d9253fd9374680ab49fb4182225" + : "95879743b4b720c20598560159cffaed4dc4d3926369cac381ba8a0dad14e7b8", + HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(payload))); + NativeTerrainReceipt.Decoded decoded = NativeTerrainReceipt.decode(encoded, "minecraft:noise"); + assertEquals(7, decoded.activationId()); + assertEquals(epoch, decoded.epochId()); + assertEquals(-2, decoded.terrain().chunkX()); + assertEquals(3, decoded.terrain().chunkZ()); + for (int x = 0; x < 16; x++) { + for (int z = 0; z < 16; z++) { + assertEquals(chunk.hasColumn(x, z), decoded.terrain().hasColumn(x, z)); + if (!chunk.hasColumn(x, z)) { + continue; + } + TerrainBoundarySignature expected = chunk.column(-32 + x, 48 + z); + TerrainBoundarySignature actual = decoded.terrain().column(-32 + x, 48 + z); + assertEquals(expected.column(), actual.column()); + assertEquals(expected.geometry(), actual.geometry()); + assertEquals(expected.samples().layout(), actual.samples().layout()); + for (int index = 0; index < expected.sampleCount(); index++) { + assertEquals(expected.biomeAtSample(index), actual.biomeAtSample(index)); + } + } + } + } + } + + private static SavedTerrainChunk chunk(boolean boundary) { + List columns = new ArrayList<>(256); + List palette = List.of( + new BoundaryColumnGeometry.Voxel("minecraft:stone", BoundaryColumnGeometry.Phase.SOLID, "", false), + new BoundaryColumnGeometry.Voxel("minecraft:water[level=0]", BoundaryColumnGeometry.Phase.FLUID, + "minecraft:water[level=0]", false), + new BoundaryColumnGeometry.Voxel("minecraft:air", BoundaryColumnGeometry.Phase.AIR, "", false), + new BoundaryColumnGeometry.Voxel("minecraft:oak_log[axis=y]", BoundaryColumnGeometry.Phase.SOLID, "", true)); + for (int x = 0; x < 16; x++) { + for (int z = 0; z < 16; z++) { + if (boundary && x != 0 && x != 15 && z != 0 && z != 15) { + columns.add(null); + continue; + } + short[] biomes = new short[96]; + for (int index = 0; index < biomes.length; index++) { + biomes[index] = (short) ((index / 8 + x + z) % 3); + } + BoundaryColumnGeometry geometry = new BoundaryColumnGeometry(-64, palette, + new int[]{128 + z, 176 + x, 260 + z, 384}, new short[]{0, 1, 3, 2}); + columns.add(new TerrainBoundarySignature(new TerrainBoundarySignature.Column(-32 + x, 48 + z, + 259 + z, 259 + z, OptionalInt.empty(), OptionalInt.of(4)), + new TerrainBoundarySignature.Samples(new TerrainBoundarySignature.VerticalLayout(-64, 4, 96), + new TerrainBoundarySignature.BiomeEncoding( + List.of("minecraft:plains", "iris:水-é-\u0000", "iris:cave-\ud83c\udf32"), biomes)), + geometry)); + } + } + return new SavedTerrainChunk(-2, 3, "minecraft:noise", columns); + } +} diff --git a/core/src/test/java/art/arcane/iris/engine/history/SavedBiomeStoreWriteTest.java b/core/src/test/java/art/arcane/iris/engine/history/SavedBiomeStoreWriteTest.java new file mode 100644 index 000000000..af00fbd76 --- /dev/null +++ b/core/src/test/java/art/arcane/iris/engine/history/SavedBiomeStoreWriteTest.java @@ -0,0 +1,326 @@ +package art.arcane.iris.engine.history; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.MockedConstruction; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Set; +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.atomic.AtomicBoolean; +import java.util.zip.CRC32; +import java.util.zip.Deflater; +import java.util.zip.DeflaterOutputStream; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +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.AdditionalAnswers.delegatesTo; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.withSettings; + +public class SavedBiomeStoreWriteTest { + @Rule + public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void recordMatchesTheCurrentLengthBodyChecksumFormat() throws Exception { + Path root = temporaryFolder.newFolder().toPath(); + SavedBiomeStore store = SavedBiomeStore.open(root); + store.claimAndPersist(chunk(0, 3L)); + assertArrayEquals(expectedRegion(), Files.readAllBytes(regionPath(root))); + } + + @Test + public void encodingAllowsOtherReadsInTheSameRegion() throws Exception { + SavedBiomeStore store = SavedBiomeStore.open(temporaryFolder.newFolder().toPath()); + CountDownLatch encoding = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + SavedBiomeChunk chunk = pausedChunk(0, 3L, encoding, release); + ExecutorService workers = Executors.newFixedThreadPool(2); + try { + Future writing = workers.submit(() -> store.claimAndPersist(chunk)); + assertTrue(encoding.await(5, TimeUnit.SECONDS)); + Future reading = workers.submit(() -> { + assertTrue(store.get(1, 0).isEmpty()); + return null; + }); + reading.get(1, TimeUnit.SECONDS); + assertFalse(writing.isDone()); + release.countDown(); + assertTrue(writing.get(5, TimeUnit.SECONDS)); + } finally { + release.countDown(); + drain(workers); + } + } + + @Test + public void appendWritesOneFrameAndPublishesOnlyAfterForce() throws Exception { + Path root = temporaryFolder.newFolder().toPath(); + SavedBiomeStore store = SavedBiomeStore.open(root); + store.claimAndPersist(chunk(0, 3L)); + CountDownLatch forcing = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + ExecutorService workers = Executors.newSingleThreadExecutor(); + try { + Future writing = workers.submit(() -> { + try (RandomAccessFile actual = new RandomAccessFile(regionPath(root).toFile(), "rw")) { + FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel())); + doAnswer(invocation -> { + forcing.countDown(); + await(release); + actual.getChannel().force(true); + return null; + }).when(channel).force(true); + try (MockedConstruction files = mockConstruction(RandomAccessFile.class, + withSettings().defaultAnswer(delegatesTo(actual)), + (file, context) -> doReturn(channel).when(file).getChannel())) { + boolean result = store.claimAndPersist(chunk(1, 3L)); + assertEquals(1, files.constructed().size()); + verify(files.constructed().getFirst()).write(any(byte[].class)); + verify(files.constructed().getFirst(), never()).writeInt(anyInt()); + return result; + } + } + }); + assertTrue(forcing.await(5, TimeUnit.SECONDS)); + assertTrue(store.cached(1, 0).isEmpty()); + assertFalse(writing.isDone()); + release.countDown(); + assertTrue(writing.get(5, TimeUnit.SECONDS)); + assertEquals(chunk(1, 3L), store.cached(1, 0).orElseThrow()); + assertEquals(chunk(1, 3L), SavedBiomeStore.open(root).get(1, 0).orElseThrow()); + } finally { + release.countDown(); + drain(workers); + } + } + + @Test + public void failedForceRollsBackWithoutPublishingAndRetainsRollbackFailure() throws Exception { + for (boolean failRollback : List.of(false, true)) { + Path root = temporaryFolder.newFolder().toPath(); + SavedBiomeStore store = SavedBiomeStore.open(root); + store.claimAndPersist(chunk(0, 3L)); + byte[] original = Files.readAllBytes(regionPath(root)); + IOException failure = new IOException("Saved biome force failed"); + IOException rollback = new IOException("Saved biome rollback force failed"); + AtomicBoolean first = new AtomicBoolean(true); + try (RandomAccessFile actual = new RandomAccessFile(regionPath(root).toFile(), "rw")) { + FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel())); + doAnswer(invocation -> { + if (first.getAndSet(false)) { + throw failure; + } + if (failRollback) { + throw rollback; + } + actual.getChannel().force(true); + return null; + }).when(channel).force(true); + try (MockedConstruction ignored = mockConstruction(RandomAccessFile.class, + withSettings().defaultAnswer(delegatesTo(actual)), + (file, context) -> doReturn(channel).when(file).getChannel())) { + assertSame(failure, assertThrows(IOException.class, () -> store.claimAndPersist(chunk(1, 3L)))); + } + } + assertEquals(failRollback ? 1 : 0, failure.getSuppressed().length); + if (failRollback) { + assertSame(rollback, failure.getSuppressed()[0]); + } + assertArrayEquals(original, Files.readAllBytes(regionPath(root))); + assertTrue(store.get(1, 0).isEmpty()); + assertTrue(SavedBiomeStore.open(root).get(1, 0).isEmpty()); + assertTrue(store.claimAndPersist(chunk(1, 3L))); + } + } + + @Test + public void claimsRecheckDuplicatesAndConflictsAfterConcurrentEncoding() throws Exception { + for (long secondActivation : List.of(3L, 4L)) { + Path root = temporaryFolder.newFolder().toPath(); + SavedBiomeStore store = SavedBiomeStore.open(root); + CountDownLatch encoding = new CountDownLatch(2); + CountDownLatch release = new CountDownLatch(1); + SavedBiomeChunk first = pausedChunk(0, 3L, encoding, release); + SavedBiomeChunk second = secondActivation == 3L ? first : pausedChunk(0, secondActivation, encoding, release); + ExecutorService workers = Executors.newFixedThreadPool(2); + try { + Future left = workers.submit(() -> claimResult(store, first)); + Future right = workers.submit(() -> claimResult(store, second)); + assertTrue(encoding.await(5, TimeUnit.SECONDS)); + release.countDown(); + ClaimResult leftResult = left.get(5, TimeUnit.SECONDS); + ClaimResult rightResult = right.get(5, TimeUnit.SECONDS); + assertTrue(leftResult.added() ^ rightResult.added()); + ClaimResult loser = leftResult.added() ? rightResult : leftResult; + if (secondActivation == 3L) { + assertNull(loser.failure()); + } else { + assertNotNull(loser.failure()); + } + long winner = leftResult.added() ? 3L : secondActivation; + assertEquals(chunk(0, winner), SavedBiomeStore.open(root).get(0, 0).orElseThrow()); + } finally { + release.countDown(); + drain(workers); + } + } + } + + @Test + public void appendReloadsRegionOffsetsAfterPruningDuringEncoding() throws Exception { + Path root = temporaryFolder.newFolder().toPath(); + SavedBiomeStore store = SavedBiomeStore.open(root); + store.claimAndPersist(chunk(0, 3L)); + store.claimAndPersist(chunk(1, 3L)); + CountDownLatch encoding = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + SavedBiomeChunk pending = pausedChunk(2, 3L, encoding, release); + ExecutorService workers = Executors.newFixedThreadPool(2); + try { + Future writing = workers.submit(() -> store.claimAndPersist(pending)); + assertTrue(encoding.await(5, TimeUnit.SECONDS)); + Future pruned = workers.submit(() -> store.discardUnstoredClaims( + WorldChunkInventory.ofPackedChunks(ChunkGenerationOwnership.packChunk(1, 0)), Set.of(3L))); + assertEquals(1, (int) pruned.get(1, TimeUnit.SECONDS)); + release.countDown(); + assertTrue(writing.get(5, TimeUnit.SECONDS)); + assertTrue(store.get(0, 0).isEmpty()); + assertEquals(chunk(1, 3L), store.get(1, 0).orElseThrow()); + SavedBiomeStore reopened = SavedBiomeStore.open(root); + assertTrue(reopened.get(0, 0).isEmpty()); + assertEquals(chunk(1, 3L), reopened.get(1, 0).orElseThrow()); + assertEquals(chunk(2, 3L), reopened.get(2, 0).orElseThrow()); + } finally { + release.countDown(); + drain(workers); + } + } + + private static ClaimResult claimResult(SavedBiomeStore store, SavedBiomeChunk chunk) { + try { + return new ClaimResult(store.claimAndPersist(chunk), null); + } catch (IOException failure) { + return new ClaimResult(false, failure); + } + } + + private static SavedBiomeChunk pausedChunk(int x, long activation, CountDownLatch encoding, CountDownLatch release) { + SavedBiomeChunk chunk = spy(chunk(x, activation)); + doAnswer(invocation -> { + if (release.getCount() > 0) { + encoding.countDown(); + await(release); + } + return invocation.callRealMethod(); + }).when(chunk).column(0, 0); + return chunk; + } + + private static void await(CountDownLatch latch) throws IOException { + try { + if (!latch.await(10, TimeUnit.SECONDS)) { + throw new IOException("Timed out waiting for saved biome write"); + } + } catch (InterruptedException failure) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted waiting for saved biome write", failure); + } + } + + private static void drain(ExecutorService workers) throws InterruptedException { + workers.shutdown(); + assertTrue(workers.awaitTermination(10, TimeUnit.SECONDS)); + } + + private static SavedBiomeChunk chunk(int x, long activation) { + SavedBiomeChunk.Cell cell = new SavedBiomeChunk.Cell(activation, "biome/main", "region/main"); + SavedBiomeChunk.Column column = new SavedBiomeChunk.Column(cell, cell, + List.of(new SavedBiomeChunk.Span(-64, 320, cell))); + SavedBiomeChunk.Builder builder = SavedBiomeChunk.builder(new SavedBiomeChunk.Header(x, 0, activation, -64, 384)); + for (int z = 0; z < 16; z++) { + for (int localX = 0; localX < 16; localX++) { + builder.column(localX, z, column); + } + } + return builder.build(); + } + + private static Path regionPath(Path root) { + return root.resolve("iris/generation/biomes/r.0.0.ibio"); + } + + private static byte[] expectedRegion() throws IOException { + ByteArrayOutputStream raw = new ByteArrayOutputStream(); + DataOutputStream data = new DataOutputStream(raw); + data.writeLong(3L); + data.writeInt(-64); + data.writeInt(384); + data.writeInt(1); + data.writeLong(3L); + data.writeByte(0); + data.writeUTF("biome/main"); + data.writeUTF("region/main"); + data.writeInt(1); + data.write(new byte[]{0, 0, 1, (byte) 0x80, 3, 0}); + data.write(new byte[256]); + ByteArrayOutputStream body = new ByteArrayOutputStream(); + DataOutputStream coordinates = new DataOutputStream(body); + coordinates.writeInt(0); + coordinates.writeInt(0); + coordinates.writeInt(raw.size()); + Deflater deflater = new Deflater(Deflater.BEST_SPEED); + try (DeflaterOutputStream compressed = new DeflaterOutputStream(coordinates, deflater)) { + raw.writeTo(compressed); + } finally { + deflater.end(); + } + ByteArrayOutputStream region = new ByteArrayOutputStream(); + DataOutputStream output = new DataOutputStream(region); + output.writeInt(0x4942494F); + output.writeInt(1); + output.writeInt(0); + output.writeInt(0); + output.writeInt(checksum(region.toByteArray())); + output.writeInt(body.size()); + output.write(body.toByteArray()); + output.writeInt(checksum(body.toByteArray())); + return region.toByteArray(); + } + + private static int checksum(byte[] bytes) { + CRC32 checksum = new CRC32(); + checksum.update(bytes); + return (int) checksum.getValue(); + } + + private record ClaimResult(boolean added, IOException failure) { + } +}