diff --git a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyCrossTileResolver.java b/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyCrossTileResolver.java index 5868d6651..90cf768d2 100644 --- a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyCrossTileResolver.java +++ b/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 outlets, int start ) { - int parallelism = HydrologyPlanningAdmission.effectiveParallelism( - Thread.currentThread() instanceof ForkJoinWorkerThread worker - ? worker.getPool().getParallelism() - : IrisPlatforms.isBound() ? MultiBurst.hydrology.parallelism() : 1); - int batchSize = parallelism <= 2 ? 1 : Math.max(1, Math.min(HydrologyPlanningAdmission.maximumTrials(), - Math.min(FALLBACK_TRIAL_PARALLELISM, parallelism))); + int parallelism; + int idleWorkers; + if (Thread.currentThread() instanceof ForkJoinWorkerThread worker) { + ForkJoinPool pool = worker.getPool(); + parallelism = HydrologyPlanningAdmission.effectiveParallelism(pool.getParallelism()); + 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> tasks = new ArrayList<>(end - start); HydrologyPlanner.PlanningSamples ownerSamples = planner.planningSamples.get(); diff --git a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyPlanner.java b/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyPlanner.java index 72eb2b249..371a0d1e8 100644 --- a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyPlanner.java +++ b/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); diff --git a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyPlanningAdmission.java b/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyPlanningAdmission.java index 6a50ff131..e2f53a068 100644 --- a/core/src/main/java/art/arcane/iris/generation/hydrology/HydrologyPlanningAdmission.java +++ b/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) { diff --git a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyFallbackParallelismTest.java b/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyFallbackParallelismTest.java index 28c76d145..3641a8c68 100644 --- a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyFallbackParallelismTest.java +++ b/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}; diff --git a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyPlanHashTest.java b/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyPlanHashTest.java index 4fa7e6e8c..92321df89 100644 --- a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyPlanHashTest.java +++ b/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 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); - assertTrue("planned columns: " + columns, columns > 0); - assertEquals(EXPECTED_PLAN_DIGEST, HexFormat.of().formatHex(canonical.digest())); + return new PlanDigest(HexFormat.of().formatHex(canonical.digest()), courses, columns); + } + + 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; diff --git a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyPlanningAdmissionTest.java b/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyPlanningAdmissionTest.java index 90a41faed..7226e3887 100644 --- a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyPlanningAdmissionTest.java +++ b/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; - assertEquals(1, HydrologyPlanningAdmission.trialLimit(1280L * 1024L * 1024L, 1)); - assertEquals(2, HydrologyPlanningAdmission.trialLimit(2L * gibibyte, 1)); - assertEquals(2, HydrologyPlanningAdmission.trialLimit(4L * gibibyte, 2)); - assertEquals(1, HydrologyPlanningAdmission.trialLimit(gibibyte / 2L, 1)); - assertEquals(8, HydrologyPlanningAdmission.trialLimit(64L * gibibyte, 8)); + long mebibyte = 1024L * 1024L; + long gibibyte = 1024L * mebibyte; + assertEquals(1, HydrologyPlanningAdmission.trialLimit(64L * mebibyte, 1)); + assertEquals(1, HydrologyPlanningAdmission.trialLimit(128L * mebibyte, 1)); + assertEquals(4, HydrologyPlanningAdmission.trialLimit(gibibyte / 2L, 1)); + 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)); } diff --git a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyRegionalFallsTest.java b/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyRegionalFallsTest.java index 9cf79e993..e2e01073f 100644 --- a/core/src/test/java/art/arcane/iris/generation/hydrology/HydrologyRegionalFallsTest.java +++ b/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);