Mantle window helping

- A caller whose submission window is full runs the next component itself instead of parking
- Pass barriers run still-queued component tasks in reverse order before waiting
- Queued component tasks reopen the submitting generation's runtime scope on whichever thread runs them
- Helpers only run queued tasks bound to their own runtime or when they hold none
- A finished submission no longer reports a task as dropped once a helper started it
- Burst worker scope test keeps one stage on a worker while the caller helps
This commit is contained in:
Brian Neumann-Fopiano committed 2026-10-22 02:56:23 -04:00
1 parent 024214d7f0
commit e91eb258aa
3 files changed
+351 -53

No files matched your search

@@ -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<MatterComponentTask> outstandingTasks = null;
List<DeferredComponent> 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<MatterComponentTask> 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<Matter> 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<Void> 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() {
@@ -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<Thread> 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<IrisContext> observed = new AtomicReference<>();
AtomicReference<Thread> observedThread = new AtomicReference<>();
RecordingComponent concurrent = new RecordingComponent(ReservedFlag.CARVED, 0, 0) {
Thread callerThread = Thread.currentThread();
Set<IrisContext> 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<Future<?>> blockers = occupyDispatcher(releaseDispatcher);
AtomicReference<Thread> generationThread = new AtomicReference<>();
Set<Thread> 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<Void> 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<Future<?>> occupyDispatcher(CountDownLatch release) {
int blockers = Math.max(1, MultiBurst.burst.parallelism()) * 4;
List<Future<?>> 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<Void> helper = CompletableFuture.runAsync(() -> task.runIfQueued(null));
await(running);
CompletableFuture<Void> 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<Matter> mantle;
private final ChunkContext context;
private final IrisEngine.GenerationRuntimeBinding runtime = mock(IrisEngine.GenerationRuntimeBinding.class);
private final ThreadLocal<Integer> 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<MantlePass> passes) {
return new TestMatterGenerator(engine, mantle, passes);
}
@@ -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<EngineTarget> observedTarget = new AtomicReference<>();
AtomicReference<Thread> observedThread = new AtomicReference<>();
Set<EngineTarget> observedTargets = new CopyOnWriteArraySet<>();
Set<Thread> 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