diff --git a/core/src/main/java/art/arcane/iris/core/pregenerator/IrisPregenerator.java b/core/src/main/java/art/arcane/iris/core/pregenerator/IrisPregenerator.java index a10aa3963..185ef7a95 100644 --- a/core/src/main/java/art/arcane/iris/core/pregenerator/IrisPregenerator.java +++ b/core/src/main/java/art/arcane/iris/core/pregenerator/IrisPregenerator.java @@ -35,6 +35,9 @@ import art.arcane.iris.util.common.scheduling.J; import art.arcane.volmlib.util.scheduling.Looper; import art.arcane.volmlib.util.scheduling.PrecisionStopwatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @@ -43,6 +46,8 @@ import java.util.concurrent.atomic.AtomicReference; public class IrisPregenerator { private static final double INVALID = 9223372036854775807d; private static final AtomicLong JOB_SEQUENCE = new AtomicLong(0L); + private static final long GATE_WAIT_MS = 50L; + private static final long RECLAIM_WAIT_MS = 150L; private final long jobId; private final PregenTask task; private final PregeneratorMethod generator; @@ -50,6 +55,8 @@ public class IrisPregenerator { private final Looper ticker; private final AtomicBoolean paused; private final AtomicBoolean shutdown; + private final ReentrantLock stateLock = new ReentrantLock(); + private final Condition stateChanged = stateLock.newCondition(); private final PregenRateTracker rateTracker; private final KList chunksPerSecondHistory; private final AtomicLong generated; @@ -193,6 +200,7 @@ public class IrisPregenerator { public void close() { shutdown.set(true); + signalState(); } public void start() { @@ -295,6 +303,7 @@ public class IrisPregenerator { reclaimTectonicPlates(mantle); } }); + shutdownStep("parallelism", PregenPerformanceProfile::restore); shutdownStep("protocol", () -> { IrisProtocolServer protocolServer = IrisServices.getOrNull(IrisProtocolServer.class); if (protocolServer != null) { @@ -337,7 +346,7 @@ public class IrisPregenerator { } previousLoaded = loaded; - J.sleep(150); + awaitStateChange(RECLAIM_WAIT_MS); } int loadedAfter = mantle.getLoadedRegionCount(); @@ -348,7 +357,7 @@ public class IrisPregenerator { private void visitRegion(int x, int z, boolean regions) { while (paused.get() && !shutdown.get()) { - J.sleep(50); + awaitStateChange(GATE_WAIT_MS); } if (shutdown.get()) { @@ -376,7 +385,7 @@ public class IrisPregenerator { if (!paused.get()) { reclaimHeapPressure(); } - J.sleep(50); + awaitStateChange(GATE_WAIT_MS); } if (shutdown.get()) { @@ -439,10 +448,32 @@ public class IrisPregenerator { public void pause() { paused.set(true); + signalState(); } public void resume() { paused.set(false); + signalState(); + } + + private void signalState() { + stateLock.lock(); + try { + stateChanged.signalAll(); + } finally { + stateLock.unlock(); + } + } + + private void awaitStateChange(long millis) { + stateLock.lock(); + try { + stateChanged.await(millis, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + stateLock.unlock(); + } } private PregenListener listenify(PregenListener listener) { diff --git a/core/src/main/java/art/arcane/iris/core/pregenerator/PregenAdmissionGate.java b/core/src/main/java/art/arcane/iris/core/pregenerator/PregenAdmissionGate.java new file mode 100644 index 000000000..60b9566cd --- /dev/null +++ b/core/src/main/java/art/arcane/iris/core/pregenerator/PregenAdmissionGate.java @@ -0,0 +1,136 @@ +/* + * Iris is a World Generator for Minecraft Bukkit Servers + * Copyright (c) 2022 Arcane Arts (Volmit Software) + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package art.arcane.iris.core.pregenerator; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.BooleanSupplier; +import java.util.function.LongSupplier; + +public final class PregenAdmissionGate { + public record Wait(long adaptiveMs, long permitMs) { + } + + private final int permits; + private final long boundMs; + private final LongSupplier clock; + private final ReentrantLock lock = new ReentrantLock(true); + private final Condition changed = lock.newCondition(); + private int free; + private volatile int freeSnapshot; + + public PregenAdmissionGate(int permits, long boundMs, LongSupplier clock) { + if (permits < 1) { + throw new IllegalArgumentException("A pregen admission gate needs at least one permit"); + } + if (boundMs < 1L) { + throw new IllegalArgumentException("A pregen admission gate needs a positive wait bound"); + } + this.permits = permits; + this.boundMs = boundMs; + this.clock = clock; + this.free = permits; + this.freeSnapshot = permits; + } + + public int permits() { + return permits; + } + + public int availablePermits() { + return freeSnapshot; + } + + public Wait admit(BooleanSupplier overAdaptiveLimit, BooleanSupplier cancelled) throws InterruptedException { + long adaptiveMs = 0L; + long permitMs = 0L; + lock.lock(); + try { + while (true) { + if (cancelled.getAsBoolean()) { + return null; + } + + boolean blockedByLimit = overAdaptiveLimit.getAsBoolean(); + if (!blockedByLimit && free > 0) { + free--; + freeSnapshot = free; + return new Wait(adaptiveMs, permitMs); + } + + long start = clock.getAsLong(); + changed.await(boundMs, TimeUnit.MILLISECONDS); + long waited = Math.max(0L, clock.getAsLong() - start); + if (blockedByLimit) { + adaptiveMs += waited; + } else { + permitMs += waited; + } + } + } finally { + lock.unlock(); + } + } + + public void release() { + lock.lock(); + try { + if (free < permits) { + free++; + freeSnapshot = free; + } + changed.signalAll(); + } finally { + lock.unlock(); + } + } + + public void wake() { + lock.lock(); + try { + changed.signalAll(); + } finally { + lock.unlock(); + } + } + + public boolean awaitDrain(long warningInterval, TimeUnit timeUnit, Runnable onWait) { + boolean interrupted = false; + lock.lock(); + try { + while (free < permits) { + boolean signalled = false; + try { + signalled = changed.await(warningInterval, timeUnit); + } catch (InterruptedException e) { + interrupted = true; + continue; + } + if (!signalled && free < permits) { + onWait.run(); + } + } + } finally { + lock.unlock(); + } + + return interrupted; + } +} diff --git a/core/src/main/java/art/arcane/iris/core/pregenerator/PregenMantleBackpressure.java b/core/src/main/java/art/arcane/iris/core/pregenerator/PregenMantleBackpressure.java index 2df82795d..ca40d3851 100644 --- a/core/src/main/java/art/arcane/iris/core/pregenerator/PregenMantleBackpressure.java +++ b/core/src/main/java/art/arcane/iris/core/pregenerator/PregenMantleBackpressure.java @@ -22,6 +22,9 @@ import art.arcane.iris.spi.IrisLogging; import art.arcane.volmlib.util.mantle.runtime.Mantle; import art.arcane.volmlib.util.math.M; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.BooleanSupplier; import java.util.function.Supplier; @@ -33,6 +36,8 @@ public final class PregenMantleBackpressure { private final Runnable onBudgetTimeout; private final Supplier diagnostics; private final BooleanSupplier cancelled; + private final ReentrantLock lock = new ReentrantLock(); + private final Condition progressed = lock.newCondition(); public PregenMantleBackpressure(Supplier mantleSupplier, int maxResidentTectonicPlates, int waitMs, long timeoutMs, Runnable onBudgetTimeout, Supplier diagnostics) { this(mantleSupplier, maxResidentTectonicPlates, waitMs, timeoutMs, onBudgetTimeout, diagnostics, () -> false); @@ -108,10 +113,7 @@ public final class PregenMantleBackpressure { + "), freed " + freed + " last pass, waited " + elapsed + "ms."); } - try { - Thread.sleep(waitMs); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); + if (!awaitProgress()) { return; } } @@ -154,15 +156,34 @@ public final class PregenMantleBackpressure { + (mantle != null ? " (" + mantle.getLoadedRegionCount() + " plates resident)" : "") + "."); } - try { - Thread.sleep(waitMs); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); + if (!awaitProgress()) { return; } } } + public void signalProgress() { + lock.lock(); + try { + progressed.signalAll(); + } finally { + lock.unlock(); + } + } + + private boolean awaitProgress() { + lock.lock(); + try { + progressed.await(waitMs, TimeUnit.MILLISECONDS); + return true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } finally { + lock.unlock(); + } + } + private boolean isCancelled() { if (Thread.currentThread().isInterrupted()) { return true; diff --git a/core/src/main/java/art/arcane/iris/core/pregenerator/PregenPerformanceProfile.java b/core/src/main/java/art/arcane/iris/core/pregenerator/PregenPerformanceProfile.java index 8c38746db..b59fe46de 100644 --- a/core/src/main/java/art/arcane/iris/core/pregenerator/PregenPerformanceProfile.java +++ b/core/src/main/java/art/arcane/iris/core/pregenerator/PregenPerformanceProfile.java @@ -28,8 +28,11 @@ import art.arcane.iris.util.project.stream.ProceduralStream; import art.arcane.iris.util.project.stream.utility.CachedDoubleStream2D; import java.util.Objects; +import java.util.concurrent.atomic.AtomicInteger; public final class PregenPerformanceProfile { + private static final AtomicInteger PARALLELISM_HOLDERS = new AtomicInteger(); + private PregenPerformanceProfile() { } @@ -45,12 +48,33 @@ public final class PregenPerformanceProfile { } if (MultiBurst.burst.raiseParallelism(pregenBurstParallelism(Runtime.getRuntime().availableProcessors()))) { + PARALLELISM_HOLDERS.incrementAndGet(); changed = true; } return changed; } + /** + * Releases one hold on the pregeneration burst parallelism. The pool returns to the parallelism it had + * before the first pregeneration raised it once the last concurrent job releases its hold. + */ + public static boolean restore() { + while (true) { + int holders = PARALLELISM_HOLDERS.get(); + if (holders <= 0) { + return false; + } + if (PARALLELISM_HOLDERS.compareAndSet(holders, holders - 1)) { + return holders == 1 && MultiBurst.burst.restoreParallelism(); + } + } + } + + static int parallelismHolders() { + return PARALLELISM_HOLDERS.get(); + } + public static void apply(Engine engine) { apply(); if (engine != null) { diff --git a/core/src/main/java/art/arcane/iris/core/pregenerator/cache/PregenCache.java b/core/src/main/java/art/arcane/iris/core/pregenerator/cache/PregenCache.java index 9170ffd2a..a9e8df5ac 100644 --- a/core/src/main/java/art/arcane/iris/core/pregenerator/cache/PregenCache.java +++ b/core/src/main/java/art/arcane/iris/core/pregenerator/cache/PregenCache.java @@ -6,10 +6,6 @@ import art.arcane.volmlib.util.documentation.RegionCoordinates; import java.io.File; public interface PregenCache { - default boolean isThreadSafe() { - return false; - } - @ChunkCoordinates boolean isChunkCached(int x, int z); @@ -32,16 +28,10 @@ public interface PregenCache { } default PregenCache sync() { - if (isThreadSafe()) return this; - return new SynchronizedCache(this); + return this; } PregenCache EMPTY = new PregenCache() { - @Override - public boolean isThreadSafe() { - return true; - } - @Override public boolean isChunkCached(int x, int z) { return false; diff --git a/core/src/main/java/art/arcane/iris/core/pregenerator/cache/PregenCacheImpl.java b/core/src/main/java/art/arcane/iris/core/pregenerator/cache/PregenCacheImpl.java index 5c1409493..920d25006 100644 --- a/core/src/main/java/art/arcane/iris/core/pregenerator/cache/PregenCacheImpl.java +++ b/core/src/main/java/art/arcane/iris/core/pregenerator/cache/PregenCacheImpl.java @@ -26,6 +26,9 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLongArray; +import java.util.concurrent.atomic.AtomicReferenceArray; public class PregenCacheImpl implements PregenCache { // Daemon threads: this static pool is never shut down, and non-daemon threads here block JVM exit @@ -36,12 +39,15 @@ public class PregenCacheImpl implements PregenCache { return thread; }); private static final AtomicBoolean DISPATCHER_REGISTERED = new AtomicBoolean(); - private static final short SIZE = 1024; + private static final int SIZE = 1024; + private static final int RETIRED_WRITE_RETRIES = 4; private final File directory; private final int maxSize; private final AtomicBoolean missingStorageReported = new AtomicBoolean(); + private final Object residencyLock = new Object(); private final Long2ObjectLinkedOpenHashMap cache = new Long2ObjectLinkedOpenHashMap<>(); + private volatile Plate recent; public PregenCacheImpl(File directory, int maxSize) { this.directory = directory; @@ -78,35 +84,119 @@ public class PregenCacheImpl implements PregenCache { @Override @ChunkCoordinates public void cacheChunk(int x, int z) { - getPlate(PowerOfTwoCoordinates.floorDivPow2(x, 10), PowerOfTwoCoordinates.floorDivPow2(z, 10)).cache( - PowerOfTwoCoordinates.localMaskPow2(PowerOfTwoCoordinates.floorDivPow2(x, 5), PowerOfTwoCoordinates.REGION_CHUNK_BITS), - PowerOfTwoCoordinates.localMaskPow2(PowerOfTwoCoordinates.floorDivPow2(z, 5), PowerOfTwoCoordinates.REGION_CHUNK_BITS), - region -> region.cache( - PowerOfTwoCoordinates.localMaskPow2(x, PowerOfTwoCoordinates.REGION_CHUNK_BITS), - PowerOfTwoCoordinates.localMaskPow2(z, PowerOfTwoCoordinates.REGION_CHUNK_BITS) - ) - ); + int plateX = PowerOfTwoCoordinates.floorDivPow2(x, 10); + int plateZ = PowerOfTwoCoordinates.floorDivPow2(z, 10); + int regionX = PowerOfTwoCoordinates.localMaskPow2(PowerOfTwoCoordinates.floorDivPow2(x, 5), PowerOfTwoCoordinates.REGION_CHUNK_BITS); + int regionZ = PowerOfTwoCoordinates.localMaskPow2(PowerOfTwoCoordinates.floorDivPow2(z, 5), PowerOfTwoCoordinates.REGION_CHUNK_BITS); + int localX = PowerOfTwoCoordinates.localMaskPow2(x, PowerOfTwoCoordinates.REGION_CHUNK_BITS); + int localZ = PowerOfTwoCoordinates.localMaskPow2(z, PowerOfTwoCoordinates.REGION_CHUNK_BITS); + + for (int attempt = 0; attempt < RETIRED_WRITE_RETRIES; attempt++) { + Plate plate = getPlate(plateX, plateZ); + plate.cache(regionX, regionZ, region -> region.cache(localX, localZ)); + if (!plate.retired) { + return; + } + } } @Override @RegionCoordinates public void cacheRegion(int x, int z) { - getPlate(PowerOfTwoCoordinates.floorDivPow2(x, 5), PowerOfTwoCoordinates.floorDivPow2(z, 5)).cache( - PowerOfTwoCoordinates.localMaskPow2(x, PowerOfTwoCoordinates.REGION_CHUNK_BITS), - PowerOfTwoCoordinates.localMaskPow2(z, PowerOfTwoCoordinates.REGION_CHUNK_BITS), - Region::cache - ); + int plateX = PowerOfTwoCoordinates.floorDivPow2(x, 5); + int plateZ = PowerOfTwoCoordinates.floorDivPow2(z, 5); + int regionX = PowerOfTwoCoordinates.localMaskPow2(x, PowerOfTwoCoordinates.REGION_CHUNK_BITS); + int regionZ = PowerOfTwoCoordinates.localMaskPow2(z, PowerOfTwoCoordinates.REGION_CHUNK_BITS); + + for (int attempt = 0; attempt < RETIRED_WRITE_RETRIES; attempt++) { + Plate plate = getPlate(plateX, plateZ); + plate.cache(regionX, regionZ, Region::cache); + if (!plate.retired) { + return; + } + } } @Override public void write() { - if (cache.isEmpty()) { + List resident; + synchronized (residencyLock) { + if (cache.isEmpty()) { + return; + } + resident = new ArrayList<>(cache.values()); + } + + writePlates(resident); + } + + @Override + public void trim(long unloadDuration) { + synchronized (residencyLock) { + if (cache.isEmpty()) { + return; + } + + long threshold = System.currentTimeMillis() - unloadDuration; + List resident = new ArrayList<>(cache.size()); + Iterator iterator = cache.values().iterator(); + while (iterator.hasNext()) { + Plate plate = iterator.next(); + if (plate.lastAccess < threshold) { + iterator.remove(); + plate.retired = true; + } + resident.add(plate); + } + + writePlates(resident); + } + } + + private Plate getPlate(int x, int z) { + long key = key(x, z); + Plate memo = recent; + if (memo != null && memo.key == key && !memo.retired) { + return memo; + } + + return residePlate(key, x, z); + } + + private Plate residePlate(long key, int x, int z) { + Plate resolved; + synchronized (residencyLock) { + resolved = cache.getAndMoveToFirst(key); + if (resolved == null) { + resolved = readPlate(key, x, z); + cache.putAndMoveToFirst(key, resolved); + List evicted = null; + while (cache.size() > maxSize) { + Plate removed = cache.removeLast(); + removed.retired = true; + if (evicted == null) { + evicted = new ArrayList<>(2); + } + evicted.add(removed); + } + if (evicted != null) { + writePlates(evicted); + } + } + } + + recent = resolved; + return resolved; + } + + private void writePlates(List plates) { + if (plates.isEmpty()) { return; } - List> futures = new ArrayList<>(cache.size()); - for (Plate plate : cache.values()) { - if (!plate.dirty) { + List> futures = new ArrayList<>(plates.size()); + for (Plate plate : plates) { + if (!plate.dirty.get()) { continue; } futures.add(CompletableFuture.runAsync(() -> writePlate(plate), DISPATCHER)); @@ -117,84 +207,41 @@ public class PregenCacheImpl implements PregenCache { } } - @Override - public void trim(long unloadDuration) { - if (cache.isEmpty()) { - return; - } - - long threshold = System.currentTimeMillis() - unloadDuration; - List> futures = new ArrayList<>(cache.size()); - Iterator iterator = cache.values().iterator(); - while (iterator.hasNext()) { - Plate plate = iterator.next(); - if (plate.lastAccess < threshold) { - iterator.remove(); - } - futures.add(CompletableFuture.runAsync(() -> writePlate(plate), DISPATCHER)); - } - - for (CompletableFuture future : futures) { - future.join(); - } - } - - private Plate getPlate(int x, int z) { - long key = key(x, z); - Plate plate = cache.getAndMoveToFirst(key); - if (plate != null) { - return plate; - } - - Plate loaded = readPlate(x, z); - cache.putAndMoveToFirst(key, loaded); - - if (cache.size() > maxSize) { - List> futures = new ArrayList<>(cache.size() - maxSize); - while (cache.size() > maxSize) { - Plate evicted = cache.removeLast(); - futures.add(CompletableFuture.runAsync(() -> writePlate(evicted), DISPATCHER)); - } - for (CompletableFuture future : futures) { - future.join(); - } - } - - return loaded; - } - - private Plate readPlate(int x, int z) { + private Plate readPlate(long key, int x, int z) { File file = fileForPlate(x, z); if (!file.exists()) { - return new Plate(x, z); + return new Plate(key, x, z); } try (DataInputStream input = new DataInputStream(new LZ4BlockInputStream(new FileInputStream(file)))) { - return readPlate(x, z, input); + return readPlate(key, x, z, input); } catch (IOException e) { IrisLogging.error("Failed to read pregen cache " + file); IrisLogging.reportError(e); } - return new Plate(x, z); + return new Plate(key, x, z); } private void writePlate(Plate plate) { - if (!plate.dirty) { - return; - } - if (!prepareCacheDirectory()) { - return; - } + synchronized (plate) { + if (!plate.dirty.compareAndSet(true, false)) { + return; + } + if (!prepareCacheDirectory()) { + plate.dirty.set(true); + return; + } - File file = null; - try { - file = fileForPlate(plate.x, plate.z); - IO.write(file, output -> new DataOutputStream(new LZ4BlockOutputStream(output)), plate::write); - plate.dirty = false; - } catch (Throwable e) { - IrisLogging.error("Failed to write pregen cache " + (file != null ? file : "c." + plate.x + "." + plate.z)); - IrisLogging.reportError(e); + File file = null; + try { + file = fileForPlate(plate.x, plate.z); + IO.write(file, output -> new DataOutputStream(new LZ4BlockOutputStream(output)), plate::write); + } catch (Throwable e) { + plate.dirty.set(true); + IrisLogging.error("Failed to write pregen cache " + (file != null ? file : "c." + plate.x + "." + plate.z)); + IrisLogging.reportError(e); + } } } @@ -243,36 +290,48 @@ public class PregenCacheImpl implements PregenCache { } private static class Plate { + private final long key; private final int x; private final int z; - private short count; - private Region[] regions; - private boolean dirty; - private long lastAccess; + private final AtomicInteger count; + private final AtomicReferenceArray regions; + private final AtomicBoolean dirty = new AtomicBoolean(); + private volatile long lastAccess; + private volatile boolean retired; - private Plate(int x, int z) { - this(x, z, (short) 0, new Region[1024]); + private Plate(long key, int x, int z) { + this(key, x, z, 0, new AtomicReferenceArray<>(SIZE)); } - private Plate(int x, int z, short count, Region[] regions) { + private Plate(long key, int x, int z, int count, AtomicReferenceArray regions) { + this.key = key; this.x = x; this.z = z; - this.count = count; + this.count = new AtomicInteger(count); this.regions = regions; this.lastAccess = System.currentTimeMillis(); } + private int completed() { + return Math.min(count.get(), SIZE); + } + private boolean cache(int x, int z, RegionCacheOp op) { lastAccess = System.currentTimeMillis(); - if (count == SIZE) { + if (completed() == SIZE) { + return false; + } + + AtomicReferenceArray current = regions; + if (current == null) { return false; } int index = PowerOfTwoCoordinates.packLocal32(x, z); - Region region = regions[index]; + Region region = current.get(index); if (region == null) { - region = new Region(); - regions[index] = region; + Region created = new Region(); + region = current.compareAndSet(index, null, created) ? created : current.get(index); } CacheResult result = op.apply(region); @@ -280,25 +339,26 @@ public class PregenCacheImpl implements PregenCache { return false; } - dirty = true; + dirty.set(true); if (result != CacheResult.COMPLETED) { return false; } - count++; - if (count == SIZE) { - regions = null; - } - return true; + return count.incrementAndGet() >= SIZE; } private boolean isCached(int x, int z, RegionPredicate predicate) { lastAccess = System.currentTimeMillis(); - if (count == SIZE) { + if (completed() == SIZE) { return true; } - Region region = regions[PowerOfTwoCoordinates.packLocal32(x, z)]; + AtomicReferenceArray current = regions; + if (current == null) { + return true; + } + + Region region = current.get(PowerOfTwoCoordinates.packLocal32(x, z)); if (region == null) { return false; } @@ -306,12 +366,15 @@ public class PregenCacheImpl implements PregenCache { } private void write(DataOutput output) throws IOException { - Varint.writeSignedVarInt(count, output); - if (regions == null) { + int total = completed(); + Varint.writeSignedVarInt(total, output); + AtomicReferenceArray current = regions; + if (total == SIZE || current == null) { return; } - for (Region region : regions) { + for (int index = 0; index < SIZE; index++) { + Region region = current.get(index); output.writeBoolean(region == null); if (region != null) { region.write(output); @@ -321,33 +384,32 @@ public class PregenCacheImpl implements PregenCache { } private static class Region { - private short count; - private long[] words; + private final AtomicInteger count; + private final AtomicLongArray words; private Region() { - this((short) 0, new long[64]); + this(0, new AtomicLongArray(64)); } - private Region(short count, long[] words) { - this.count = count; + private Region(int count, AtomicLongArray words) { + this.count = new AtomicInteger(count); this.words = words; } + private int completed() { + return Math.min(count.get(), SIZE); + } + private CacheResult cache() { - if (count == SIZE) { - return CacheResult.NOOP; - } - count = SIZE; - words = null; - return CacheResult.COMPLETED; + return count.getAndSet(SIZE) >= SIZE ? CacheResult.NOOP : CacheResult.COMPLETED; } private CacheResult cache(int x, int z) { - if (count == SIZE) { + if (completed() == SIZE) { return CacheResult.NOOP; } - long[] value = words; + AtomicLongArray value = words; if (value == null) { return CacheResult.NOOP; } @@ -355,73 +417,79 @@ public class PregenCacheImpl implements PregenCache { int index = PowerOfTwoCoordinates.packLocal32(x, z); int wordIndex = index >> 6; long bit = 1L << (index & 63); - boolean current = (value[wordIndex] & bit) != 0L; - if (current) { - return CacheResult.NOOP; + while (true) { + long current = value.get(wordIndex); + if ((current & bit) != 0L) { + return CacheResult.NOOP; + } + if (value.compareAndSet(wordIndex, current, current | bit)) { + break; + } } - count++; - if (count == SIZE) { - words = null; - return CacheResult.COMPLETED; - } - - value[wordIndex] = value[wordIndex] | bit; - return CacheResult.SET; + return count.incrementAndGet() >= SIZE ? CacheResult.COMPLETED : CacheResult.SET; } private boolean isCached() { - return count == SIZE; + return completed() == SIZE; } private boolean isCached(int x, int z) { - int index = PowerOfTwoCoordinates.packLocal32(x, z); - if (count == SIZE) { + if (completed() == SIZE) { return true; } - return (words[index >> 6] & (1L << (index & 63))) != 0L; + + AtomicLongArray value = words; + if (value == null) { + return true; + } + + int index = PowerOfTwoCoordinates.packLocal32(x, z); + return (value.get(index >> 6) & (1L << (index & 63))) != 0L; } private void write(DataOutput output) throws IOException { - Varint.writeSignedVarInt(count, output); - if (words == null) { + int total = completed(); + Varint.writeSignedVarInt(total, output); + AtomicLongArray value = words; + if (total == SIZE || value == null) { return; } - for (long word : words) { - Varint.writeUnsignedVarLong(word, output); + for (int index = 0; index < value.length(); index++) { + Varint.writeUnsignedVarLong(value.get(index), output); } } } - private static Plate readPlate(int x, int z, DataInput input) throws IOException { + private static Plate readPlate(long key, int x, int z, DataInput input) throws IOException { int count = Varint.readSignedVarInt(input); - if (count == 1024) { - return new Plate(x, z, SIZE, null); + if (count == SIZE) { + return new Plate(key, x, z, SIZE, null); } - Region[] regions = new Region[1024]; - for (int i = 0; i < regions.length; i++) { + AtomicReferenceArray regions = new AtomicReferenceArray<>(SIZE); + for (int index = 0; index < SIZE; index++) { boolean isNull = input.readBoolean(); if (!isNull) { - regions[i] = readRegion(input); + regions.set(index, readRegion(input)); } } - return new Plate(x, z, (short) count, regions); + return new Plate(key, x, z, count, regions); } private static Region readRegion(DataInput input) throws IOException { int count = Varint.readSignedVarInt(input); - if (count == 1024) { + if (count == SIZE) { return new Region(SIZE, null); } - long[] words = new long[64]; - for (int i = 0; i < words.length; i++) { - words[i] = Varint.readUnsignedVarLong(input); + AtomicLongArray words = new AtomicLongArray(64); + for (int index = 0; index < words.length(); index++) { + words.set(index, Varint.readUnsignedVarLong(input)); } - return new Region((short) count, words); + return new Region(count, words); } } diff --git a/core/src/main/java/art/arcane/iris/core/pregenerator/cache/SynchronizedCache.java b/core/src/main/java/art/arcane/iris/core/pregenerator/cache/SynchronizedCache.java deleted file mode 100644 index 2605ada76..000000000 --- a/core/src/main/java/art/arcane/iris/core/pregenerator/cache/SynchronizedCache.java +++ /dev/null @@ -1,50 +0,0 @@ -package art.arcane.iris.core.pregenerator.cache; - -record SynchronizedCache(PregenCache cache) implements PregenCache { - @Override - public boolean isThreadSafe() { - return true; - } - - @Override - public boolean isChunkCached(int x, int z) { - synchronized (cache) { - return cache.isChunkCached(x, z); - } - } - - @Override - public boolean isRegionCached(int x, int z) { - synchronized (cache) { - return cache.isRegionCached(x, z); - } - } - - @Override - public void cacheChunk(int x, int z) { - synchronized (cache) { - cache.cacheChunk(x, z); - } - } - - @Override - public void cacheRegion(int x, int z) { - synchronized (cache) { - cache.cacheRegion(x, z); - } - } - - @Override - public void write() { - synchronized (cache) { - cache.write(); - } - } - - @Override - public void trim(long unloadDuration) { - synchronized (cache) { - cache.trim(unloadDuration); - } - } -} 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 811b5f285..53218d1c7 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 @@ -22,6 +22,7 @@ import art.arcane.iris.spi.IrisLogging; import art.arcane.iris.core.IrisPaperLikeBackendMode; import art.arcane.iris.core.IrisRuntimeSchedulerMode; import art.arcane.iris.core.IrisSettings; +import art.arcane.iris.core.pregenerator.PregenAdmissionGate; import art.arcane.iris.core.pregenerator.PregenDiagnostics; import art.arcane.iris.core.pregenerator.PregenListener; import art.arcane.iris.core.pregenerator.PregenMantleBackpressure; @@ -51,7 +52,6 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; @@ -68,6 +68,7 @@ public class AsyncPregenMethod implements PregeneratorMethod { private static final int ADAPTIVE_RECOVERY_INTERVAL = 8; private static final long CLOSE_DRAIN_WARNING_SECONDS = 60L; private static final long FLUSH_TIMEOUT_SECONDS = 120L; + private static final long ADMISSION_WAIT_BOUND_MS = 500L; private final World world; private final IrisRuntimeSchedulerMode runtimeSchedulerMode; private final IrisPaperLikeBackendMode paperLikeBackendMode; @@ -82,7 +83,7 @@ public class AsyncPregenMethod implements PregeneratorMethod { private final String chunkAccessMode; private final ChunkRequestExecutor executor; private final Executor slowRequestExecutor; - private final Semaphore semaphore; + private final PregenAdmissionGate admission; private final int threads; private final int slowRequestWarningSeconds; private final int slowRequestWarnIntervalMs; @@ -111,7 +112,6 @@ public class AsyncPregenMethod implements PregeneratorMethod { private final AtomicLong lastProgressAt = new AtomicLong(M.ms()); private final AtomicBoolean closing = new AtomicBoolean(); private final AtomicBoolean holdsWorkerBoost = new AtomicBoolean(); - private final Object permitMonitor = new Object(); private volatile Engine metricsEngine; private volatile Mantle cachedMantle; private final PregenMantleBackpressure backpressure; @@ -161,7 +161,7 @@ public class AsyncPregenMethod implements PregeneratorMethod { this.runtimeCpuThreads = detectedCpuThreads; this.effectiveWorkerThreads = workerThreadsForCap; this.recommendedRuntimeConcurrencyCap = configuredThreads; - this.semaphore = new Semaphore(this.threads, true); + this.admission = new PregenAdmissionGate(this.threads, ADMISSION_WAIT_BOUND_MS, M::ms); this.slowRequestWarningSeconds = pregen.getChunkLoadTimeoutSeconds(); this.slowRequestExecutor = CompletableFuture.delayedExecutor(this.slowRequestWarningSeconds, TimeUnit.SECONDS); this.slowRequestWarnIntervalMs = pregen.getTimeoutWarnIntervalMs(); @@ -527,11 +527,7 @@ public class AsyncPregenMethod implements PregeneratorMethod { } catch (Throwable e) { IrisLogging.reportError(e); } finally { - try { - markFinished(success); - } finally { - semaphore.release(); - } + markFinished(success); } } @@ -587,7 +583,7 @@ public class AsyncPregenMethod implements PregeneratorMethod { int next = Math.max(adaptiveMinInFlightLimit, current - 1); if (adaptiveInFlightLimit.compareAndSet(current, next)) { logAdaptiveLimit("decrease", next); - notifyPermitWaiters(); + admission.wake(); return; } } @@ -603,7 +599,7 @@ public class AsyncPregenMethod implements PregeneratorMethod { int next = nextAdaptiveInFlightLimit(current, threads); if (adaptiveInFlightLimit.compareAndSet(current, next)) { logAdaptiveLimit("increase", next); - notifyPermitWaiters(); + admission.wake(); return; } } @@ -689,7 +685,7 @@ public class AsyncPregenMethod implements PregeneratorMethod { private String metricsSnapshot() { long stalledFor = Math.max(0L, M.ms() - lastProgressAt.get()); return "world=" + world.getName() - + " permits=" + semaphore.availablePermits() + "/" + threads + + " permits=" + admission.availablePermits() + "/" + threads + " adaptiveLimit=" + adaptiveInFlightLimit.get() + " inFlight=" + inFlight.get() + " submitted=" + submitted.get() @@ -716,13 +712,8 @@ public class AsyncPregenMethod implements PregeneratorMethod { if (after < 0) { inFlight.compareAndSet(after, 0); } - notifyPermitWaiters(); - } - - private void notifyPermitWaiters() { - synchronized (permitMonitor) { - permitMonitor.notifyAll(); - } + admission.release(); + backpressure.signalProgress(); } private void recordAdaptiveWait(long waitedMs) { @@ -818,14 +809,12 @@ public class AsyncPregenMethod implements PregeneratorMethod { @Override public void close() { closing.set(true); - notifyPermitWaiters(); + admission.wake(); // A stop request interrupts the pregen worker; shield the drain and flush so chunks still hit disk. boolean interrupted = Thread.interrupted(); try { - interrupted |= awaitDrain( - semaphore, - threads, + interrupted |= admission.awaitDrain( CLOSE_DRAIN_WARNING_SECONDS, TimeUnit.SECONDS, () -> IrisLogging.warn("Async pregen is still draining outstanding chunks. " + metricsSnapshot()) @@ -843,26 +832,6 @@ public class AsyncPregenMethod implements PregeneratorMethod { } } - static boolean awaitDrain( - Semaphore semaphore, - int permits, - long warningInterval, - TimeUnit timeUnit, - Runnable onWait - ) { - boolean interrupted = false; - while (true) { - try { - if (semaphore.tryAcquire(permits, warningInterval, timeUnit)) { - return interrupted; - } - onWait.run(); - } catch (InterruptedException e) { - interrupted = true; - } - } - } - private boolean isCancelled() { return closing.get() || Thread.currentThread().isInterrupted(); } @@ -890,37 +859,26 @@ public class AsyncPregenMethod implements PregeneratorMethod { return; } + PregenAdmissionGate.Wait waited; try { - long waitStart = M.ms(); - synchronized (permitMonitor) { - while (inFlight.get() >= adaptiveInFlightLimit.get()) { - if (isCancelled()) { - return; - } - - permitMonitor.wait(500L); - } - } - long adaptiveWait = Math.max(0L, M.ms() - waitStart); - if (adaptiveWait > 0L) { - recordAdaptiveWait(adaptiveWait); - } - - long permitWaitStart = M.ms(); - while (!semaphore.tryAcquire(1, TimeUnit.SECONDS)) { - if (isCancelled()) { - return; - } - } - long permitWait = Math.max(0L, M.ms() - permitWaitStart); - if (permitWait > 0L) { - recordPermitWait(permitWait); - } + waited = admission.admit(() -> inFlight.get() >= adaptiveInFlightLimit.get(), this::isCancelled); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } + if (waited == null) { + return; + } + + if (waited.adaptiveMs() > 0L) { + recordAdaptiveWait(waited.adaptiveMs()); + } + + if (waited.permitMs() > 0L) { + recordPermitWait(waited.permitMs()); + } + regionPending.computeIfAbsent(rkey(x >> 5, z >> 5), k -> new AtomicInteger(1)).incrementAndGet(); markSubmitted(); executor.generate(x, z, listener); diff --git a/core/src/main/java/art/arcane/iris/util/common/parallel/MultiBurst.java b/core/src/main/java/art/arcane/iris/util/common/parallel/MultiBurst.java index 88d6c48d6..f918668a5 100644 --- a/core/src/main/java/art/arcane/iris/util/common/parallel/MultiBurst.java +++ b/core/src/main/java/art/arcane/iris/util/common/parallel/MultiBurst.java @@ -10,6 +10,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.ForkJoinWorkerThread; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.IntSupplier; public class MultiBurst extends MultiBurstSupport { @@ -21,6 +22,7 @@ public class MultiBurst extends MultiBurstSupport { * plans never pin the workers that chunk generation's short stage and mantle tasks need. */ public static final MultiBurst hydrology = new MultiBurst("Iris Hydrology", () -> Math.max(2, IrisSettings.getThreadCount(IrisSettings.get().getConcurrency().getParallelism()))); + private final AtomicInteger parallelismBaseline = new AtomicInteger(); public MultiBurst() { this("Iris"); @@ -41,16 +43,38 @@ public class MultiBurst extends MultiBurstSupport { /** * Raises the pool's parallelism to at least {@code target} workers. Pregeneration fans every * chunk's stages and mantle windows into this pool while some tasks block on unmanaged waits, - * so a pregeneration box wants more workers than cores; the pool never shrinks back. + * so a pregeneration box wants more workers than cores. {@link #restoreParallelism()} puts it back. */ public boolean raiseParallelism(int target) { if (!(service() instanceof ForkJoinPool pool) || target <= pool.getParallelism()) { return false; } + parallelismBaseline.compareAndSet(0, pool.getParallelism()); pool.setParallelism(target); return true; } + /** + * Returns the pool to the parallelism it had before the first {@link #raiseParallelism(int)} that took + * effect. A pool that was never raised, or that is not a {@link ForkJoinPool}, is left alone. + */ + public boolean restoreParallelism() { + int baseline = parallelismBaseline.getAndSet(0); + if (baseline <= 0 || !(service() instanceof ForkJoinPool pool) || pool.getParallelism() <= baseline) { + return false; + } + + try { + pool.setParallelism(baseline); + } catch (RuntimeException e) { + IrisLogging.reportError("Iris could not return the burst pool to its pre-pregeneration parallelism of " + + baseline + "; it stays at " + pool.getParallelism() + ".", e); + return false; + } + + return true; + } + public int parallelism() { return service() instanceof ForkJoinPool pool ? pool.getParallelism() : -1; } diff --git a/core/src/test/java/art/arcane/iris/core/pregenerator/PregenAdmissionGateTest.java b/core/src/test/java/art/arcane/iris/core/pregenerator/PregenAdmissionGateTest.java new file mode 100644 index 000000000..10e37013c --- /dev/null +++ b/core/src/test/java/art/arcane/iris/core/pregenerator/PregenAdmissionGateTest.java @@ -0,0 +1,189 @@ +package art.arcane.iris.core.pregenerator; + +import org.junit.Test; + +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.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +public class PregenAdmissionGateTest { + private static final long GENEROUS_BOUND_MS = 60_000L; + + @Test + public void admissionConsumesOnePermitAndReturnsItOnRelease() throws Exception { + PregenAdmissionGate gate = new PregenAdmissionGate(2, GENEROUS_BOUND_MS, System::currentTimeMillis); + + assertEquals(2, gate.availablePermits()); + assertNotNull(gate.admit(() -> false, () -> false)); + assertEquals(1, gate.availablePermits()); + gate.release(); + assertEquals(2, gate.availablePermits()); + } + + @Test + public void releaseNeverPushesThePermitCountAboveTheConfiguredCeiling() throws Exception { + PregenAdmissionGate gate = new PregenAdmissionGate(1, GENEROUS_BOUND_MS, System::currentTimeMillis); + + gate.release(); + gate.release(); + + assertEquals(1, gate.availablePermits()); + assertNotNull(gate.admit(() -> false, () -> false)); + assertEquals(0, gate.availablePermits()); + } + + @Test + public void aBlockedAdmissionWakesOnTheReleaseSignalRatherThanTheTimeout() throws Exception { + PregenAdmissionGate gate = new PregenAdmissionGate(1, GENEROUS_BOUND_MS, System::currentTimeMillis); + assertNotNull(gate.admit(() -> false, () -> false)); + CountDownLatch started = new CountDownLatch(1); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future blocked = executor.submit(() -> { + started.countDown(); + return gate.admit(() -> false, () -> false); + }); + + assertTrue(started.await(5, TimeUnit.SECONDS)); + while (gate.availablePermits() != 0) { + Thread.onSpinWait(); + } + gate.release(); + + assertNotNull(blocked.get(5, TimeUnit.SECONDS)); + } finally { + executor.shutdownNow(); + } + } + + @Test + public void aBlockedAdmissionWakesOnTheAdaptiveLimitSignalWithoutWaitingOutTheBound() throws Exception { + PregenAdmissionGate gate = new PregenAdmissionGate(4, GENEROUS_BOUND_MS, System::currentTimeMillis); + AtomicBoolean overLimit = new AtomicBoolean(true); + CountDownLatch started = new CountDownLatch(1); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future blocked = executor.submit(() -> { + started.countDown(); + return gate.admit(overLimit::get, () -> false); + }); + + assertTrue(started.await(5, TimeUnit.SECONDS)); + Thread.yield(); + overLimit.set(false); + gate.wake(); + + assertNotNull(blocked.get(5, TimeUnit.SECONDS)); + assertEquals(3, gate.availablePermits()); + } finally { + executor.shutdownNow(); + } + } + + @Test + public void aCancelledAdmissionReturnsNothingAndConsumesNoPermit() throws Exception { + PregenAdmissionGate gate = new PregenAdmissionGate(1, GENEROUS_BOUND_MS, System::currentTimeMillis); + + assertNull(gate.admit(() -> true, () -> true)); + assertEquals(1, gate.availablePermits()); + } + + @Test + public void aCancellationRaisedWhileBlockedEndsTheWait() throws Exception { + PregenAdmissionGate gate = new PregenAdmissionGate(1, GENEROUS_BOUND_MS, System::currentTimeMillis); + assertNotNull(gate.admit(() -> false, () -> false)); + AtomicBoolean cancelled = new AtomicBoolean(); + CountDownLatch started = new CountDownLatch(1); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future blocked = executor.submit(() -> { + started.countDown(); + return gate.admit(() -> false, cancelled::get); + }); + + assertTrue(started.await(5, TimeUnit.SECONDS)); + Thread.yield(); + cancelled.set(true); + gate.wake(); + + assertNull(blocked.get(5, TimeUnit.SECONDS)); + assertEquals(0, gate.availablePermits()); + } finally { + executor.shutdownNow(); + } + } + + @Test + public void theAdaptiveAndPermitWaitsAreAttributedSeparately() throws Exception { + AtomicLong clock = new AtomicLong(); + AtomicBoolean overLimit = new AtomicBoolean(true); + PregenAdmissionGate gate = new PregenAdmissionGate(1, 5L, () -> clock.addAndGet(10L)); + assertNotNull(gate.admit(() -> false, () -> false)); + + AtomicInteger passes = new AtomicInteger(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future blocked = executor.submit(() -> gate.admit(() -> { + if (passes.incrementAndGet() > 2) { + overLimit.set(false); + } + return overLimit.get(); + }, () -> false)); + + while (passes.get() <= 3) { + Thread.onSpinWait(); + } + gate.release(); + + PregenAdmissionGate.Wait wait = blocked.get(5, TimeUnit.SECONDS); + assertNotNull(wait); + assertTrue("adaptive wait was not attributed: " + wait, wait.adaptiveMs() > 0L); + assertTrue("permit wait was not attributed: " + wait, wait.permitMs() > 0L); + } finally { + executor.shutdownNow(); + } + } + + @Test + public void drainReturnsOnlyAfterEveryPermitIsBack() throws Exception { + PregenAdmissionGate gate = new PregenAdmissionGate(2, GENEROUS_BOUND_MS, System::currentTimeMillis); + assertNotNull(gate.admit(() -> false, () -> false)); + assertNotNull(gate.admit(() -> false, () -> false)); + AtomicInteger warnings = new AtomicInteger(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future drain = executor.submit(() -> gate.awaitDrain(5L, TimeUnit.MILLISECONDS, warnings::incrementAndGet)); + + while (warnings.get() < 2) { + Thread.onSpinWait(); + } + gate.release(); + gate.release(); + + assertEquals(Boolean.FALSE, drain.get(5, TimeUnit.SECONDS)); + assertEquals(2, gate.availablePermits()); + } finally { + executor.shutdownNow(); + } + } + + @Test(expected = IllegalArgumentException.class) + public void aGateWithoutPermitsIsRejected() { + new PregenAdmissionGate(0, GENEROUS_BOUND_MS, System::currentTimeMillis); + } + + @Test(expected = IllegalArgumentException.class) + public void aGateWithoutAWaitBoundIsRejected() { + new PregenAdmissionGate(1, 0L, System::currentTimeMillis); + } +} diff --git a/core/src/test/java/art/arcane/iris/core/pregenerator/PregenPerformanceProfileTest.java b/core/src/test/java/art/arcane/iris/core/pregenerator/PregenPerformanceProfileTest.java index 2ea3869fe..56966becb 100644 --- a/core/src/test/java/art/arcane/iris/core/pregenerator/PregenPerformanceProfileTest.java +++ b/core/src/test/java/art/arcane/iris/core/pregenerator/PregenPerformanceProfileTest.java @@ -107,4 +107,41 @@ public class PregenPerformanceProfileTest { assertEquals(before + 1, MultiBurst.burst.parallelism()); assertFalse(MultiBurst.burst.raiseParallelism(before)); } + + @Test + public void restoringBurstParallelismReturnsThePoolToItsPrePregenerationSize() { + int before = MultiBurst.ioBurst.parallelism(); + assertTrue(MultiBurst.ioBurst.raiseParallelism(before + 3)); + assertEquals(before + 3, MultiBurst.ioBurst.parallelism()); + + assertTrue(MultiBurst.ioBurst.restoreParallelism()); + assertEquals(before, MultiBurst.ioBurst.parallelism()); + assertFalse(MultiBurst.ioBurst.restoreParallelism()); + assertEquals(before, MultiBurst.ioBurst.parallelism()); + } + + @Test + public void aSecondRaiseKeepsTheOriginalBaselineSoOneRestoreIsEnough() { + int before = MultiBurst.ioBurst.parallelism(); + assertTrue(MultiBurst.ioBurst.raiseParallelism(before + 2)); + assertTrue(MultiBurst.ioBurst.raiseParallelism(before + 5)); + + assertTrue(MultiBurst.ioBurst.restoreParallelism()); + assertEquals(before, MultiBurst.ioBurst.parallelism()); + } + + @Test + public void theBurstParallelismHoldIsReleasedByAMatchingRestore() { + while (PregenPerformanceProfile.parallelismHolders() > 0) { + PregenPerformanceProfile.restore(); + } + assertFalse(PregenPerformanceProfile.restore()); + assertEquals(0, PregenPerformanceProfile.parallelismHolders()); + + PregenPerformanceProfile.apply(); + int holders = PregenPerformanceProfile.parallelismHolders(); + PregenPerformanceProfile.restore(); + + assertEquals(Math.max(0, holders - 1), PregenPerformanceProfile.parallelismHolders()); + } } diff --git a/core/src/test/java/art/arcane/iris/core/pregenerator/PregenWaitPolicyContractTest.java b/core/src/test/java/art/arcane/iris/core/pregenerator/PregenWaitPolicyContractTest.java new file mode 100644 index 000000000..e9908925b --- /dev/null +++ b/core/src/test/java/art/arcane/iris/core/pregenerator/PregenWaitPolicyContractTest.java @@ -0,0 +1,111 @@ +package art.arcane.iris.core.pregenerator; + +import org.junit.Test; + +import java.io.IOException; +import java.net.URISyntaxException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Stream; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class PregenWaitPolicyContractTest { + @Test + public void noPregenSourceSleepPollsWhileWaitingForCapacity() throws IOException { + List offenders = new ArrayList<>(); + Path pregenRoot = sourceRoot().resolve("art/arcane/iris/core/pregenerator"); + try (Stream sources = Files.walk(pregenRoot)) { + for (Path source : sources.filter(path -> path.toString().endsWith(".java")).toList()) { + String text = Files.readString(source); + if (text.contains("Thread.sleep(") || text.contains("J.sleep(")) { + offenders.add(source.getFileName().toString()); + } + } + } + + assertEquals("sleep polling on the pregen path: " + offenders, List.of(), offenders); + } + + @Test + public void mantleBackpressureWaitsOnAProgressSignalWithABoundedTimeout() throws IOException { + String backpressure = read("art/arcane/iris/core/pregenerator/PregenMantleBackpressure.java"); + + assertTrue(backpressure.contains("progressed.await(waitMs, TimeUnit.MILLISECONDS)")); + assertTrue(method(backpressure, "public void enforceMantleBudget()").contains("awaitProgress()")); + assertTrue(method(backpressure, "public void awaitHeapHeadroom()").contains("awaitProgress()")); + } + + @Test + public void chunkCompletionSignalsBothTheAdmissionGateAndTheBackpressureWait() throws IOException { + String finished = method(read("art/arcane/iris/core/pregenerator/methods/AsyncPregenMethod.java"), + "private void markFinished(boolean success)"); + + assertTrue(finished, finished.contains("admission.release()")); + assertTrue(finished, finished.contains("backpressure.signalProgress()")); + } + + @Test + public void pauseResumeAndStopSignalThePregenGateInsteadOfLettingItPoll() throws IOException { + String pregenerator = read("art/arcane/iris/core/pregenerator/IrisPregenerator.java"); + + assertTrue(method(pregenerator, "public void pause()").contains("signalState()")); + assertTrue(method(pregenerator, "public void resume()").contains("signalState()")); + assertTrue(method(pregenerator, "public void close()").contains("signalState()")); + assertTrue(method(pregenerator, "private void visitRegion(").contains("awaitStateChange(GATE_WAIT_MS)")); + assertTrue(method(pregenerator, "private void reclaimTectonicPlates(").contains("awaitStateChange(RECLAIM_WAIT_MS)")); + } + + @Test + public void aFinishedPregenReleasesTheBurstParallelismItRaised() throws IOException { + String shutdown = method(read("art/arcane/iris/core/pregenerator/IrisPregenerator.java"), "private void shutdown()"); + + assertTrue(shutdown, shutdown.contains("shutdownStep(\"parallelism\", PregenPerformanceProfile::restore)")); + } + + private static Path sourceRoot() throws IOException { + Path current; + try { + current = Path.of(PregenWaitPolicyContractTest.class.getProtectionDomain() + .getCodeSource() + .getLocation() + .toURI()); + } catch (URISyntaxException e) { + throw new IOException("Unable to resolve the core source root", e); + } + while (current != null && !Files.isDirectory(current.resolve("src/main/java"))) { + current = current.getParent(); + } + if (current == null) { + throw new IOException("Unable to locate the core source root"); + } + return current.resolve("src/main/java"); + } + + private static String read(String relativePath) throws IOException { + return Files.readString(sourceRoot().resolve(relativePath)); + } + + private static String method(String source, String signature) { + int start = source.indexOf(signature); + assertTrue("Missing source contract signature: " + signature, start >= 0); + int openBrace = source.indexOf('{', start); + assertTrue("Missing source contract method body: " + signature, openBrace >= 0); + int depth = 0; + for (int index = openBrace; index < source.length(); index++) { + char current = source.charAt(index); + if (current == '{') { + depth++; + } else if (current == '}') { + depth--; + if (depth == 0) { + return source.substring(start, index + 1); + } + } + } + throw new IllegalArgumentException("Unclosed source contract method: " + signature); + } +} diff --git a/core/src/test/java/art/arcane/iris/core/pregenerator/cache/PregenCacheConcurrencyTest.java b/core/src/test/java/art/arcane/iris/core/pregenerator/cache/PregenCacheConcurrencyTest.java new file mode 100644 index 000000000..a960113b6 --- /dev/null +++ b/core/src/test/java/art/arcane/iris/core/pregenerator/cache/PregenCacheConcurrencyTest.java @@ -0,0 +1,194 @@ +package art.arcane.iris.core.pregenerator.cache; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CyclicBarrier; +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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class PregenCacheConcurrencyTest { + private static final int THREADS = 8; + + private File directory; + private ExecutorService pool; + + @Before + public void setUp() throws Exception { + directory = Files.createTempDirectory("iris-pregen-cache-concurrency").toFile(); + pool = Executors.newFixedThreadPool(THREADS); + } + + @After + public void tearDown() throws Exception { + pool.shutdownNow(); + pool.awaitTermination(30, TimeUnit.SECONDS); + deleteRecursively(directory); + } + + @Test + public void everyConcurrentlyCachedChunkIsReadBackAsCached() throws Exception { + PregenCache cache = PregenCache.create(directory); + int chunksPerThread = 512; + runAll(thread -> () -> { + for (int index = 0; index < chunksPerThread; index++) { + int x = (thread * chunksPerThread) + index; + cache.cacheChunk(x, thread); + } + return null; + }); + + for (int thread = 0; thread < THREADS; thread++) { + for (int index = 0; index < chunksPerThread; index++) { + int x = (thread * chunksPerThread) + index; + assertTrue("chunk " + x + "," + thread, cache.isChunkCached(x, thread)); + } + } + } + + @Test + public void concurrentWritersToOneRegionNeverLoseABit() throws Exception { + PregenCache cache = PregenCache.create(directory); + runAll(thread -> () -> { + for (int index = thread; index < 1024; index += THREADS) { + cache.cacheChunk(index & 31, index >> 5); + } + return null; + }); + + for (int index = 0; index < 1024; index++) { + assertTrue("chunk " + index, cache.isChunkCached(index & 31, index >> 5)); + } + assertTrue(cache.isRegionCached(0, 0)); + } + + @Test + public void aRegionIsOnlyReportedCachedOnceEveryChunkInItLanded() throws Exception { + PregenCache cache = PregenCache.create(directory); + AtomicInteger seenComplete = new AtomicInteger(); + runAll(thread -> () -> { + for (int index = thread; index < 1023; index += THREADS) { + cache.cacheChunk(index & 31, index >> 5); + if (cache.isRegionCached(0, 0)) { + seenComplete.incrementAndGet(); + } + } + return null; + }); + + assertEquals(0, seenComplete.get()); + assertFalse(cache.isRegionCached(0, 0)); + cache.cacheChunk(1023 & 31, 1023 >> 5); + assertTrue(cache.isRegionCached(0, 0)); + } + + @Test + public void concurrentReadsAndWritesNeverInventACachedChunk() throws Exception { + PregenCache cache = PregenCache.create(directory); + AtomicInteger falsePositives = new AtomicInteger(); + runAll(thread -> () -> { + if (thread % 2 == 0) { + for (int index = 0; index < 2048; index++) { + cache.cacheChunk(index, 0); + } + return null; + } + + for (int index = 0; index < 2048; index++) { + if (cache.isChunkCached(index, 7000)) { + falsePositives.incrementAndGet(); + } + } + return null; + }); + + assertEquals(0, falsePositives.get()); + } + + @Test + public void aConcurrentWriteAndFlushRoundTripsThroughDiskWithoutLosingChunks() throws Exception { + PregenCache cache = PregenCache.create(directory); + runAll(thread -> () -> { + for (int index = 0; index < 256; index++) { + cache.cacheChunk((thread * 256) + index, 4); + if (index % 64 == 0) { + cache.write(); + } + } + return null; + }); + cache.write(); + + PregenCache reopened = PregenCache.create(directory); + for (int thread = 0; thread < THREADS; thread++) { + for (int index = 0; index < 256; index++) { + int x = (thread * 256) + index; + assertTrue("chunk " + x + " lost across a flush", reopened.isChunkCached(x, 4)); + } + } + } + + @Test + public void plateEvictionUnderConcurrencyStillPersistsEveryChunk() throws Exception { + PregenCacheImpl cache = new PregenCacheImpl(directory, 1); + int platesPerThread = 6; + runAll(thread -> () -> { + for (int plate = 0; plate < platesPerThread; plate++) { + cache.cacheChunk(plate * 1024, thread); + } + return null; + }); + cache.write(); + + PregenCache reopened = PregenCache.create(directory); + for (int thread = 0; thread < THREADS; thread++) { + for (int plate = 0; plate < platesPerThread; plate++) { + assertTrue("chunk " + (plate * 1024) + "," + thread + " lost to eviction", + reopened.isChunkCached(plate * 1024, thread)); + } + } + } + + private void runAll(ThreadTask task) throws Exception { + CyclicBarrier barrier = new CyclicBarrier(THREADS); + List> futures = new ArrayList<>(THREADS); + for (int thread = 0; thread < THREADS; thread++) { + Callable body = task.forThread(thread); + futures.add(pool.submit(() -> { + barrier.await(30, TimeUnit.SECONDS); + return body.call(); + })); + } + + for (Future future : futures) { + future.get(60, TimeUnit.SECONDS); + } + } + + private static void deleteRecursively(File file) { + File[] children = file.listFiles(); + if (children != null) { + for (File child : children) { + deleteRecursively(child); + } + } + file.delete(); + } + + private interface ThreadTask { + Callable forThread(int thread); + } +} diff --git a/core/src/test/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenMethodConcurrencyCapTest.java b/core/src/test/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenMethodConcurrencyCapTest.java index a95d0209e..9a065a227 100644 --- a/core/src/test/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenMethodConcurrencyCapTest.java +++ b/core/src/test/java/art/arcane/iris/core/pregenerator/methods/AsyncPregenMethodConcurrencyCapTest.java @@ -1,17 +1,18 @@ package art.arcane.iris.core.pregenerator.methods; +import art.arcane.iris.core.pregenerator.PregenAdmissionGate; import org.junit.Test; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; -import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; @@ -101,13 +102,13 @@ public class AsyncPregenMethodConcurrencyCapTest { @Test public void closeDrainWaitsPastWarningIntervalsUntilEveryPermitReturns() throws Exception { - Semaphore semaphore = new Semaphore(0); + PregenAdmissionGate gate = new PregenAdmissionGate(2, 5L, System::currentTimeMillis); + assertNotNull(gate.admit(() -> false, () -> false)); + assertNotNull(gate.admit(() -> false, () -> false)); AtomicInteger warnings = new AtomicInteger(); ExecutorService executor = Executors.newSingleThreadExecutor(); try { - Future drain = executor.submit(() -> AsyncPregenMethod.awaitDrain( - semaphore, - 2, + Future drain = executor.submit(() -> gate.awaitDrain( 10L, TimeUnit.MILLISECONDS, warnings::incrementAndGet @@ -116,11 +117,12 @@ public class AsyncPregenMethodConcurrencyCapTest { while (warnings.get() == 0) { Thread.onSpinWait(); } - semaphore.release(2); + gate.release(); + gate.release(); assertFalse(drain.get(1L, TimeUnit.SECONDS)); assertTrue(warnings.get() > 0); - assertEquals(0, semaphore.availablePermits()); + assertEquals(2, gate.availablePermits()); } finally { executor.shutdownNow(); }