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 19db47d82..bd7ef2f5e 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 @@ -3,6 +3,7 @@ package art.arcane.iris.generation.mantle; import art.arcane.iris.generation.runtime.IrisComplex; import art.arcane.iris.generation.runtime.Engine; import art.arcane.iris.generation.runtime.EnginePlatformHooks; +import art.arcane.iris.generation.runtime.IrisEngine; import art.arcane.iris.spi.IrisLogging; import art.arcane.iris.generation.concurrent.MultiBurst; import art.arcane.iris.generation.context.ChunkContext; @@ -106,6 +107,7 @@ public interface MatterGenerator { // write into released chunks. List outstandingTasks = null; List deferredComponents = null; + IrisEngine.GenerationRuntimeBinding generationRuntime = scopedGenerationRuntime(); SubmissionWindow submissions = multicore && !DISPATCHER.ownsCurrentThread() ? new SubmissionWindow() : null; @@ -231,7 +233,8 @@ public interface MatterGenerator { for (int componentIndex = 0; componentIndex < eligibleComponentCount; componentIndex++) { MantleComponent component = eligibleComponents[componentIndex]; outstandingTasks.add(runComponentAsync( - chunk, component, writer, finalPassX, finalPassZ, context, submissions)); + chunk, component, writer, finalPassX, finalPassZ, context, submissions, + generationRuntime)); } } else { for (int componentIndex = 0; componentIndex < eligibleComponentCount; componentIndex++) { @@ -248,6 +251,7 @@ public interface MatterGenerator { } if (asyncComponents) { + runQueuedComponentTasks(outstandingTasks, generationRuntime); awaitComponentTasks(outstandingTasks); } else if (deferredComponents != null) { runDeferredComponents(deferredComponents, writer, context); @@ -382,6 +386,19 @@ public interface MatterGenerator { return true; } + private IrisEngine.GenerationRuntimeBinding scopedGenerationRuntime() { + return getEngine() instanceof IrisEngine irisEngine && irisEngine.hasGenerationRuntimeScope() + ? irisEngine.captureGenerationRuntimeBinding() + : null; + } + + private IrisEngine.GenerationRuntimeScope openGenerationRuntime(IrisEngine.GenerationRuntimeBinding binding) { + if (binding == null || !(getEngine() instanceof IrisEngine irisEngine)) { + return null; + } + return irisEngine.openGenerationRuntimeScope(binding); + } + private boolean shouldGenerateComponent(MantleComponent component) { if (!component.isEnabled()) { return false; @@ -398,7 +415,8 @@ public interface MatterGenerator { int chunkX, int chunkZ, ChunkContext context, - SubmissionWindow submissions + SubmissionWindow submissions, + IrisEngine.GenerationRuntimeBinding generationRuntime ) { MantleFlag flag = component.getFlag(); MatterTaskKey key = new MatterTaskKey(getMantle(), chunkX, chunkZ, flag); @@ -428,24 +446,25 @@ public interface MatterGenerator { if (running != null) { return running; } - submissions.acquire(); + boolean submit = submissions.reserve(); MatterComponentTask existing = IN_FLIGHT_COMPONENTS.putIfAbsent(key, task); if (existing != null) { - submissions.release(); + if (submit) { + submissions.release(); + } return existing; } try { + if (!submit) { + completeComponentTask(task, chunk, component, writer, chunkX, chunkZ, context, null); + return task; + } IrisContext callerContext = IrisContext.get(); submissions.track(task); - task.setSubmission(DISPATCHER.submit(() -> completeComponentTask( - task, - chunk, - component, - writer, - chunkX, - chunkZ, - context, - callerContext))); + Runnable work = () -> runComponentTask( + task, chunk, component, writer, chunkX, chunkZ, context, callerContext, generationRuntime); + task.queue(work, generationRuntime); + task.setSubmission(DISPATCHER.submit(work)); return task; } catch (Throwable throwable) { task.cancel(throwable); @@ -489,6 +508,20 @@ public interface MatterGenerator { deferredComponents.clear(); } + private static void runQueuedComponentTasks(List tasks, + IrisEngine.GenerationRuntimeBinding helperRuntime) { + boolean interrupted = Thread.interrupted(); + try { + for (int index = tasks.size() - 1; index >= 0; index--) { + tasks.get(index).runIfQueued(helperRuntime); + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + /** * 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. @@ -592,19 +625,42 @@ public interface MatterGenerator { int chunkZ, ChunkContext context, IrisContext callerContext + ) { + Throwable failure = runComponentTask(task, chunk, component, writer, chunkX, chunkZ, context, callerContext, null); + if (failure instanceof RuntimeException runtime) { + throw runtime; + } + if (failure instanceof Error error) { + throw error; + } + if (failure != null) { + throw new CompletionException(failure); + } + } + + private Throwable runComponentTask( + MatterComponentTask task, + MantleChunk chunk, + MantleComponent component, + MantleWriter writer, + int chunkX, + int chunkZ, + ChunkContext context, + IrisContext callerContext, + IrisEngine.GenerationRuntimeBinding generationRuntime ) { if (!task.start()) { - return; + return null; } Throwable failure = null; try { - runComponentWithContext(chunk, component, writer, chunkX, chunkZ, context, callerContext); + runComponentWithContext(chunk, component, writer, chunkX, chunkZ, context, callerContext, generationRuntime); } catch (Throwable throwable) { failure = throwable; - throw throwable; } finally { task.finish(failure); } + return failure; } private void runComponentWithContext( @@ -614,17 +670,19 @@ public interface MatterGenerator { int chunkX, int chunkZ, ChunkContext context, - IrisContext callerContext + IrisContext callerContext, + IrisEngine.GenerationRuntimeBinding generationRuntime ) { - if (callerContext == null) { + if (callerContext == null && generationRuntime == null) { runComponentInline(chunk, component, writer, chunkX, chunkZ, context); return; } - try (IrisContext.Scope ignored = IrisContext.open( - callerContext.getEngine(), - callerContext.getGenerationSessionId(), - callerContext.getChunkContext())) { + try (IrisEngine.GenerationRuntimeScope runtimeScope = openGenerationRuntime(generationRuntime); + IrisContext.Scope contextScope = callerContext == null ? null : IrisContext.open( + callerContext.getEngine(), + callerContext.getGenerationSessionId(), + callerContext.getChunkContext())) { runComponentInline(chunk, component, writer, chunkX, chunkZ, context); } } @@ -647,6 +705,8 @@ public interface MatterGenerator { private final MatterTaskKey key; private final CompletableFuture future = new CompletableFuture<>(); private volatile Future submission; + private volatile Runnable queuedWork; + private volatile IrisEngine.GenerationRuntimeBinding queuedRuntime; private volatile Throwable failure; private boolean started; private boolean slowWarningIssued; @@ -667,6 +727,22 @@ public interface MatterGenerator { return true; } + synchronized boolean isStarted() { + return started; + } + + void queue(Runnable work, IrisEngine.GenerationRuntimeBinding runtime) { + queuedRuntime = runtime; + queuedWork = work; + } + + void runIfQueued(IrisEngine.GenerationRuntimeBinding helperRuntime) { + Runnable work = queuedWork; + if (work != null && !future.isDone() && (helperRuntime == null || helperRuntime == queuedRuntime)) { + work.run(); + } + } + synchronized void setSubmission(Future submission) { this.submission = submission; if (failure != null) { @@ -720,7 +796,7 @@ public interface MatterGenerator { continue; } Future submitted = submission; - boolean dropped = submitted != null && submitted.isDone(); + boolean dropped = submitted != null && submitted.isDone() && !isStarted(); long waited = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); if (dropped) { IllegalStateException dispatchFailure = new IllegalStateException( @@ -775,13 +851,20 @@ public interface MatterGenerator { private final Semaphore slots = new Semaphore(COMPONENT_SUBMISSION_LIMIT); private volatile Throwable failure; - void acquire() { - slots.acquireUninterruptibly(); + boolean reserve() { Throwable observed = failure; + if (observed != null) { + throw new CompletionException(observed); + } + if (!slots.tryAcquire()) { + return false; + } + observed = failure; if (observed != null) { slots.release(); throw new CompletionException(observed); } + return true; } void release() { 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 d330fe25c..f49e8c3db 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 @@ -2,6 +2,7 @@ package art.arcane.iris.generation.mantle; import art.arcane.iris.configuration.IrisSettings; import art.arcane.iris.generation.runtime.Engine; +import art.arcane.iris.generation.runtime.IrisEngine; import art.arcane.iris.generation.terrain.IrisDimension; import art.arcane.iris.testsupport.Await; import art.arcane.iris.testsupport.PlatformBinding; @@ -76,11 +77,13 @@ public class MatterGeneratorConcurrencyTest { @Test public void externalWindowsBoundManagedBlockedComponentSubmissions() throws Exception { GeneratorFixture fixture = new GeneratorFixture(true); - int maximum = 2 * MatterGenerator.COMPONENT_SUBMISSION_LIMIT; + int pooled = 2 * MatterGenerator.COMPONENT_SUBMISSION_LIMIT; + int maximum = pooled + 2; CountDownLatch admitted = new CountDownLatch(maximum); CountDownLatch excess = new CountDownLatch(1); CountDownLatch release = new CountDownLatch(1); AtomicInteger started = new AtomicInteger(); + AtomicInteger startedOnDispatcher = new AtomicInteger(); AtomicInteger completed = new AtomicInteger(); RecordingComponent component = new RecordingComponent(ReservedFlag.OBJECT, 0, 64) { @Override @@ -88,6 +91,9 @@ public class MatterGeneratorConcurrencyTest { if (started.incrementAndGet() > maximum) { excess.countDown(); } + if (MatterGenerator.DISPATCHER.ownsCurrentThread()) { + startedOnDispatcher.incrementAndGet(); + } admitted.countDown(); awaitManaged(release); completed.incrementAndGet(); @@ -102,6 +108,7 @@ public class MatterGeneratorConcurrencyTest { await(admitted); assertFalse(excess.await(300L, TimeUnit.MILLISECONDS)); assertEquals(maximum, started.get()); + assertEquals(pooled, startedOnDispatcher.get()); assertEquals(maximum, activeClaims(fixture, 0, 4) + activeClaims(fixture, 32, 4)); verify(fixture.mantle.getChunk(-4, -4), never()).release(); } finally { @@ -358,14 +365,23 @@ public class MatterGeneratorConcurrencyTest { } @Test - public void multicoreComponentRunsOnTheDispatcher() { + public void multicoreComponentsSpreadAcrossTheDispatcherWhileTheCallerHelps() { GeneratorFixture fixture = new GeneratorFixture(); Thread caller = Thread.currentThread(); - Set threads = ConcurrentHashMap.newKeySet(); + AtomicInteger dispatcherRuns = new AtomicInteger(); + AtomicInteger completed = new AtomicInteger(); + CountDownLatch dispatched = new CountDownLatch(1); RecordingComponent ordinary = new RecordingComponent(ReservedFlag.OBJECT, 0, 16) { @Override public void generateLayer(MantleWriter writer, int x, int z, ChunkContext context) { - threads.add(Thread.currentThread()); + if (Thread.currentThread() == caller) { + await(dispatched); + } else { + assertTrue(MatterGenerator.DISPATCHER.ownsCurrentThread()); + dispatcherRuns.incrementAndGet(); + dispatched.countDown(); + } + completed.incrementAndGet(); } }; TestMatterGenerator generator = fixture.generator(List.of( @@ -374,41 +390,157 @@ public class MatterGeneratorConcurrencyTest { generator.generateMatter(0, 0, true, fixture.context); - assertFalse(threads.isEmpty()); - assertFalse(threads.contains(caller)); + assertEquals(9, completed.get()); + assertTrue(dispatcherRuns.get() > 0); } @Test - public void asyncComponentReceivesCallerContextAndCallerScopeIsRestored() { + public void componentsReceiveTheCallerContextOnEveryThreadAndCallerScopeIsRestored() { GeneratorFixture fixture = new GeneratorFixture(); - AtomicReference observed = new AtomicReference<>(); - AtomicReference observedThread = new AtomicReference<>(); - RecordingComponent concurrent = new RecordingComponent(ReservedFlag.CARVED, 0, 0) { + Thread callerThread = Thread.currentThread(); + Set observed = ConcurrentHashMap.newKeySet(); + AtomicInteger dispatcherRuns = new AtomicInteger(); + CountDownLatch dispatched = new CountDownLatch(1); + RecordingComponent concurrent = new RecordingComponent(ReservedFlag.CARVED, 0, 16) { @Override public void generateLayer(MantleWriter writer, int x, int z, ChunkContext context) { - observed.set(IrisContext.require()); - observedThread.set(Thread.currentThread()); + observed.add(IrisContext.require()); + if (Thread.currentThread() == callerThread) { + await(dispatched); + } else { + assertTrue(MatterGenerator.DISPATCHER.ownsCurrentThread()); + dispatcherRuns.incrementAndGet(); + dispatched.countDown(); + } } }; TestMatterGenerator generator = fixture.generator(List.of( - new MantlePass(List.of(concurrent), 0, 0) + new MantlePass(List.of(concurrent), 1, 0) )); assertNull(IrisContext.get()); try (IrisContext.Scope ignored = IrisContext.open(fixture.engine, 91L, fixture.context)) { IrisContext caller = IrisContext.require(); - Thread callerThread = Thread.currentThread(); generator.generateMatter(0, 0, true, fixture.context); - assertNotSame(callerThread, observedThread.get()); - assertSame(fixture.engine, observed.get().getEngine()); - assertSame(fixture.context, observed.get().getChunkContext()); - assertEquals(91L, observed.get().getGenerationSessionId()); + assertTrue(dispatcherRuns.get() > 0); + for (IrisContext context : observed) { + assertSame(fixture.engine, context.getEngine()); + assertSame(fixture.context, context.getChunkContext()); + assertEquals(91L, context.getGenerationSessionId()); + } assertSame(caller, IrisContext.require()); } assertNull(IrisContext.get()); } + @Test + public void externalBarrierRunsItsQueuedComponentsWhileTheDispatcherIsBusy() throws Exception { + GeneratorFixture fixture = new GeneratorFixture(true); + CountDownLatch releaseDispatcher = new CountDownLatch(1); + List> blockers = occupyDispatcher(releaseDispatcher); + AtomicReference generationThread = new AtomicReference<>(); + Set threads = ConcurrentHashMap.newKeySet(); + AtomicInteger completed = new AtomicInteger(); + RecordingComponent component = new RecordingComponent(ReservedFlag.OBJECT, 0, 16) { + @Override + public void generateLayer(MantleWriter writer, int x, int z, ChunkContext context) { + threads.add(Thread.currentThread()); + completed.incrementAndGet(); + } + }; + TestMatterGenerator generator = fixture.generator(List.of(new MantlePass(List.of(component), 1, 0))); + try { + CompletableFuture generation = CompletableFuture.runAsync(() -> { + generationThread.set(Thread.currentThread()); + generator.generateMatter(0, 0, true, fixture.context); + }); + generation.get(5L, TimeUnit.SECONDS); + assertEquals(9, completed.get()); + assertEquals(Set.of(generationThread.get()), threads); + assertEquals(0, activeClaims(fixture, 0, 1)); + } finally { + releaseDispatcher.countDown(); + for (Future blocker : blockers) { + blocker.get(10L, TimeUnit.SECONDS); + } + } + } + + @Test + public void dispatcherComponentsRunInsideTheCallersGenerationRuntime() { + GeneratorFixture fixture = new GeneratorFixture(true, true); + Thread caller = Thread.currentThread(); + fixture.bindRuntime(caller); + IrisEngine engine = (IrisEngine) fixture.engine; + AtomicInteger dispatcherRuns = new AtomicInteger(); + AtomicInteger unscopedRuns = new AtomicInteger(); + AtomicInteger completed = new AtomicInteger(); + CountDownLatch dispatched = new CountDownLatch(1); + RecordingComponent component = new RecordingComponent(ReservedFlag.OBJECT, 0, 64) { + @Override + public void generateLayer(MantleWriter writer, int x, int z, ChunkContext context) { + if (!engine.hasGenerationRuntimeScope() || engine.captureGenerationRuntimeBinding() != fixture.runtime) { + unscopedRuns.incrementAndGet(); + } + if (Thread.currentThread() == caller) { + await(dispatched); + } else { + dispatcherRuns.incrementAndGet(); + dispatched.countDown(); + } + completed.incrementAndGet(); + } + }; + TestMatterGenerator generator = fixture.generator(List.of(new MantlePass(List.of(component), 4, 0))); + + generator.generateMatter(0, 0, true, fixture.context); + + assertEquals(81, completed.get()); + assertEquals(0, unscopedRuns.get()); + assertTrue(dispatcherRuns.get() > 0); + assertEquals(fixture.openedRuntimes.get(), fixture.closedRuntimes.get()); + assertTrue(fixture.openedRuntimes.get() >= dispatcherRuns.get()); + } + + @Test + public void helpersOnlyRunQueuedComponentsBoundToTheirOwnRuntime() { + GeneratorFixture fixture = new GeneratorFixture(); + IrisEngine.GenerationRuntimeBinding own = mock(IrisEngine.GenerationRuntimeBinding.class); + IrisEngine.GenerationRuntimeBinding foreign = mock(IrisEngine.GenerationRuntimeBinding.class); + AtomicInteger runs = new AtomicInteger(); + MatterGenerator.MatterComponentTask bound = new MatterGenerator.MatterComponentTask( + new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT)); + bound.queue(runs::incrementAndGet, own); + MatterGenerator.MatterComponentTask unbound = new MatterGenerator.MatterComponentTask( + new MatterGenerator.MatterTaskKey(fixture.mantle, 1, 0, ReservedFlag.OBJECT)); + unbound.queue(runs::incrementAndGet, null); + + bound.runIfQueued(foreign); + unbound.runIfQueued(foreign); + assertEquals(0, runs.get()); + bound.runIfQueued(own); + unbound.runIfQueued(null); + bound.runIfQueued(null); + assertEquals(3, runs.get()); + } + + private static List> occupyDispatcher(CountDownLatch release) { + int blockers = Math.max(1, MultiBurst.burst.parallelism()) * 4; + List> futures = new ArrayList<>(blockers); + for (int index = 0; index < blockers; index++) { + futures.add(MultiBurst.burst.submit(() -> { + try { + assertTrue(release.await(30L, TimeUnit.SECONDS)); + } catch (InterruptedException interruption) { + Thread.currentThread().interrupt(); + } + })); + } + return futures; + } + + @Test public void terrainPhasePreparesContentReachWithoutPublishingPlanned() { GeneratorFixture fixture = new GeneratorFixture(true); @@ -701,6 +833,42 @@ public class MatterGeneratorConcurrencyTest { assertFalse(MatterGenerator.IN_FLIGHT_COMPONENTS.containsKey(key)); } + @Test + public void componentStartedByAHelperIsNotDroppedWhenItsSubmissionFinishesFirst() throws Exception { + GeneratorFixture fixture = new GeneratorFixture(); + MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 3, 0, ReservedFlag.OBJECT); + MatterGenerator.MatterComponentTask task = new MatterGenerator.MatterComponentTask(key); + CountDownLatch running = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch polls = new CountDownLatch(2); + task.queue(() -> { + if (task.start()) { + running.countDown(); + await(release); + task.finish(null); + } + }, null); + Future submission = mock(Future.class); + when(submission.isDone()).thenAnswer(invocation -> { + polls.countDown(); + return true; + }); + task.setSubmission(submission); + CompletableFuture helper = CompletableFuture.runAsync(() -> task.runIfQueued(null)); + await(running); + CompletableFuture waiter = CompletableFuture.runAsync( + () -> task.await(MatterGenerator.COMPONENT_TASK_TIMEOUT_MS)); + try { + await(polls); + assertFalse(waiter.isDone()); + } finally { + release.countDown(); + } + helper.get(5L, TimeUnit.SECONDS); + waiter.get(5L, TimeUnit.SECONDS); + verify(submission, never()).cancel(anyBoolean()); + } + @Test public void interruptedWaiterDrainsRunningTaskAndRestoresInterrupt() throws Exception { GeneratorFixture fixture = new GeneratorFixture(); @@ -814,22 +982,49 @@ public class MatterGeneratorConcurrencyTest { private final Engine engine; private final Mantle mantle; private final ChunkContext context; + private final IrisEngine.GenerationRuntimeBinding runtime = mock(IrisEngine.GenerationRuntimeBinding.class); + private final ThreadLocal runtimeDepth = ThreadLocal.withInitial(() -> 0); + private final AtomicInteger openedRuntimes = new AtomicInteger(); + private final AtomicInteger closedRuntimes = new AtomicInteger(); + private volatile Thread boundThread; private GeneratorFixture() { this(false); } + private GeneratorFixture(boolean lockingChunks) { + this(lockingChunks, false); + } + /** * @param lockingChunks true to give every chunk a real per-flag lock and flag state, so a * generation that reaches a chunk another generation is working on * blocks exactly like production */ @SuppressWarnings("unchecked") - private GeneratorFixture(boolean lockingChunks) { + private GeneratorFixture(boolean lockingChunks, boolean runtimeBoundCaller) { IrisDimension dimension = mock(IrisDimension.class); when(dimension.isUseMantle()).thenReturn(true); EngineMantle engineMantle = mock(EngineMantle.class); - engine = mock(Engine.class); + if (runtimeBoundCaller) { + IrisEngine irisEngine = mock(IrisEngine.class); + when(irisEngine.hasGenerationRuntimeScope()).thenAnswer(invocation -> runtimeBound()); + when(irisEngine.captureGenerationRuntimeBinding()).thenAnswer(invocation -> runtimeBound() ? runtime : null); + when(irisEngine.openGenerationRuntimeScope(runtime)).thenAnswer(invocation -> { + openedRuntimes.incrementAndGet(); + runtimeDepth.set(runtimeDepth.get() + 1); + IrisEngine.GenerationRuntimeScope scope = mock(IrisEngine.GenerationRuntimeScope.class); + doAnswer(close -> { + runtimeDepth.set(runtimeDepth.get() - 1); + closedRuntimes.incrementAndGet(); + return null; + }).when(scope).close(); + return scope; + }); + engine = irisEngine; + } else { + engine = mock(Engine.class); + } when(engine.getDimension()).thenReturn(dimension); when(engine.getMantle()).thenReturn(engineMantle); mantle = mock(Mantle.class); @@ -846,6 +1041,14 @@ public class MatterGeneratorConcurrencyTest { when(context.getGenerationSessionId()).thenReturn(91L); } + private void bindRuntime(Thread thread) { + boundThread = thread; + } + + private boolean runtimeBound() { + return Thread.currentThread() == boundThread || runtimeDepth.get() > 0; + } + private TestMatterGenerator generator(List passes) { return new TestMatterGenerator(engine, mantle, passes); } diff --git a/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineGenerationRuntimeScopeTest.java b/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineGenerationRuntimeScopeTest.java index 9b0f9177b..81f5720fd 100644 --- a/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineGenerationRuntimeScopeTest.java +++ b/core/src/test/java/art/arcane/iris/generation/runtime/IrisEngineGenerationRuntimeScopeTest.java @@ -42,6 +42,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.FutureTask; import java.util.concurrent.atomic.AtomicBoolean; @@ -431,25 +432,36 @@ public class IrisEngineGenerationRuntimeScopeTest { }; MultiBurst workerPool = new MultiBurst("Generation Runtime Scope Test", Thread.NORM_PRIORITY, () -> 2); when(detached.target.getBurster()).thenReturn(workerPool); - AtomicReference observedTarget = new AtomicReference<>(); - AtomicReference observedThread = new AtomicReference<>(); + Set observedTargets = new CopyOnWriteArraySet<>(); + Set workerThreads = new CopyOnWriteArraySet<>(); + CountDownLatch workerRan = new CountDownLatch(1); + Thread caller = Thread.currentThread(); EngineStage stage = (x, z, blocks, biomes, multicore, context) -> { - observedTarget.set(engine.getTarget()); - observedThread.set(Thread.currentThread()); + observedTargets.add(engine.getTarget()); + if (Thread.currentThread() == caller) { + try { + assertTrue(workerRan.await(5L, TimeUnit.SECONDS)); + } catch (InterruptedException interruption) { + Thread.currentThread().interrupt(); + throw new AssertionError(interruption); + } + return; + } + workerThreads.add(Thread.currentThread()); + workerRan.countDown(); }; ChunkContext context = mock(ChunkContext.class); when(context.getGenerationSessionId()).thenReturn(17L); - Thread caller = Thread.currentThread(); try (IrisEngine.GenerationRuntimeScope ignored = engine.openGenerationRuntimeScope(binding)) { - mode.burst(stage).generate(0, 0, null, null, true, context); + mode.burst(stage, stage).generate(0, 0, null, null, true, context); } finally { mode.close(); workerPool.close(); } - assertSame(detached.target, observedTarget.get()); - assertNotSame(caller, observedThread.get()); + assertEquals(Set.of(detached.target), observedTargets); + assertTrue(!workerThreads.isEmpty() && !workerThreads.contains(caller)); } @Test