Sliding mantle submissions

- Replaced the barrier after every 8 component submissions with a per-caller cap that frees a slot as each component finishes and stops on the first failure
- Inline windows claim components without waiting and run components another generation holds after the rest of the pass
- Prefetched only the radius of passes that generate in the current phase
- Added straggler and held-claim tests and modeled claims in the locking chunk fixture

core/src/main/java/art/arcane/iris/generation/mantle/MatterGenerator.java

@@ -23,13 +23,14 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 public interface MatterGenerator {
     MultiBurst DISPATCHER = MultiBurst.burst;
     ConcurrentHashMap<MatterTaskKey, MatterComponentTask> IN_FLIGHT_COMPONENTS = new ConcurrentHashMap<>();
-    int COMPONENT_SUBMISSION_BATCH_SIZE = Math.max(1, Math.min(8, Runtime.getRuntime().availableProcessors()));
+    int COMPONENT_SUBMISSION_LIMIT = Math.max(1, Math.min(8, Runtime.getRuntime().availableProcessors()));
     long COMPONENT_TASK_POLL_MS = 1000L;
     long COMPONENT_TASK_TIMEOUT_MS = Long.getLong("iris.mantle.componentTimeout", 120000L);
 
@@ -87,7 +88,7 @@ public interface MatterGenerator {
         if (passPlans.length == 0) {
             return;
         }
-        int prefetchRadius = passPlans[0].passChunkRadius();
+        int prefetchRadius = phasePrefetchRadius(passPlans, phase);
         LongOpenHashSet partialChunks = new LongOpenHashSet();
         List<MantleComponent> requiredComponents = enabledComponents();
         boolean terrainOnAccess = false;
@@ -104,8 +105,9 @@ public interface MatterGenerator {
             // close() releases the cached chunks, even when a pass throws, or detached pool threads
             // write into released chunks.
             List<MatterComponentTask> outstandingTasks = null;
-            List<MatterComponentTask> submittedTasks = multicore && !DISPATCHER.ownsCurrentThread()
+            List<DeferredComponent> deferredComponents = null;
-                    ? new ArrayList<>(COMPONENT_SUBMISSION_BATCH_SIZE)
+            SubmissionWindow submissions = multicore && !DISPATCHER.ownsCurrentThread()
+                    ? new SubmissionWindow()
                     : null;
 
             try {
@@ -229,15 +231,17 @@ public interface MatterGenerator {
                                 for (int componentIndex = 0; componentIndex < eligibleComponentCount; componentIndex++) {
                                     MantleComponent component = eligibleComponents[componentIndex];
                                     outstandingTasks.add(runComponentAsync(
-                                            chunk, component, writer, finalPassX, finalPassZ, context, submittedTasks));
+                                            chunk, component, writer, finalPassX, finalPassZ, context, submissions));
-                                    if (submittedTasks != null && submittedTasks.size() >= COMPONENT_SUBMISSION_BATCH_SIZE) {
-                                        awaitComponentTasks(submittedTasks);
-                                    }
                                 }
                             } else {
                                 for (int componentIndex = 0; componentIndex < eligibleComponentCount; componentIndex++) {
                                     MantleComponent component = eligibleComponents[componentIndex];
-                                    runComponentInline(chunk, component, writer, finalPassX, finalPassZ, context);
+                                    if (!tryRunComponentInline(chunk, component, writer, finalPassX, finalPassZ, context)) {
+                                        if (deferredComponents == null) {
+                                            deferredComponents = new ArrayList<>();
+                                        }
+                                        deferredComponents.add(new DeferredComponent(chunk, component, finalPassX, finalPassZ));
+                                    }
                                 }
                             }
                         }
@@ -245,9 +249,8 @@ public interface MatterGenerator {
 
                     if (asyncComponents) {
                         awaitComponentTasks(outstandingTasks);
-                        if (submittedTasks != null) {
+                    } else if (deferredComponents != null) {
-                            submittedTasks.clear();
+                        runDeferredComponents(deferredComponents, writer, context);
-                        }
                     }
                 }
 
@@ -320,6 +323,25 @@ public interface MatterGenerator {
         return new MatterGenerationPlan(plans, writerChunkRadius * 2, phase);
     }
 
+    private int phasePrefetchRadius(MatterPassPlan[] passPlans, MatterGenerationPhase phase) {
+        int radius = 0;
+        for (MatterPassPlan passPlan : passPlans) {
+            if (passPlan.passChunkRadius() > radius && generatesInPhase(passPlan.pass(), phase)) {
+                radius = passPlan.passChunkRadius();
+            }
+        }
+        return radius;
+    }
+
+    private boolean generatesInPhase(MantlePass pass, MatterGenerationPhase phase) {
+        for (MantleComponent component : pass.components()) {
+            if (phase.includes(component) && shouldGenerateComponent(component)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     private List<MantleComponent> enabledComponents() {
         List<MantleComponent> components = new ArrayList<>();
         for (MantlePass pass : getComponents()) {
@@ -376,7 +398,7 @@ public interface MatterGenerator {
             int chunkX,
             int chunkZ,
             ChunkContext context,
-            List<MatterComponentTask> submittedTasks
+            SubmissionWindow submissions
     ) {
         MantleFlag flag = component.getFlag();
         MatterTaskKey key = new MatterTaskKey(getMantle(), chunkX, chunkZ, flag);
@@ -386,19 +408,35 @@ public interface MatterGenerator {
             return task;
         }
 
-        MatterComponentTask existing = IN_FLIGHT_COMPONENTS.putIfAbsent(key, task);
+        if (submissions == null) {
-        if (existing != null) {
+            MatterComponentTask existing = IN_FLIGHT_COMPONENTS.putIfAbsent(key, task);
-            return existing;
+            if (existing != null) {
-        }
+                return existing;
-
+            }
-        try {
+            try {
-            if (DISPATCHER.ownsCurrentThread()) {
                 // A pool thread runs its claim inline: a task queued behind pool threads that block
                 // on unmanaged waits (structure builds, cache locks) could starve the whole pool.
                 completeComponentTask(task, chunk, component, writer, chunkX, chunkZ, context, IrisContext.get());
                 return task;
+            } catch (Throwable throwable) {
+                task.cancel(throwable);
+                throw throwable;
             }
+        }
+
+        MatterComponentTask running = IN_FLIGHT_COMPONENTS.get(key);
+        if (running != null) {
+            return running;
+        }
+        submissions.acquire();
+        MatterComponentTask existing = IN_FLIGHT_COMPONENTS.putIfAbsent(key, task);
+        if (existing != null) {
+            submissions.release();
+            return existing;
+        }
+        try {
             IrisContext callerContext = IrisContext.get();
+            submissions.track(task);
             task.setSubmission(DISPATCHER.submit(() -> completeComponentTask(
                     task,
                     chunk,
@@ -408,7 +446,6 @@ public interface MatterGenerator {
                     chunkZ,
                     context,
                     callerContext)));
-            submittedTasks.add(task);
             return task;
         } catch (Throwable throwable) {
             task.cancel(throwable);
@@ -416,6 +453,42 @@ public interface MatterGenerator {
         }
     }
 
+    private boolean tryRunComponentInline(
+            MantleChunk<Matter> chunk,
+            MantleComponent component,
+            MantleWriter writer,
+            int chunkX,
+            int chunkZ,
+            ChunkContext context
+    ) {
+        MantleFlag flag = component.getFlag();
+        if (!chunk.tryClaim(flag)) {
+            return chunk.isFlagged(flag);
+        }
+        boolean generated = false;
+        try {
+            writer.withComponentPriority(component.getPriority(),
+                    () -> component.generateLayer(writer, chunkX, chunkZ, context));
+            generated = true;
+        } finally {
+            if (generated) {
+                chunk.completeClaim(flag);
+            } else {
+                chunk.abandonClaim(flag);
+            }
+        }
+        return true;
+    }
+
+    private void runDeferredComponents(List<DeferredComponent> deferredComponents, MantleWriter writer,
+                                       ChunkContext context) {
+        for (DeferredComponent deferred : deferredComponents) {
+            runComponentInline(deferred.chunk(), deferred.component(), writer, deferred.chunkX(), deferred.chunkZ(),
+                    context);
+        }
+        deferredComponents.clear();
+    }
+
     /**
      * Pass barrier. Waits for every launched task, even if one of them failed, so that no task is
      * still writing through the writer once this returns. The first failure is rethrown.
@@ -698,6 +771,36 @@ public interface MatterGenerator {
         }
     }
 
+    final class SubmissionWindow {
+        private final Semaphore slots = new Semaphore(COMPONENT_SUBMISSION_LIMIT);
+        private volatile Throwable failure;
+
+        void acquire() {
+            slots.acquireUninterruptibly();
+            Throwable observed = failure;
+            if (observed != null) {
+                slots.release();
+                throw new CompletionException(observed);
+            }
+        }
+
+        void release() {
+            slots.release();
+        }
+
+        void track(MatterComponentTask task) {
+            task.future.whenComplete((ignored, throwable) -> {
+                if (throwable != null && failure == null) {
+                    failure = throwable;
+                }
+                slots.release();
+            });
+        }
+    }
+
+    record DeferredComponent(MantleChunk<Matter> chunk, MantleComponent component, int chunkX, int chunkZ) {
+    }
+
     record MatterPassPlan(MantlePass pass, int passChunkRadius, int downstreamBlockRadius) {
     }
 

core/src/test/java/art/arcane/iris/generation/mantle/MatterGeneratorConcurrencyTest.java

@@ -76,7 +76,7 @@ public class MatterGeneratorConcurrencyTest {
     @Test
     public void externalWindowsBoundManagedBlockedComponentSubmissions() throws Exception {
         GeneratorFixture fixture = new GeneratorFixture(true);
-        int maximum = 2 * MatterGenerator.COMPONENT_SUBMISSION_BATCH_SIZE;
+        int maximum = 2 * MatterGenerator.COMPONENT_SUBMISSION_LIMIT;
         CountDownLatch admitted = new CountDownLatch(maximum);
         CountDownLatch excess = new CountDownLatch(1);
         CountDownLatch release = new CountDownLatch(1);
@@ -115,10 +115,10 @@ public class MatterGeneratorConcurrencyTest {
     }
 
     @Test
-    public void externalWindowDefersSharedClaimsUntilItsOwnBatchesComplete() throws Exception {
+    public void externalWindowDefersSharedClaimsUntilItsOwnSubmissionsComplete() throws Exception {
         GeneratorFixture fixture = new GeneratorFixture(true);
         List<MatterGenerator.MatterComponentTask> shared = new ArrayList<>();
-        for (int z = -4; z < -4 + MatterGenerator.COMPONENT_SUBMISSION_BATCH_SIZE; z++) {
+        for (int z = -4; z < -4 + MatterGenerator.COMPONENT_SUBMISSION_LIMIT; z++) {
             MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(
                     fixture.mantle, -4, z, ReservedFlag.OBJECT);
             MatterGenerator.MatterComponentTask task = new MatterGenerator.MatterComponentTask(key);
@@ -137,7 +137,7 @@ public class MatterGeneratorConcurrencyTest {
         CompletableFuture<Void> generation = CompletableFuture.runAsync(
                 () -> generator.generateMatter(0, 0, true, fixture.context));
         try {
-            Await.until("all privately owned component batches", Duration.ofSeconds(10L),
+            Await.until("all privately owned component submissions", Duration.ofSeconds(10L),
                     () -> completed.get() == 81 - shared.size());
             assertFalse(generation.isDone());
         } finally {
@@ -149,16 +149,17 @@ public class MatterGeneratorConcurrencyTest {
     }
 
     @Test
-    public void failedSubmissionBatchDrainsItsWritersBeforeReturning() throws Exception {
+    public void failedSubmissionDrainsItsWritersBeforeReturning() throws Exception {
         GeneratorFixture fixture = new GeneratorFixture(true);
         CountDownLatch failed = new CountDownLatch(1);
-        CountDownLatch started = new CountDownLatch(MatterGenerator.COMPONENT_SUBMISSION_BATCH_SIZE - 1);
+        CountDownLatch started = new CountDownLatch(MatterGenerator.COMPONENT_SUBMISSION_LIMIT - 1);
         CountDownLatch release = new CountDownLatch(1);
         IllegalStateException failure = new IllegalStateException("Component rejected its input");
         RecordingComponent component = new RecordingComponent(ReservedFlag.OBJECT, 0, 64) {
             @Override
             public void generateLayer(MantleWriter writer, int x, int z, ChunkContext context) {
                 if (x == -4 && z == -4) {
+                    await(started);
                     failed.countDown();
                     throw failure;
                 }
@@ -172,7 +173,7 @@ public class MatterGeneratorConcurrencyTest {
         try {
             await(failed);
             await(started);
-            if (MatterGenerator.COMPONENT_SUBMISSION_BATCH_SIZE > 1) {
+            if (MatterGenerator.COMPONENT_SUBMISSION_LIMIT > 1) {
                 assertFalse(generation.isDone());
                 verify(fixture.mantle.getChunk(-4, -4), never()).release();
             }
@@ -186,6 +187,69 @@ public class MatterGeneratorConcurrencyTest {
         assertEquals(0, activeClaims(fixture, 0, 4));
     }
 
+    @Test
+    public void externalWindowKeepsSubmittingPastAStragglingComponent() throws Exception {
+        GeneratorFixture fixture = new GeneratorFixture(true);
+        CountDownLatch release = new CountDownLatch(1);
+        AtomicInteger completed = new AtomicInteger();
+        RecordingComponent component = new RecordingComponent(ReservedFlag.OBJECT, 0, 64) {
+            @Override
+            public void generateLayer(MantleWriter writer, int x, int z, ChunkContext context) {
+                if (x == -4 && z == -4) {
+                    awaitManaged(release);
+                }
+                completed.incrementAndGet();
+            }
+        };
+        TestMatterGenerator generator = fixture.generator(List.of(new MantlePass(List.of(component), 4, 0)));
+        CompletableFuture<Void> generation = CompletableFuture.runAsync(
+                () -> generator.generateMatter(0, 0, true, fixture.context));
+        try {
+            Await.until("every chunk but the straggler", Duration.ofSeconds(10L), () -> completed.get() == 80);
+            assertFalse(generation.isDone());
+        } finally {
+            release.countDown();
+        }
+        generation.get(10L, TimeUnit.SECONDS);
+        assertEquals(81, completed.get());
+        assertEquals(0, activeClaims(fixture, 0, 4));
+    }
+
+    @Test
+    public void inlineWindowFinishesItsOwnChunksBeforeJoiningAClaimAnotherGenerationHolds() throws Exception {
+        GeneratorFixture fixture = new GeneratorFixture(true);
+        MantleChunk<Matter> held = fixture.mantle.getChunk(1, 0);
+        CountDownLatch claimed = new CountDownLatch(1);
+        CountDownLatch releaseClaim = new CountDownLatch(1);
+        Set<Long> completed = ConcurrentHashMap.newKeySet();
+        RecordingComponent component = new RecordingComponent(ReservedFlag.OBJECT, 0, 16) {
+            @Override
+            public void generateLayer(MantleWriter writer, int x, int z, ChunkContext context) {
+                completed.add((((long) x) << 32) ^ (z & 0xffffffffL));
+            }
+        };
+        TestMatterGenerator generator = fixture.generator(List.of(new MantlePass(List.of(component), 1, 0)));
+        CompletableFuture<Void> holder = CompletableFuture.runAsync(() -> {
+            assertTrue(held.tryClaim(ReservedFlag.OBJECT));
+            claimed.countDown();
+            await(releaseClaim);
+            held.completeClaim(ReservedFlag.OBJECT);
+        });
+        await(claimed);
+        CompletableFuture<Void> generation = CompletableFuture.runAsync(
+                () -> generator.generateMatter(0, 0, false, fixture.context));
+        try {
+            Await.until("the eight unclaimed chunks", Duration.ofSeconds(10L), () -> completed.size() == 8);
+            assertFalse(generation.isDone());
+        } finally {
+            releaseClaim.countDown();
+        }
+        holder.get(10L, TimeUnit.SECONDS);
+        generation.get(10L, TimeUnit.SECONDS);
+        assertEquals(8, completed.size());
+        assertTrue(held.isFlagged(ReservedFlag.OBJECT));
+    }
+
     @Test
     public void boundedAndReentrantGenerationMatchSerialComponentResults() throws Exception {
         Map<Long, Long> serial = componentResults(false, false);
@@ -812,6 +876,25 @@ public class MatterGeneratorConcurrencyTest {
                 }
                 return null;
             }).when(chunk).raiseFlagSuspend(any(), any(Runnable.class));
+            when(chunk.tryClaim(any())).thenAnswer(invocation -> {
+                if (flagged.contains(invocation.getArgument(0)) || !lock.tryLock()) {
+                    return false;
+                }
+                if (flagged.contains(invocation.getArgument(0))) {
+                    lock.unlock();
+                    return false;
+                }
+                return true;
+            });
+            doAnswer(invocation -> {
+                flagged.add(invocation.getArgument(0));
+                lock.unlock();
+                return null;
+            }).when(chunk).completeClaim(any());
+            doAnswer(invocation -> {
+                lock.unlock();
+                return null;
+            }).when(chunk).abandonClaim(any());
             return chunk;
         }