mirror of
https://github.com/VolmitSoftware/Iris.git
synced 2026-10-26 05:50:39 +00:00
Merged wave1 into FIX1
This commit is contained in:
commit
5c4bce0c1a
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() {
|
||||
|
||||
+223
-20
@@ -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);
|
||||
}
|
||||
|
||||
+20
-8
@@ -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
|
||||
|
||||
Reference in new issue
Block a user