Idle-aware fallback trial batches

- Fallback trials budget a measured 128 MB each instead of 1 GB

- Trial batches grow with idle hydrology workers up to the trial limit

core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyCrossTileResolver.java

@@ -20,14 +20,29 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.ForkJoinWorkerThread;
 
 final class HydrologyCrossTileResolver {
     private static final int FALLBACK_TRIAL_PARALLELISM = 16;
+    private static final int MINIMUM_FALLBACK_TRIAL_BATCH = 2;
     private final HydrologyPlanner planner;
+    private final int trialLimit;
 
-    HydrologyCrossTileResolver(HydrologyPlanner planner) {
+    HydrologyCrossTileResolver(HydrologyPlanner planner, int trialLimit) {
+        if (trialLimit < 1) {
+            throw new IllegalArgumentException("Hydrology fallback trial limit must be positive.");
+        }
         this.planner = planner;
+        this.trialLimit = trialLimit;
+    }
+
+    static int fallbackTrialBatch(int parallelism, int idleWorkers, int trialLimit) {
+        if (parallelism <= 2) {
+            return 1;
+        }
+        int workers = Math.clamp(idleWorkers + 1L, MINIMUM_FALLBACK_TRIAL_BATCH, parallelism);
+        return Math.max(1, Math.min(workers, Math.min(trialLimit, FALLBACK_TRIAL_PARALLELISM)));
     }
 
     HydrologyOwnerDraft compileOwnerDraft(
@@ -285,12 +300,18 @@ final class HydrologyCrossTileResolver {
             List<OutletCandidate> outlets,
             int start
     ) {
-        int parallelism = HydrologyPlanningAdmission.effectiveParallelism(
+        int parallelism;
-                Thread.currentThread() instanceof ForkJoinWorkerThread worker
+        int idleWorkers;
-                        ? worker.getPool().getParallelism()
+        if (Thread.currentThread() instanceof ForkJoinWorkerThread worker) {
-                        : IrisPlatforms.isBound() ? MultiBurst.hydrology.parallelism() : 1);
+            ForkJoinPool pool = worker.getPool();
-        int batchSize = parallelism <= 2 ? 1 : Math.max(1, Math.min(HydrologyPlanningAdmission.maximumTrials(),
+            parallelism = HydrologyPlanningAdmission.effectiveParallelism(pool.getParallelism());
-                Math.min(FALLBACK_TRIAL_PARALLELISM, parallelism)));
+            idleWorkers = pool.getParallelism() - pool.getActiveThreadCount();
+        } else {
+            parallelism = HydrologyPlanningAdmission.effectiveParallelism(
+                    IrisPlatforms.isBound() ? MultiBurst.hydrology.parallelism() : 1);
+            idleWorkers = 0;
+        }
+        int batchSize = fallbackTrialBatch(parallelism, idleWorkers, trialLimit);
         int end = Math.min(outlets.size(), start + batchSize);
         ArrayList<Callable<FallbackTrial>> tasks = new ArrayList<>(end - start);
         HydrologyPlanner.PlanningSamples ownerSamples = planner.planningSamples.get();

core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyPlanner.java

@@ -197,7 +197,7 @@ public final class HydrologyPlanner {
                 geometrySampler,
                 settings.seaLevel()
         );
-        this.crossTile = new HydrologyCrossTileResolver(this);
+        this.crossTile = new HydrologyCrossTileResolver(this, HydrologyPlanningAdmission.maximumTrials());
         this.outletPlanner = new HydrologyOutletPlanner(this);
         this.sourcePlanner = new HydrologySourcePlanner(this);
         this.routeGeometry = new HydrologyRouteGeometry(this);

core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyPlanningAdmission.java

@@ -10,6 +10,8 @@ import java.util.function.BooleanSupplier;
 
 final class HydrologyPlanningAdmission {
     private static final long BYTES_PER_ROOT = 2L * 1024L * 1024L * 1024L;
+    private static final long BYTES_PER_TRIAL = 128L * 1024L * 1024L;
+    private static final int MAXIMUM_TRIAL_LIMIT = 16;
     private static final int MAXIMUM_ROOTS = rootLimit(
             Runtime.getRuntime().maxMemory(), Runtime.getRuntime().availableProcessors());
     private static final int MAXIMUM_TRIALS = trialLimit(Runtime.getRuntime().maxMemory(), MAXIMUM_ROOTS);
@@ -48,7 +50,7 @@ final class HydrologyPlanningAdmission {
         if (maximumRoots < 1) {
             throw new IllegalArgumentException("Maximum hydrology roots must be positive.");
         }
-        return (int) Math.clamp(maximumHeap / maximumRoots / (1024L * 1024L * 1024L), 1L, 16L);
+        return (int) Math.clamp(maximumHeap / maximumRoots / BYTES_PER_TRIAL, 1L, MAXIMUM_TRIAL_LIMIT);
     }
 
     static int rootLimit(long maximumHeap, int processors) {

core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyFallbackParallelismTest.java

@@ -36,7 +36,8 @@ public class HydrologyFallbackParallelismTest {
     @Test(timeout = 15000)
     public void publicationRejectionConsumesTheNextCompletedOutletWithoutRebuildingIt() throws Exception {
         int parallelism = HydrologyPlanningAdmission.effectiveParallelism(4);
-        int batchSize = parallelism <= 2 ? 1 : Math.min(parallelism, HydrologyPlanningAdmission.maximumTrials());
+        int batchSize = HydrologyCrossTileResolver.fallbackTrialBatch(parallelism, parallelism - 1,
+                HydrologyPlanningAdmission.maximumTrials());
         AtomicIntegerArray builds = new AtomicIntegerArray(4);
         Fixture fixture = fixture(outlet -> {
             builds.incrementAndGet(outlet);
@@ -77,6 +78,23 @@ public class HydrologyFallbackParallelismTest {
         }
     }
 
+    @Test
+    public void fallbackTrialBatchesGrowIntoIdleWorkersWithinTheTrialLimit() {
+        assertEquals(1, HydrologyCrossTileResolver.fallbackTrialBatch(1, 0, 16));
+        assertEquals(1, HydrologyCrossTileResolver.fallbackTrialBatch(2, 1, 16));
+        assertEquals(2, HydrologyCrossTileResolver.fallbackTrialBatch(12, 0, 16));
+        assertEquals(2, HydrologyCrossTileResolver.fallbackTrialBatch(12, 1, 16));
+        assertEquals(6, HydrologyCrossTileResolver.fallbackTrialBatch(12, 5, 16));
+        assertEquals(12, HydrologyCrossTileResolver.fallbackTrialBatch(12, 11, 16));
+        assertEquals(12, HydrologyCrossTileResolver.fallbackTrialBatch(12, 40, 16));
+        assertEquals(16, HydrologyCrossTileResolver.fallbackTrialBatch(32, 31, 16));
+        assertEquals(16, HydrologyCrossTileResolver.fallbackTrialBatch(32, 31, 64));
+        assertEquals(8, HydrologyCrossTileResolver.fallbackTrialBatch(12, 11, 8));
+        assertEquals(2, HydrologyCrossTileResolver.fallbackTrialBatch(12, 11, 2));
+        assertEquals(1, HydrologyCrossTileResolver.fallbackTrialBatch(12, 11, 1));
+        assertEquals(1, HydrologyCrossTileResolver.fallbackTrialBatch(12, 0, 1));
+    }
+
     @Test(timeout = 15000)
     public void limitedWorkersStopAfterTheFirstAcceptedOutlet() throws Exception {
         int[] poolSizes = Runtime.getRuntime().availableProcessors() <= 2 ? new int[]{1, 2, 16} : new int[]{1, 2};

core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyPlanHashTest.java

@@ -5,6 +5,7 @@ import art.arcane.iris.generation.hydrology.policy.SurfaceRiverPolicy;
 import art.arcane.iris.generation.hydrology.cave.HydrologyCavePlan;
 import org.junit.Test;
 
+import java.lang.reflect.Field;
 import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
@@ -13,6 +14,9 @@ import java.util.Comparator;
 import java.util.List;
 import java.util.HexFormat;
 import java.util.Map;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
+import java.util.function.UnaryOperator;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -28,12 +32,34 @@ public class HydrologyPlanHashTest {
 
     @Test
     public void fullTilePlansAreByteIdentical() {
+        PlanDigest plan = planDigest(planner -> planner);
+        assertTrue("planned courses: " + plan.courses(), plan.courses() > 0);
+        assertTrue("planned columns: " + plan.columns(), plan.columns() > 0);
+        assertEquals(EXPECTED_PLAN_DIGEST, plan.digest());
+    }
+
+    @Test(timeout = 180000)
+    public void fallbackTrialBatchSizeNeverChangesThePlan() throws Exception {
+        for (int trialLimit : new int[]{1, 2, 8, 16}) {
+            ForkJoinPool pool = new ForkJoinPool(16);
+            try {
+                PlanDigest plan = pool.submit(() -> planDigest(planner -> withTrialLimit(planner, trialLimit)))
+                        .get(60L, TimeUnit.SECONDS);
+                assertEquals("trial limit " + trialLimit, EXPECTED_PLAN_DIGEST, plan.digest());
+            } finally {
+                pool.shutdownNow();
+                assertTrue(pool.awaitTermination(5L, TimeUnit.SECONDS));
+            }
+        }
+    }
+
+    private static PlanDigest planDigest(UnaryOperator<HydrologyPlanner> configure) {
         MessageDigest canonical = sha256();
         int courses = 0;
         int columns = 0;
         for (long seed : new long[]{77L, 19L}) {
             HydrologyPlannerSettings settings = planSettings();
-            HydrologyPlanner planner = new HydrologyPlanner(seed, settings, rollingCoast(112));
+            HydrologyPlanner planner = configure.apply(new HydrologyPlanner(seed, settings, rollingCoast(112)));
             for (HydrologyTileKey key : TILES) {
                 HydrologyTile tile = planner.plan(key);
                 courses += tile.courses().size();
@@ -41,9 +67,18 @@ public class HydrologyPlanHashTest {
                 render(tile, canonical);
             }
         }
-        assertTrue("planned courses: " + courses, courses > 0);
+        return new PlanDigest(HexFormat.of().formatHex(canonical.digest()), courses, columns);
-        assertTrue("planned columns: " + columns, columns > 0);
+    }
-        assertEquals(EXPECTED_PLAN_DIGEST, HexFormat.of().formatHex(canonical.digest()));
+
+    private static HydrologyPlanner withTrialLimit(HydrologyPlanner planner, int trialLimit) {
+        try {
+            Field crossTile = HydrologyPlanner.class.getDeclaredField("crossTile");
+            crossTile.setAccessible(true);
+            crossTile.set(planner, new HydrologyCrossTileResolver(planner, trialLimit));
+            return planner;
+        } catch (ReflectiveOperationException failure) {
+            throw new IllegalStateException(failure);
+        }
     }
 
     private static void render(HydrologyTile tile, MessageDigest digest) {
@@ -223,6 +258,9 @@ public class HydrologyPlanHashTest {
         );
     }
 
+    private record PlanDigest(String digest, int courses, int columns) {
+    }
+
     private static final class CanonicalDigest {
         private final MessageDigest digest;
 

core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyPlanningAdmissionTest.java

@@ -50,12 +50,16 @@ public class HydrologyPlanningAdmissionTest {
 
     @Test
     public void trialLimitSharesWorkingMemoryBetweenAdmittedRoots() {
-        long gibibyte = 1024L * 1024L * 1024L;
+        long mebibyte = 1024L * 1024L;
-        assertEquals(1, HydrologyPlanningAdmission.trialLimit(1280L * 1024L * 1024L, 1));
+        long gibibyte = 1024L * mebibyte;
-        assertEquals(2, HydrologyPlanningAdmission.trialLimit(2L * gibibyte, 1));
+        assertEquals(1, HydrologyPlanningAdmission.trialLimit(64L * mebibyte, 1));
-        assertEquals(2, HydrologyPlanningAdmission.trialLimit(4L * gibibyte, 2));
+        assertEquals(1, HydrologyPlanningAdmission.trialLimit(128L * mebibyte, 1));
-        assertEquals(1, HydrologyPlanningAdmission.trialLimit(gibibyte / 2L, 1));
+        assertEquals(4, HydrologyPlanningAdmission.trialLimit(gibibyte / 2L, 1));
-        assertEquals(8, HydrologyPlanningAdmission.trialLimit(64L * gibibyte, 8));
+        assertEquals(4, HydrologyPlanningAdmission.trialLimit(gibibyte, 2));
+        assertEquals(10, HydrologyPlanningAdmission.trialLimit(1280L * mebibyte, 1));
+        assertEquals(16, HydrologyPlanningAdmission.trialLimit(2L * gibibyte, 1));
+        assertEquals(16, HydrologyPlanningAdmission.trialLimit(6L * gibibyte, 3));
+        assertEquals(16, HydrologyPlanningAdmission.trialLimit(16L * gibibyte, 8));
         assertEquals(16, HydrologyPlanningAdmission.trialLimit(Long.MAX_VALUE, 8));
         assertThrows(IllegalArgumentException.class, () -> HydrologyPlanningAdmission.trialLimit(gibibyte, 0));
     }

core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyRegionalFallsTest.java

@@ -40,7 +40,7 @@ public final class HydrologyRegionalFallsTest {
         SurfaceBounds bounds = new SurfaceBounds(-2, -2, 1, 2);
         HydrologyFootprintCompiler compiler = new HydrologyFootprintCompiler(SETTINGS, TERRAIN, planner.geometrySampler);
         compiler.seedRegionalSurface(network, bounds);
-        HydrologyCrossTileResolver.MaterializedHydrology materialized = new HydrologyCrossTileResolver(planner)
+        HydrologyCrossTileResolver.MaterializedHydrology materialized = planner.crossTile
                 .materializeFinalHydrology(new HydrologyCaveCourseFilter.Result(network.nodes(), network.edges(),
                         network.outlets(), network.courses(), plans), new ArrayList<>(), compiler, true);
         HydrologyCaveCourseFilter.Result clipped = HydrologyCrossTileResolver.clipRegionalPlans(materialized.result(), network, bounds);
@@ -108,7 +108,7 @@ public final class HydrologyRegionalFallsTest {
         compiler.seedRegionalSurface(regional, bounds);
         HydrologyCaveCourseFilter.Result result = new HydrologyCaveCourseFilter.Result(metadata.nodes(), metadata.edges(),
                 metadata.outlets(), metadata.courses(), plans);
-        HydrologyCrossTileResolver.MaterializedHydrology materialized = new HydrologyCrossTileResolver(planner)
+        HydrologyCrossTileResolver.MaterializedHydrology materialized = planner.crossTile
                 .materializeFinalHydrology(result, new ArrayList<>(), compiler, true);
         assertEquals("A clipped raster cannot replace the full regional containment proof", 1, validations.get());
         HydrologyCaveCourseFilter.Result clipped = HydrologyCrossTileResolver.clipRegionalPlans(materialized.result(), regional, bounds);