diff --git a/core/src/main/java/art/arcane/iris/generation/mantle/MatterGenerator.java b/core/src/main/java/art/arcane/iris/generation/mantle/MatterGenerator.java index 5b42014ed..19db47d82 100644 --- a/core/src/main/java/art/arcane/iris/generation/mantle/MatterGenerator.java +++ b/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 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 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 outstandingTasks = null; - List submittedTasks = multicore && !DISPATCHER.ownsCurrentThread() - ? new ArrayList<>(COMPONENT_SUBMISSION_BATCH_SIZE) + List deferredComponents = null; + 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)); - if (submittedTasks != null && submittedTasks.size() >= COMPONENT_SUBMISSION_BATCH_SIZE) { - awaitComponentTasks(submittedTasks); - } + chunk, component, writer, finalPassX, finalPassZ, context, submissions)); } } 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) { - submittedTasks.clear(); - } + } else if (deferredComponents != null) { + 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 enabledComponents() { List components = new ArrayList<>(); for (MantlePass pass : getComponents()) { @@ -376,7 +398,7 @@ public interface MatterGenerator { int chunkX, int chunkZ, ChunkContext context, - List 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 (existing != null) { - return existing; - } - - try { - if (DISPATCHER.ownsCurrentThread()) { + if (submissions == null) { + MatterComponentTask existing = IN_FLIGHT_COMPONENTS.putIfAbsent(key, task); + if (existing != null) { + return existing; + } + try { // 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 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 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 chunk, MantleComponent component, int chunkX, int chunkZ) { + } + record MatterPassPlan(MantlePass pass, int passChunkRadius, int downstreamBlockRadius) { } diff --git a/core/src/test/java/art/arcane/iris/generation/mantle/MatterGeneratorConcurrencyTest.java b/core/src/test/java/art/arcane/iris/generation/mantle/MatterGeneratorConcurrencyTest.java index ad9cf6d22..d330fe25c 100644 --- a/core/src/test/java/art/arcane/iris/generation/mantle/MatterGeneratorConcurrencyTest.java +++ b/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 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 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 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 held = fixture.mantle.getChunk(1, 0); + CountDownLatch claimed = new CountDownLatch(1); + CountDownLatch releaseClaim = new CountDownLatch(1); + Set 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 holder = CompletableFuture.runAsync(() -> { + assertTrue(held.tryClaim(ReservedFlag.OBJECT)); + claimed.countDown(); + await(releaseClaim); + held.completeClaim(ReservedFlag.OBJECT); + }); + await(claimed); + CompletableFuture 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 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; }