From baf64fb9d120979b0d8163c0b7ea26a4a9c49d8e Mon Sep 17 00:00:00 2001 From: Brian Neumann-Fopiano Date: Tue, 13 Oct 2026 14:47:41 -0400 Subject: [PATCH] Shared pending terrain samples - Overlapping terrain batches and point requests share pending coordinate samples - Failed terrain loads release reservations for retry - Provisional worker results mark every caller and stay out of the shared cache - Concurrency tests cover overlap, recursion, interruption, failure and single-worker batching --- .../hydrology/HydrologyBasisCache.java | 193 ++++++++- .../hydrology/HydrologyFootprintCompiler.java | 9 +- .../hydrology/HydrologyForkJoin.java | 11 +- .../hydrology/HydrologyBasisCacheTest.java | 366 +++++++++++++++++- .../hydrology/HydrologyForkJoinTest.java | 39 ++ 5 files changed, 581 insertions(+), 37 deletions(-) diff --git a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyBasisCache.java b/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyBasisCache.java index 5cbc7e2aa..884d60b8d 100644 --- a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyBasisCache.java +++ b/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyBasisCache.java @@ -1,8 +1,15 @@ package art.arcane.iris.generation.hydrology; +import art.arcane.iris.generation.stream.ProvisionalSampling; import art.arcane.volmlib.util.cache.CacheKey; import it.unimi.dsi.fastutil.longs.Long2ObjectLinkedOpenHashMap; import it.unimi.dsi.fastutil.longs.Long2ObjectMap; +import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; + +import java.util.Objects; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.RejectedExecutionException; final class HydrologyBasisCache { private static final int MAXIMUM_ENTRIES = 32_768; @@ -39,23 +46,52 @@ final class HydrologyBasisCache { seed(samples, full); } - boolean findWithoutSlope(long packed, HydrologyTerrainSample[] samples, int index) { - Stripe stripe = stripe(withoutSlope, packed); - synchronized (stripe) { - samples[index] = stripe.samples.getAndMoveToFirst(packed); - return samples[index] != null || stripe.samples.containsKey(packed); + HydrologyTerrainSample[] sampleWithoutSlopeBatch(long[] coordinates, int count, + HydrologyNaturalTerrainSampler sampler) { + Objects.checkFromIndexSize(0, count, coordinates.length); + if (count > HydrologyTerrainSampler.MAXIMUM_BATCH_SIZE) { + throw new IllegalArgumentException("Terrain basis batch exceeds its sample limit"); } - } - - HydrologyTerrainSample rememberWithoutSlope(long packed, HydrologyTerrainSample sample) { - Stripe stripe = stripe(withoutSlope, packed); - synchronized (stripe) { - if (stripe.samples.containsKey(packed)) { - return stripe.samples.getAndMoveToFirst(packed); + HydrologyTerrainSample[] samples = new HydrologyTerrainSample[count]; + Pending[] pending = new Pending[count]; + long[] ownedCoordinates = new long[count]; + Pending[] owned = new Pending[count]; + int ownedCount = 0; + try { + for (int index = 0; index < count; index++) { + long packed = coordinates[index]; + Stripe stripe = stripe(withoutSlope, packed); + synchronized (stripe) { + HydrologyTerrainSample cached = stripe.samples.getAndMoveToFirst(packed); + if (cached != null || stripe.samples.containsKey(packed)) { + samples[index] = cached; + continue; + } + Pending loading = stripe.pending.get(packed); + if (loading == null) { + loading = new Pending(); + stripe.pending.put(packed, loading); + ownedCoordinates[ownedCount] = packed; + owned[ownedCount++] = loading; + } + pending[index] = loading; + } } - stripe.put(packed, sample); - return sample; + if (ownedCount > 0) { + loadBatch(ownedCoordinates, owned, ownedCount, sampler); + } + } catch (RuntimeException | Error failure) { + for (int index = 0; index < ownedCount; index++) { + fail(withoutSlope, ownedCoordinates[index], owned[index], failure); + } + throw failure; } + for (int index = 0; index < count; index++) { + if (pending[index] != null) { + samples[index] = pending[index].await(); + } + } + return samples; } void seedWithoutSlope(Long2ObjectMap samples) { @@ -64,15 +100,78 @@ final class HydrologyBasisCache { private HydrologyTerrainSample sample(int x, int z, HydrologyNaturalTerrainSampler sampler, boolean omitSlope) { long packed = RiverFootprint.pack(x, z); - Stripe stripe = stripe(omitSlope ? withoutSlope : full, packed); + Stripe[] stripes = omitSlope ? withoutSlope : full; + Stripe stripe = stripe(stripes, packed); + Pending loading; + boolean owner = false; synchronized (stripe) { HydrologyTerrainSample cached = stripe.samples.getAndMoveToFirst(packed); if (cached != null || stripe.samples.containsKey(packed)) { return cached; } - HydrologyTerrainSample sampled = omitSlope ? sampler.sampleBasisWithoutSlope(x, z) : sampler.sampleBasis(x, z); - stripe.put(packed, sampled); - return sampled; + loading = stripe.pending.get(packed); + if (loading == null) { + loading = new Pending(); + stripe.pending.put(packed, loading); + owner = true; + } + } + if (owner) { + try { + HydrologyTerrainSample sampled = ProvisionalSampling.memoizable(() -> omitSlope + ? sampler.sampleBasisWithoutSlope(x, z) : sampler.sampleBasis(x, z)); + complete(stripes, packed, loading, sampled, false); + } catch (ProvisionalSampling.Unmemoizable provisional) { + complete(stripes, packed, loading, provisional.value(), true); + } catch (RuntimeException | Error failure) { + fail(stripes, packed, loading, failure); + throw failure; + } + } + return loading.await(); + } + + private void loadBatch(long[] coordinates, Pending[] pending, int count, HydrologyNaturalTerrainSampler sampler) { + HydrologyTerrainSample[] samples; + boolean provisional = false; + try { + samples = ProvisionalSampling.memoizable(() -> sampler.sampleBasisWithoutSlopeBatch(coordinates, count)); + } catch (ProvisionalSampling.Unmemoizable value) { + samples = value.value(); + provisional = true; + } + if (samples.length < count) { + throw new IllegalStateException("Terrain basis batch returned too few samples"); + } + for (int index = 0; index < count; index++) { + complete(withoutSlope, coordinates[index], pending[index], samples[index], provisional); + } + } + + private void complete(Stripe[] stripes, long packed, Pending pending, HydrologyTerrainSample sample, + boolean provisional) { + Stripe stripe = stripe(stripes, packed); + synchronized (stripe) { + if (stripe.samples.containsKey(packed)) { + sample = stripe.samples.getAndMoveToFirst(packed); + provisional = false; + } else if (!provisional) { + stripe.put(packed, sample); + } + stripe.pending.remove(packed); + pending.sample = sample; + pending.provisional = provisional; + pending.complete.countDown(); + } + } + + private void fail(Stripe[] stripes, long packed, Pending pending, Throwable failure) { + Stripe stripe = stripe(stripes, packed); + synchronized (stripe) { + if (stripe.pending.remove(packed, pending)) { + pending.failure = failure; + pending.complete.countDown(); + } } } @@ -92,6 +191,7 @@ final class HydrologyBasisCache { private static final class Stripe { private final int maximumEntries; private final Long2ObjectLinkedOpenHashMap samples; + private final Long2ObjectOpenHashMap pending = new Long2ObjectOpenHashMap<>(); private Stripe(int maximumEntries) { this.maximumEntries = maximumEntries; @@ -105,4 +205,61 @@ final class HydrologyBasisCache { samples.putAndMoveToFirst(packed, value); } } + + private static final class Pending implements ForkJoinPool.ManagedBlocker { + private final Thread owner = Thread.currentThread(); + private final CountDownLatch complete = new CountDownLatch(1); + private HydrologyTerrainSample sample; + private Throwable failure; + private boolean provisional; + + private HydrologyTerrainSample await() { + if (!isReleasable() && owner == Thread.currentThread()) { + throw new IllegalStateException("Recursive terrain basis sampling"); + } + boolean interrupted = false; + boolean managed = true; + try { + while (!isReleasable()) { + try { + if (managed) { + ForkJoinPool.managedBlock(this); + } else { + complete.await(); + } + } catch (InterruptedException interruption) { + interrupted = true; + managed = false; + } catch (RejectedExecutionException unavailable) { + managed = false; + } + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + if (failure instanceof Error error) { + throw error; + } + if (failure instanceof RuntimeException runtime) { + throw runtime; + } + if (provisional) { + ProvisionalSampling.mark(); + } + return sample; + } + + @Override + public boolean block() throws InterruptedException { + complete.await(); + return true; + } + + @Override + public boolean isReleasable() { + return complete.getCount() == 0; + } + } } diff --git a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyFootprintCompiler.java b/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyFootprintCompiler.java index e3ef3caec..ba9047cfa 100644 --- a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyFootprintCompiler.java +++ b/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyFootprintCompiler.java @@ -946,8 +946,6 @@ final class HydrologyFootprintCompiler { HydrologyTerrainSample cached = terrainBases.get(key); if (cached != null) { samples[index] = cached; - } else if (sharedBasis != null && sharedBasis.findWithoutSlope(key, samples, index)) { - continue; } else if (naturalSampler == null) { samples[index] = sampleTerrainBasis(RiverFootprint.unpackX(key), RiverFootprint.unpackZ(key)); } else { @@ -956,10 +954,11 @@ final class HydrologyFootprintCompiler { } } if (missingCount > 0) { - HydrologyTerrainSample[] loaded = naturalSampler.sampleBasisWithoutSlopeBatch(missing, missingCount); + HydrologyTerrainSample[] loaded = sharedBasis == null + ? naturalSampler.sampleBasisWithoutSlopeBatch(missing, missingCount) + : sharedBasis.sampleWithoutSlopeBatch(missing, missingCount, naturalSampler); for (int index = 0; index < missingCount; index++) { - samples[positions[index]] = sharedBasis == null ? loaded[index] - : sharedBasis.rememberWithoutSlope(missing[index], loaded[index]); + samples[positions[index]] = loaded[index]; } } for (int index = 0; index < count; index++) { diff --git a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyForkJoin.java b/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyForkJoin.java index 533fb7845..b26c1148b 100644 --- a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyForkJoin.java +++ b/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyForkJoin.java @@ -1,5 +1,7 @@ package art.arcane.iris.generation.hydrology; +import art.arcane.iris.generation.stream.ProvisionalSampling; + import java.util.ArrayList; import java.util.List; import java.util.Objects; @@ -91,6 +93,7 @@ public final class HydrologyForkJoin { private final CountDownLatch complete = new CountDownLatch(1); private T result; private Throwable failure; + private boolean provisional; Task(Callable work) { this.work = Objects.requireNonNull(work); @@ -136,6 +139,9 @@ public final class HydrologyForkJoin { if (failure != null) { throw unwrap(failure); } + if (provisional) { + ProvisionalSampling.mark(); + } return result; } @@ -156,7 +162,10 @@ public final class HydrologyForkJoin { return; } try { - result = call(work); + result = ProvisionalSampling.memoizable(() -> call(work)); + } catch (ProvisionalSampling.Unmemoizable value) { + result = value.value(); + provisional = true; } catch (RuntimeException | Error failed) { failure = failed; } finally { diff --git a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyBasisCacheTest.java b/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyBasisCacheTest.java index 1d0c01696..d6de76886 100644 --- a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyBasisCacheTest.java +++ b/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyBasisCacheTest.java @@ -1,5 +1,6 @@ package art.arcane.iris.generation.hydrology; +import art.arcane.iris.generation.stream.ProvisionalSampling; import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; import org.junit.Test; @@ -9,29 +10,347 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.ForkJoinPool; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; public class HydrologyBasisCacheTest { @Test - public void batchLookupPreservesNullAndFirstPublishedSample() { + public void batchLookupPreservesNullDuplicatesAndEviction() { + CountingSampler sampler = new CountingSampler(); HydrologyBasisCache cache = new HydrologyBasisCache(1); - HydrologyTerrainSample[] samples = new HydrologyTerrainSample[1]; - long key = RiverFootprint.pack(-17, Integer.MAX_VALUE); - assertTrue(!cache.findWithoutSlope(key, samples, 0)); - assertNull(cache.rememberWithoutSlope(key, null)); - assertTrue(cache.findWithoutSlope(key, samples, 0)); - assertNull(samples[0]); - HydrologyTerrainSample later = HydrologyTerrainSample.openLand(73, 0D, "land"); - assertNull(cache.rememberWithoutSlope(key, later)); - cache.rememberWithoutSlope(key + 1, later); - assertTrue(!cache.findWithoutSlope(key, samples, 0)); - assertEquals(later, cache.rememberWithoutSlope(key + 1, null)); + long key = RiverFootprint.pack(Integer.MIN_VALUE, Integer.MAX_VALUE); + assertArrayEquals(new HydrologyTerrainSample[2], + cache.sampleWithoutSlopeBatch(new long[]{key, key}, 2, sampler)); + assertNull(cache.sampleWithoutSlope(Integer.MIN_VALUE, Integer.MAX_VALUE, sampler)); + assertEquals(1, sampler.slopeFreeCalls.get()); + long later = RiverFootprint.pack(73, -17); + assertEquals(87, cache.sampleWithoutSlopeBatch(new long[]{later}, 1, sampler)[0].naturalHeight()); + assertNull(cache.sampleWithoutSlopeBatch(new long[]{key}, 1, sampler)[0]); + assertEquals(3, sampler.slopeFreeCalls.get()); + } + + @Test(timeout = 10_000) + public void overlappingBatchesDispatchOnlyTheirOwnedCoordinates() throws Exception { + HydrologyBasisCache cache = new HydrologyBasisCache(); + CountDownLatch firstStarted = new CountDownLatch(1); + CountDownLatch secondStarted = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger batchCalls = new AtomicInteger(); + CountingSampler sampler = new CountingSampler() { + @Override + public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) { + if (batchCalls.getAndIncrement() == 0) { + assertEquals(2, count); + firstStarted.countDown(); + } else { + assertEquals(1, count); + assertEquals(RiverFootprint.pack(3, 0), coordinates[0]); + secondStarted.countDown(); + } + await(release); + return super.sampleBasisWithoutSlopeBatch(coordinates, count); + } + }; + ExecutorService workers = Executors.newFixedThreadPool(2); + try { + Future first = workers.submit(() -> cache.sampleWithoutSlopeBatch( + new long[]{RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)}, 2, sampler)); + assertTrue(firstStarted.await(5, TimeUnit.SECONDS)); + Future second = workers.submit(() -> cache.sampleWithoutSlopeBatch( + new long[]{RiverFootprint.pack(2, 0), RiverFootprint.pack(3, 0)}, 2, sampler)); + assertTrue(secondStarted.await(5, TimeUnit.SECONDS)); + release.countDown(); + assertSame(first.get(5, TimeUnit.SECONDS)[1], second.get(5, TimeUnit.SECONDS)[0]); + assertEquals(3, sampler.slopeFreeCalls.get()); + assertEquals(2, batchCalls.get()); + } finally { + release.countDown(); + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test(timeout = 10_000) + public void pointAndBatchSharePendingSamplesWithoutHoldingStripeLocks() throws Exception { + HydrologyBasisCache cache = new HydrologyBasisCache(1); + CountDownLatch pointStarted = new CountDownLatch(1); + CountDownLatch batchStarted = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountingSampler sampler = new CountingSampler() { + @Override + public HydrologyTerrainSample sampleBasisWithoutSlope(int x, int z) { + if (x == 1) { + pointStarted.countDown(); + await(release); + } + return super.sampleBasisWithoutSlope(x, z); + } + + @Override + public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) { + assertEquals(1, count); + assertEquals(RiverFootprint.pack(2, 0), coordinates[0]); + batchStarted.countDown(); + return super.sampleBasisWithoutSlopeBatch(coordinates, count); + } + }; + ExecutorService workers = Executors.newFixedThreadPool(2); + try { + Future point = workers.submit(() -> cache.sampleWithoutSlope(1, 0, sampler)); + assertTrue(pointStarted.await(5, TimeUnit.SECONDS)); + Future batch = workers.submit(() -> cache.sampleWithoutSlopeBatch( + new long[]{RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)}, 2, sampler)); + assertTrue(batchStarted.await(5, TimeUnit.SECONDS)); + release.countDown(); + assertSame(point.get(5, TimeUnit.SECONDS), batch.get(5, TimeUnit.SECONDS)[0]); + assertEquals(2, sampler.slopeFreeCalls.get()); + } finally { + release.countDown(); + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test(timeout = 10_000) + public void nestedBatchesFinishOnSingleWorkerPool() throws Exception { + HydrologyBasisCache cache = new HydrologyBasisCache(); + ForkJoinPool workers = new ForkJoinPool(1); + CountingSampler sampler = new CountingSampler() { + @Override + public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) { + List values = HydrologyForkJoin.invokeAll(List.of( + () -> sampleBasisWithoutSlope(RiverFootprint.unpackX(coordinates[0]), 0), + () -> sampleBasisWithoutSlope(RiverFootprint.unpackX(coordinates[1]), 0)), workers); + return values.toArray(HydrologyTerrainSample[]::new); + } + }; + try { + workers.submit(() -> HydrologyForkJoin.invokeAll(List.of( + () -> cache.sampleWithoutSlopeBatch(new long[]{RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)}, 2, sampler), + () -> cache.sampleWithoutSlopeBatch(new long[]{RiverFootprint.pack(2, 0), RiverFootprint.pack(1, 0)}, 2, sampler)), workers)) + .get(5, TimeUnit.SECONDS); + assertEquals(2, sampler.slopeFreeCalls.get()); + } finally { + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + public void failedMalformedAndRecursiveBatchesReleaseEveryReservation() { + HydrologyBasisCache cache = new HydrologyBasisCache(); + long[] coordinates = {RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)}; + CountingSampler failed = new CountingSampler() { + @Override + public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] keys, int count) { + throw new IllegalArgumentException("failure"); + } + }; + assertThrows(IllegalArgumentException.class, () -> cache.sampleWithoutSlopeBatch(coordinates, 2, failed)); + CountingSampler malformed = new CountingSampler() { + @Override + public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] keys, int count) { + return new HydrologyTerrainSample[1]; + } + }; + assertThrows(IllegalStateException.class, () -> cache.sampleWithoutSlopeBatch(coordinates, 2, malformed)); + CountingSampler absent = new CountingSampler() { + @Override + public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] keys, int count) { + return null; + } + }; + assertThrows(NullPointerException.class, () -> cache.sampleWithoutSlopeBatch(coordinates, 2, absent)); + CountingSampler recursive = new CountingSampler() { + @Override + public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] keys, int count) { + cache.sampleWithoutSlope(2, 0, this); + throw new AssertionError("Recursive sampling must not wait on itself"); + } + }; + assertThrows(IllegalStateException.class, () -> cache.sampleWithoutSlopeBatch(coordinates, 2, recursive)); + CountingSampler successful = new CountingSampler(); + assertEquals(2, cache.sampleWithoutSlopeBatch(coordinates, 2, successful).length); + assertEquals(2, successful.slopeFreeCalls.get()); + } + + @Test(timeout = 10_000) + public void interruptedPointWaiterReceivesBatchResultAndRestoresInterrupt() throws Exception { + HydrologyBasisCache cache = new HydrologyBasisCache(); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicReference waiting = new AtomicReference<>(); + CountingSampler sampler = new CountingSampler() { + @Override + public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) { + started.countDown(); + await(release); + return super.sampleBasisWithoutSlopeBatch(coordinates, count); + } + }; + ExecutorService workers = Executors.newFixedThreadPool(2); + try { + Future batch = workers.submit(() -> cache.sampleWithoutSlopeBatch( + new long[]{RiverFootprint.pack(1, 0)}, 1, sampler)); + assertTrue(started.await(5, TimeUnit.SECONDS)); + Future point = workers.submit(() -> { + waiting.set(Thread.currentThread()); + HydrologyTerrainSample sample = cache.sampleWithoutSlope(1, 0, sampler); + assertTrue(Thread.interrupted()); + return sample; + }); + awaitWaiting(waiting); + waiting.get().interrupt(); + release.countDown(); + assertSame(batch.get(5, TimeUnit.SECONDS)[0], point.get(5, TimeUnit.SECONDS)); + assertEquals(1, sampler.slopeFreeCalls.get()); + } finally { + release.countDown(); + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test(timeout = 10_000) + public void batchFailureReachesPointWaiterAndAllowsAllCoordinatesToRetry() throws Exception { + HydrologyBasisCache cache = new HydrologyBasisCache(); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicReference waiting = new AtomicReference<>(); + IllegalArgumentException failure = new IllegalArgumentException("retry"); + CountingSampler sampler = new CountingSampler() { + @Override + public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) { + started.countDown(); + await(release); + throw failure; + } + }; + long[] coordinates = {RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)}; + ExecutorService workers = Executors.newFixedThreadPool(2); + try { + Future batch = workers.submit(() -> assertThrows(IllegalArgumentException.class, + () -> cache.sampleWithoutSlopeBatch(coordinates, 2, sampler))); + assertTrue(started.await(5, TimeUnit.SECONDS)); + Future point = workers.submit(() -> { + waiting.set(Thread.currentThread()); + return assertThrows(IllegalArgumentException.class, () -> cache.sampleWithoutSlope(2, 0, sampler)); + }); + awaitWaiting(waiting); + release.countDown(); + assertSame(failure, batch.get(5, TimeUnit.SECONDS)); + assertSame(failure, point.get(5, TimeUnit.SECONDS)); + CountingSampler successful = new CountingSampler(); + assertEquals(2, cache.sampleWithoutSlopeBatch(coordinates, 2, successful).length); + assertEquals(2, successful.slopeFreeCalls.get()); + } finally { + release.countDown(); + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + public void provisionalResultsAreNotRetained() { + HydrologyBasisCache cache = new HydrologyBasisCache(); + CountingSampler sampler = new CountingSampler() { + @Override + public HydrologyTerrainSample sampleBasisWithoutSlope(int x, int z) { + ProvisionalSampling.mark(); + return super.sampleBasisWithoutSlope(x, z); + } + }; + long[] coordinates = {RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)}; + cache.sampleWithoutSlopeBatch(coordinates, 2, sampler); + cache.sampleWithoutSlopeBatch(coordinates, 2, sampler); + cache.sampleWithoutSlope(1, 0, sampler); + cache.sampleWithoutSlope(1, 0, sampler); + assertEquals(6, sampler.slopeFreeCalls.get()); + } + + @Test(timeout = 10_000) + public void provisionalWorkerResultMarksSharedCacheWaitersAndIsNotRetained() throws Exception { + HydrologyBasisCache cache = new HydrologyBasisCache(); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicReference waiting = new AtomicReference<>(); + ExecutorService providerWorkers = Executors.newFixedThreadPool(2); + CountingSampler sampler = new CountingSampler() { + @Override + public HydrologyTerrainSample[] sampleBasisWithoutSlopeBatch(long[] coordinates, int count) { + List values = HydrologyForkJoin.invokeAll(List.of(() -> { + started.countDown(); + await(release); + ProvisionalSampling.mark(); + return sampleBasisWithoutSlope(RiverFootprint.unpackX(coordinates[0]), 0); + }, () -> sampleBasisWithoutSlope(RiverFootprint.unpackX(coordinates[1]), 0)), providerWorkers); + return values.toArray(HydrologyTerrainSample[]::new); + } + }; + ExecutorService callers = Executors.newFixedThreadPool(2); + long[] coordinates = {RiverFootprint.pack(1, 0), RiverFootprint.pack(2, 0)}; + try { + Future batch = callers.submit(() -> cache.sampleWithoutSlopeBatch(coordinates, 2, sampler)); + assertTrue(started.await(5, TimeUnit.SECONDS)); + Future point = callers.submit(() -> { + waiting.set(Thread.currentThread()); + return assertThrows(ProvisionalSampling.Unmemoizable.class, + () -> ProvisionalSampling.memoizable(() -> cache.sampleWithoutSlope(1, 0, sampler))); + }); + awaitWaiting(waiting); + release.countDown(); + assertSame(batch.get(5, TimeUnit.SECONDS)[0], point.get(5, TimeUnit.SECONDS).value()); + cache.sampleWithoutSlopeBatch(coordinates, 2, sampler); + assertEquals(4, sampler.slopeFreeCalls.get()); + } finally { + release.countDown(); + callers.shutdownNow(); + providerWorkers.shutdownNow(); + assertTrue(callers.awaitTermination(5, TimeUnit.SECONDS)); + assertTrue(providerWorkers.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test(timeout = 10_000) + public void seedWinsPendingSample() throws Exception { + HydrologyBasisCache cache = new HydrologyBasisCache(); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountingSampler sampler = new CountingSampler() { + @Override + public HydrologyTerrainSample sampleBasisWithoutSlope(int x, int z) { + started.countDown(); + await(release); + return super.sampleBasisWithoutSlope(x, z); + } + }; + ExecutorService workers = Executors.newSingleThreadExecutor(); + try { + Future result = workers.submit(() -> cache.sampleWithoutSlopeBatch( + new long[]{RiverFootprint.pack(1, 0)}, 1, sampler)); + assertTrue(started.await(5, TimeUnit.SECONDS)); + HydrologyTerrainSample seeded = HydrologyTerrainSample.openLand(73, 0D, "seed"); + Long2ObjectOpenHashMap seeds = new Long2ObjectOpenHashMap<>(); + seeds.put(RiverFootprint.pack(1, 0), seeded); + cache.seedWithoutSlope(seeds); + release.countDown(); + assertSame(seeded, result.get(5, TimeUnit.SECONDS)[0]); + assertSame(seeded, cache.sampleWithoutSlope(1, 0, sampler)); + } finally { + release.countDown(); + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + } } @Test @@ -122,7 +441,28 @@ public class HydrologyBasisCacheTest { } } - private static final class CountingSampler implements HydrologyNaturalTerrainSampler { + private static void awaitWaiting(AtomicReference thread) { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (System.nanoTime() < deadline) { + Thread current = thread.get(); + if (current != null && current.getState() == Thread.State.WAITING) { + return; + } + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1)); + } + throw new AssertionError("Cache waiter did not wait for its owner"); + } + + private static void await(CountDownLatch latch) { + try { + assertTrue(latch.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException interruption) { + Thread.currentThread().interrupt(); + throw new AssertionError(interruption); + } + } + + private static class CountingSampler implements HydrologyNaturalTerrainSampler { private final AtomicInteger fullCalls = new AtomicInteger(); private final AtomicInteger slopeFreeCalls = new AtomicInteger(); private final AtomicInteger failures = new AtomicInteger(); diff --git a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyForkJoinTest.java b/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyForkJoinTest.java index 77a6e3c19..610889807 100644 --- a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyForkJoinTest.java +++ b/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyForkJoinTest.java @@ -1,5 +1,6 @@ package art.arcane.iris.generation.hydrology; +import art.arcane.iris.generation.stream.ProvisionalSampling; import org.junit.Test; import java.util.ArrayList; @@ -23,6 +24,44 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; public class HydrologyForkJoinTest { + @Test(timeout = 10_000) + public void workerProvisionalAnswersReachEveryAwaitingCaller() throws Exception { + HydrologyForkJoin.Task task = new HydrologyForkJoin.Task<>(() -> { + ProvisionalSampling.mark(); + return 73; + }); + ExecutorService workers = Executors.newFixedThreadPool(2); + try { + workers.submit(task).get(5, TimeUnit.SECONDS); + for (int attempt = 0; attempt < 2; attempt++) { + ProvisionalSampling.Unmemoizable result = workers.submit(() -> assertThrows( + ProvisionalSampling.Unmemoizable.class, + () -> ProvisionalSampling.memoizable(task::await))).get(5, TimeUnit.SECONDS); + assertEquals(Integer.valueOf(73), result.value()); + } + } finally { + workers.shutdownNow(); + assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test(timeout = 10_000) + public void provisionalWorkerMarksTheBatchCaller() { + ExecutorService workers = Executors.newFixedThreadPool(2); + try { + ProvisionalSampling.Unmemoizable result = assertThrows(ProvisionalSampling.Unmemoizable.class, + () -> ProvisionalSampling.memoizable(() -> HydrologyForkJoin.invokeAll(List.of( + () -> 1, + () -> { + ProvisionalSampling.mark(); + return 2; + }), workers))); + assertEquals(List.of(1, 2), result.value()); + } finally { + workers.shutdownNow(); + } + } + @Test public void blockedGridBatchNeverHelpsUnrelatedOwnerWork() throws Exception { ForkJoinPool pool = new ForkJoinPool(2, ForkJoinPool.defaultForkJoinWorkerThreadFactory,