Shared concurrent raster builds

- Made concurrent demands for one chunk wait for a single water raster build
- Added a test that a second demand reuses the build already running

core/src/main/java/art/arcane/iris/generation/hydrology/engine/HydrologyRuntime.java

@@ -72,6 +72,7 @@ public final class HydrologyRuntime {
     private final RasterLru rasters;
     private final ThreadLocal<LastRaster> last;
     private final Set<Long> queued;
+    private final ConcurrentHashMap<Long, CompletableFuture<ChunkWaterRaster>> building;
     private final RasterFailures failures;
     private final RasterFailures windowPreviewFailures;
     private final RasterFailures regionPreviewFailures;
@@ -94,6 +95,7 @@ public final class HydrologyRuntime {
         this.rasters = null;
         this.last = null;
         this.queued = null;
+        this.building = null;
         this.failures = null;
         this.windowPreviewFailures = null;
         this.regionPreviewFailures = null;
@@ -120,6 +122,7 @@ public final class HydrologyRuntime {
         this.rasters = new RasterLru();
         this.last = ThreadLocal.withInitial(LastRaster::new);
         this.queued = ConcurrentHashMap.newKeySet();
+        this.building = new ConcurrentHashMap<>();
         this.failures = new RasterFailures(ticker);
         this.windowPreviewFailures = new RasterFailures(ticker);
         this.regionPreviewFailures = new RasterFailures(ticker);
@@ -234,7 +237,7 @@ public final class HydrologyRuntime {
         requireOpen();
         long key = key(chunkX, chunkZ);
         ChunkWaterRaster known = remembered(key, slot);
-        return known != null ? known : remember(key, planned(chunkX, chunkZ, key));
+        return known != null ? known : build(chunkX, chunkZ, key, slot);
     }
 
     public ChunkWaterRaster rasterSnapshot(int chunkX, int chunkZ) {
@@ -501,6 +504,41 @@ public final class HydrologyRuntime {
         }
     }
 
+    private ChunkWaterRaster build(int chunkX, int chunkZ, long key, LastRaster slot) {
+        CompletableFuture<ChunkWaterRaster> claim = new CompletableFuture<>();
+        CompletableFuture<ChunkWaterRaster> running = building.putIfAbsent(key, claim);
+        if (running != null) {
+            ChunkWaterRaster built = built(running);
+            slot.remember(key, built);
+            return built;
+        }
+        try {
+            ChunkWaterRaster known = remembered(key, slot);
+            ChunkWaterRaster raster = known != null ? known : remember(key, planned(chunkX, chunkZ, key));
+            claim.complete(raster);
+            return raster;
+        } catch (RuntimeException | Error failure) {
+            claim.completeExceptionally(failure);
+            throw failure;
+        } finally {
+            building.remove(key, claim);
+        }
+    }
+
+    private static ChunkWaterRaster built(CompletableFuture<ChunkWaterRaster> running) {
+        try {
+            return running.join();
+        } catch (CompletionException failure) {
+            if (failure.getCause() instanceof RuntimeException cause) {
+                throw cause;
+            }
+            if (failure.getCause() instanceof Error cause) {
+                throw cause;
+            }
+            throw failure;
+        }
+    }
+
     private ChunkWaterRaster planned(int chunkX, int chunkZ, long chunk) {
         try {
             List<WindowKey> windowKeys = neighbourhood(chunkX, chunkZ);

core/src/test/java/art/arcane/iris/generation/hydrology/engine/HydrologyRuntimeTest.java

@@ -215,6 +215,34 @@ public class HydrologyRuntimeTest {
         assertEquals(planned + 1L, runtime.previewRevision());
     }
 
+    @Test(timeout = 180_000L)
+    public void concurrentDemandsShareOneRasterBuild() throws Exception {
+        FakePlanners planners = new FakePlanners(settings(0, false));
+        HydrologyRuntime runtime = runtime(planners, () -> false);
+        runtime.prepareWindowsAround(0, 0, 0, 0, 1).get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+        CountDownLatch gate = new CountDownLatch(1);
+        planners.naturalGate = gate;
+        AtomicReference<ChunkWaterRaster> first = new AtomicReference<>();
+        AtomicReference<ChunkWaterRaster> second = new AtomicReference<>();
+        Thread leader = new Thread(() -> first.set(runtime.raster(0, 0)));
+        Thread follower = new Thread(() -> second.set(runtime.raster(0, 0)));
+        try {
+            leader.start();
+            await(() -> planners.naturalSamples.get() > 0, "the first demand starts sampling");
+            follower.start();
+            await(() -> follower.getState() == Thread.State.WAITING || follower.getState() == Thread.State.TIMED_WAITING,
+                    "the second demand waits");
+            assertEquals(0, planners.naturalSamplesOn(follower));
+        } finally {
+            gate.countDown();
+            leader.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+            follower.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+        }
+        assertFalse(first.get().provisional());
+        assertSame(first.get(), second.get());
+        assertEquals(0, planners.naturalSamplesOn(follower));
+    }
+
     @Test(timeout = 180_000L)
     public void failedRasterDoesNotAdvancePreviewRevision() throws Exception {
         FakePlanners planners = new FakePlanners(settings(0, false));