mirror of
https://github.com/VolmitSoftware/Iris.git
synced 2026-10-11 19:14:11 +00:00
Preserved mantle task ownership
- Keep component cancellation separate from actual completion before releasing writers and deduplication claims - Reject cancelled queued component claims before they can start writing - Retain failed runtime assemblies and unpublished runtimes for shutdown retry - Gate storage and data release on completed runtime cleanup - Cover cancellation, shared writers, interrupted waits, and failed shutdown retries
This commit is contained in:
Brian Neumann-Fopiano
@@ -175,7 +175,7 @@ final class EngineHotloader {
|
||||
broadcastStudioHotload(false, "");
|
||||
} catch (Throwable e) {
|
||||
if (!published) {
|
||||
if (replacementData != null) {
|
||||
if (replacementData != null && !engine.shutdownSequence.retainsData(replacementData)) {
|
||||
replacementData.unregisterEngine(engine);
|
||||
Throwable replacementDataFailure = runCleanup(null, replacementData::close);
|
||||
if (replacementDataFailure != null) {
|
||||
|
||||
@@ -209,7 +209,12 @@ final class EngineRuntimeBuilder {
|
||||
next.generation().mantle(),
|
||||
null);
|
||||
if (retirementFailure != null) {
|
||||
retirementFailure = engine.shutdownSequence.closeRuntime(next, retirementFailure);
|
||||
Throwable cleanupFailure = engine.shutdownSequence.closeRuntime(
|
||||
next, previous == null ? null : previous.generation().mantle(), null);
|
||||
if (cleanupFailure != null) {
|
||||
engine.shutdownSequence.retainUnpublishedRuntime(next);
|
||||
retirementFailure = EngineShutdownSequence.appendFailure(retirementFailure, cleanupFailure);
|
||||
}
|
||||
engine.lifecycleState = LifecycleState.FAILED;
|
||||
throw new IllegalStateException("Failed to retire the previous Iris engine runtime.", retirementFailure);
|
||||
}
|
||||
@@ -229,6 +234,7 @@ final class EngineRuntimeBuilder {
|
||||
engine.getClosing().set(true);
|
||||
engine.backgroundTasks.closeBackgroundTaskAdmission();
|
||||
engine.lifecycleState = LifecycleState.FAILED;
|
||||
Throwable cleanupFailure = null;
|
||||
try {
|
||||
engine.getGenerationSessions().sealAndAwait(
|
||||
"failed world manager start",
|
||||
@@ -236,11 +242,17 @@ final class EngineRuntimeBuilder {
|
||||
true
|
||||
);
|
||||
} catch (Throwable drainFailure) {
|
||||
e.addSuppressed(drainFailure);
|
||||
cleanupFailure = drainFailure;
|
||||
}
|
||||
engine.shutdownSequence.closeRuntime(next, e);
|
||||
if (engine.runtime == next) {
|
||||
engine.runtime = null;
|
||||
if (cleanupFailure == null) {
|
||||
cleanupFailure = engine.shutdownSequence.closeRuntime(next, null);
|
||||
}
|
||||
if (cleanupFailure == null) {
|
||||
if (engine.runtime == next) {
|
||||
engine.runtime = null;
|
||||
}
|
||||
} else {
|
||||
e.addSuppressed(cleanupFailure);
|
||||
}
|
||||
throw new IllegalStateException("Failed to start the Iris world manager.", e);
|
||||
}
|
||||
|
||||
@@ -18,6 +18,7 @@
|
||||
|
||||
package art.arcane.iris.engine;
|
||||
|
||||
import art.arcane.iris.core.loader.IrisData;
|
||||
import art.arcane.iris.engine.EngineBackgroundTasks.BackgroundTaskDrain;
|
||||
import art.arcane.iris.engine.EngineRuntimeBuilder.RuntimeAssembly;
|
||||
import art.arcane.iris.engine.IrisEngine.LifecycleState;
|
||||
@@ -29,6 +30,11 @@ import art.arcane.iris.engine.mantle.EngineMantle;
|
||||
import art.arcane.iris.spi.IrisLogging;
|
||||
import art.arcane.iris.spi.IrisServices;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.IdentityHashMap;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
/**
|
||||
* Owns the ordered teardown of a single {@link IrisEngine}, both for a normal close and for a
|
||||
* construction that failed part way through. Every resource release is gated on the previous one
|
||||
@@ -38,6 +44,10 @@ final class EngineShutdownSequence {
|
||||
private static final long CLOSE_RETRY_DRAIN_TIMEOUT_MILLIS = 5000L;
|
||||
|
||||
private final IrisEngine engine;
|
||||
private final Set<RuntimeAssembly> incompleteAssemblies = ConcurrentHashMap.newKeySet();
|
||||
private final Set<EngineRuntime> unpublishedRuntimes = ConcurrentHashMap.newKeySet();
|
||||
private final Set<EngineTarget> incompleteTargets = Collections.synchronizedSet(
|
||||
Collections.newSetFromMap(new IdentityHashMap<>()));
|
||||
private boolean runtimeReleased;
|
||||
private boolean targetReleased;
|
||||
private boolean engineDataReleased;
|
||||
@@ -172,17 +182,21 @@ final class EngineShutdownSequence {
|
||||
EngineTickRegistry.unregisterTicking(engine);
|
||||
engine.lifecycleState = LifecycleState.FAILED;
|
||||
Throwable cleanupFailure = null;
|
||||
boolean sessionsDrained = true;
|
||||
try {
|
||||
engine.getGenerationSessions().sealAndAwait("failed initialization", 0L, true);
|
||||
} catch (Throwable e) {
|
||||
sessionsDrained = false;
|
||||
cleanupFailure = appendFailure(cleanupFailure, e);
|
||||
}
|
||||
engine.backgroundTasks.cancelBackgroundTasks("failed initialization");
|
||||
BackgroundTaskDrain backgroundDrain = engine.backgroundTasks.drainBackgroundTasks("failed initialization");
|
||||
cleanupFailure = appendFailure(cleanupFailure, backgroundDrain.failure());
|
||||
if (!backgroundDrain.allowsResourceRelease()) {
|
||||
cleanupFailure = appendFailure(cleanupFailure,
|
||||
new IllegalStateException("Iris background tasks remain active after failed initialization."));
|
||||
if (!sessionsDrained || !backgroundDrain.allowsResourceRelease()) {
|
||||
if (!backgroundDrain.allowsResourceRelease()) {
|
||||
cleanupFailure = appendFailure(cleanupFailure,
|
||||
new IllegalStateException("Iris background tasks remain active after failed initialization."));
|
||||
}
|
||||
if (cleanupFailure != original) {
|
||||
original.addSuppressed(cleanupFailure);
|
||||
}
|
||||
@@ -192,27 +206,19 @@ final class EngineShutdownSequence {
|
||||
() -> NativeStructureOwnershipStore.close(engine));
|
||||
cleanupFailure = appendFailure(cleanupFailure, ownershipFailure);
|
||||
if (ownershipFailure == null) {
|
||||
Throwable runtimeFailure = closeDetachedGenerationRuntimes(null);
|
||||
if (runtimeFailure == null) {
|
||||
runtimeFailure = closeRuntime(engine.runtime, null);
|
||||
cleanupFailure = releaseRuntime(cleanupFailure);
|
||||
if (runtimeReleased) {
|
||||
cleanupFailure = releaseTarget(cleanupFailure);
|
||||
}
|
||||
cleanupFailure = appendFailure(cleanupFailure, runtimeFailure);
|
||||
if (runtimeFailure != null) {
|
||||
if (cleanupFailure != original) {
|
||||
original.addSuppressed(cleanupFailure);
|
||||
}
|
||||
return;
|
||||
if (targetReleased) {
|
||||
cleanupFailure = releaseEngineDataForShutdown(cleanupFailure);
|
||||
}
|
||||
if (engineDataReleased) {
|
||||
cleanupFailure = releasePreservation(cleanupFailure);
|
||||
}
|
||||
if (runtimeReleased && targetReleased && engineDataReleased && preservationReleased) {
|
||||
engine.closed = true;
|
||||
}
|
||||
engine.runtime = null;
|
||||
cleanupFailure = runCleanup(cleanupFailure, engine.publishedTarget::close);
|
||||
cleanupFailure = runCleanup(cleanupFailure, engine.engineDataStore::releaseEngineData);
|
||||
engine.closed = true;
|
||||
cleanupFailure = runCleanup(cleanupFailure, () -> {
|
||||
PreservationRegistry registry = IrisServices.getOrNull(PreservationRegistry.class);
|
||||
if (registry != null) {
|
||||
registry.dereference();
|
||||
}
|
||||
});
|
||||
}
|
||||
if (cleanupFailure != null && cleanupFailure != original) {
|
||||
original.addSuppressed(cleanupFailure);
|
||||
@@ -223,41 +229,55 @@ final class EngineShutdownSequence {
|
||||
if (assembly == null) {
|
||||
return failure;
|
||||
}
|
||||
failure = runCleanup(failure, () -> {
|
||||
if (assembly.worldManager != null) {
|
||||
assembly.worldManager.close();
|
||||
}
|
||||
});
|
||||
failure = runCleanup(failure, () -> {
|
||||
if (assembly.effects != null) {
|
||||
assembly.effects.close();
|
||||
}
|
||||
});
|
||||
failure = runCleanup(failure, () -> {
|
||||
if (assembly.mode != null) {
|
||||
assembly.mode.close();
|
||||
}
|
||||
});
|
||||
Throwable complexFailure = runCleanup(null, () -> {
|
||||
if (assembly.complex != null) {
|
||||
assembly.complex.close();
|
||||
}
|
||||
});
|
||||
failure = appendFailure(failure, complexFailure);
|
||||
if (complexFailure != null) {
|
||||
return failure;
|
||||
incompleteAssemblies.add(assembly);
|
||||
RuntimeAssembly previous = engine == null ? null : engine.runtimeAssembly.get();
|
||||
if (engine != null) {
|
||||
engine.runtimeAssembly.set(assembly);
|
||||
}
|
||||
failure = runCleanup(failure, () -> {
|
||||
if (assembly.hash32 != null) {
|
||||
assembly.hash32.cancel(true);
|
||||
Throwable assemblyFailure;
|
||||
try {
|
||||
assemblyFailure = closeAssemblyResources(assembly);
|
||||
} finally {
|
||||
if (engine != null) {
|
||||
if (previous == null) {
|
||||
engine.runtimeAssembly.remove();
|
||||
} else {
|
||||
engine.runtimeAssembly.set(previous);
|
||||
}
|
||||
}
|
||||
});
|
||||
if (assembly.mantle != null && assembly.ownsMantle) {
|
||||
assembly.ownsMantle = false;
|
||||
failure = runCleanup(failure, assembly.mantle::saveAllNow);
|
||||
failure = runCleanup(failure, assembly.mantle::close);
|
||||
}
|
||||
return failure;
|
||||
if (assemblyFailure == null) {
|
||||
incompleteAssemblies.remove(assembly);
|
||||
} else if (engine != null) {
|
||||
engine.lifecycleState = LifecycleState.FAILED;
|
||||
}
|
||||
return appendFailure(failure, assemblyFailure);
|
||||
}
|
||||
|
||||
void retainUnpublishedRuntime(EngineRuntime runtime) {
|
||||
unpublishedRuntimes.add(runtime);
|
||||
}
|
||||
|
||||
boolean retainsData(IrisData data) {
|
||||
if (engine.runtime != null && engine.runtime.generation().data() == data) {
|
||||
return true;
|
||||
}
|
||||
for (RuntimeAssembly assembly : incompleteAssemblies) {
|
||||
if (assembly.target.getData() == data) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
for (EngineRuntime runtime : unpublishedRuntimes) {
|
||||
if (runtime.generation().data() == data) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
for (EngineTarget target : incompleteTargets.toArray(new EngineTarget[0])) {
|
||||
if (target.getData() == data) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
Throwable closeRuntime(EngineRuntime engineRuntime, Throwable failure) {
|
||||
@@ -272,8 +292,11 @@ final class EngineShutdownSequence {
|
||||
if (engineRuntime == null) {
|
||||
return failure;
|
||||
}
|
||||
failure = runCleanup(failure, engineRuntime.worldManager()::close);
|
||||
failure = runCleanup(failure, engineRuntime.effects()::close);
|
||||
Throwable serviceFailure = runCleanup(null, engineRuntime.worldManager()::close);
|
||||
serviceFailure = runCleanup(serviceFailure, engineRuntime.effects()::close);
|
||||
if (serviceFailure != null) {
|
||||
return appendFailure(failure, serviceFailure);
|
||||
}
|
||||
return closeGenerationRuntime(engineRuntime.generation(), retainedMantle, failure);
|
||||
}
|
||||
|
||||
@@ -289,7 +312,10 @@ final class EngineShutdownSequence {
|
||||
if (generationRuntime == null) {
|
||||
return failure;
|
||||
}
|
||||
failure = runCleanup(failure, generationRuntime.mode()::close);
|
||||
Throwable modeFailure = runCleanup(null, generationRuntime.mode()::close);
|
||||
if (modeFailure != null) {
|
||||
return appendFailure(failure, modeFailure);
|
||||
}
|
||||
Throwable complexFailure = runCleanup(null, generationRuntime.complex()::close);
|
||||
failure = appendFailure(failure, complexFailure);
|
||||
if (complexFailure != null) {
|
||||
@@ -297,8 +323,7 @@ final class EngineShutdownSequence {
|
||||
}
|
||||
failure = runCleanup(failure, () -> generationRuntime.hash32().cancel(true));
|
||||
if (generationRuntime.mantle() != retainedMantle) {
|
||||
failure = runCleanup(failure, generationRuntime.mantle()::saveAllNow);
|
||||
failure = runCleanup(failure, generationRuntime.mantle()::close);
|
||||
failure = appendFailure(failure, closeMantle(generationRuntime.mantle()));
|
||||
}
|
||||
return failure;
|
||||
}
|
||||
@@ -320,6 +345,18 @@ final class EngineShutdownSequence {
|
||||
}
|
||||
|
||||
Throwable closeDetachedTarget(EngineTarget target, Throwable failure) {
|
||||
for (RuntimeAssembly assembly : incompleteAssemblies) {
|
||||
if (assembly.target == target) {
|
||||
return appendFailure(failure,
|
||||
new IllegalStateException("Iris runtime assembly still owns the detached target."));
|
||||
}
|
||||
}
|
||||
for (EngineRuntime runtime : unpublishedRuntimes) {
|
||||
if (runtime.generation().target() == target) {
|
||||
return appendFailure(failure,
|
||||
new IllegalStateException("Iris unpublished runtime still owns the detached target."));
|
||||
}
|
||||
}
|
||||
failure = runCleanup(failure, () -> target.getData().unregisterEngine(engine));
|
||||
failure = runCleanup(failure, target::close);
|
||||
failure = runCleanup(failure, target.getData()::close);
|
||||
@@ -330,6 +367,10 @@ final class EngineShutdownSequence {
|
||||
if (runtimeReleased) {
|
||||
return failure;
|
||||
}
|
||||
Throwable incompleteFailure = closeIncompleteRuntimes();
|
||||
if (incompleteFailure != null) {
|
||||
return appendFailure(failure, incompleteFailure);
|
||||
}
|
||||
Throwable detachedFailure = closeDetachedGenerationRuntimes(null);
|
||||
if (detachedFailure != null) {
|
||||
return appendFailure(failure, detachedFailure);
|
||||
@@ -343,6 +384,86 @@ final class EngineShutdownSequence {
|
||||
return failure;
|
||||
}
|
||||
|
||||
private Throwable closeAssemblyResources(RuntimeAssembly assembly) {
|
||||
Throwable failure = runCleanup(null, () -> {
|
||||
if (assembly.worldManager != null) {
|
||||
assembly.worldManager.close();
|
||||
}
|
||||
});
|
||||
failure = runCleanup(failure, () -> {
|
||||
if (assembly.effects != null) {
|
||||
assembly.effects.close();
|
||||
}
|
||||
});
|
||||
if (failure != null) {
|
||||
return failure;
|
||||
}
|
||||
failure = runCleanup(null, () -> {
|
||||
if (assembly.mode != null) {
|
||||
assembly.mode.close();
|
||||
}
|
||||
});
|
||||
if (failure != null) {
|
||||
return failure;
|
||||
}
|
||||
failure = runCleanup(null, () -> {
|
||||
if (assembly.complex != null) {
|
||||
assembly.complex.close();
|
||||
}
|
||||
});
|
||||
if (failure != null) {
|
||||
return failure;
|
||||
}
|
||||
failure = runCleanup(null, () -> {
|
||||
if (assembly.hash32 != null) {
|
||||
assembly.hash32.cancel(true);
|
||||
}
|
||||
});
|
||||
if (assembly.mantle != null && assembly.ownsMantle) {
|
||||
Throwable mantleFailure = closeMantle(assembly.mantle);
|
||||
if (mantleFailure == null) {
|
||||
assembly.ownsMantle = false;
|
||||
}
|
||||
failure = appendFailure(failure, mantleFailure);
|
||||
}
|
||||
return failure;
|
||||
}
|
||||
|
||||
private Throwable closeIncompleteRuntimes() {
|
||||
Throwable failure = null;
|
||||
for (RuntimeAssembly assembly : incompleteAssemblies.toArray(new RuntimeAssembly[0])) {
|
||||
Throwable assemblyFailure = closeAssembly(assembly, null);
|
||||
if (assemblyFailure == null && assembly.target != engine.publishedTarget) {
|
||||
incompleteTargets.add(assembly.target);
|
||||
}
|
||||
failure = appendFailure(failure, assemblyFailure);
|
||||
}
|
||||
EngineMantle retainedMantle = engine.runtime == null ? null : engine.runtime.generation().mantle();
|
||||
for (EngineRuntime unpublished : unpublishedRuntimes.toArray(new EngineRuntime[0])) {
|
||||
Throwable runtimeFailure = closeRuntime(unpublished, retainedMantle, null);
|
||||
if (runtimeFailure == null) {
|
||||
unpublishedRuntimes.remove(unpublished);
|
||||
if (unpublished.generation().target() != engine.publishedTarget) {
|
||||
incompleteTargets.add(unpublished.generation().target());
|
||||
}
|
||||
}
|
||||
failure = appendFailure(failure, runtimeFailure);
|
||||
}
|
||||
for (EngineTarget target : incompleteTargets.toArray(new EngineTarget[0])) {
|
||||
Throwable targetFailure = closeDetachedTarget(target, null);
|
||||
if (targetFailure == null) {
|
||||
incompleteTargets.remove(target);
|
||||
}
|
||||
failure = appendFailure(failure, targetFailure);
|
||||
}
|
||||
return failure;
|
||||
}
|
||||
|
||||
private static Throwable closeMantle(EngineMantle mantle) {
|
||||
Throwable failure = runCleanup(null, mantle::saveAllNow);
|
||||
return failure == null ? runCleanup(null, mantle::close) : failure;
|
||||
}
|
||||
|
||||
private Throwable closeDetachedGenerationRuntimes(Throwable failure) {
|
||||
GenerationRuntime[] detachedRuntimes = engine.detachedGenerationRuntimes.toArray(new GenerationRuntime[0]);
|
||||
for (GenerationRuntime detached : detachedRuntimes) {
|
||||
|
||||
@@ -22,12 +22,13 @@ import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.ForkJoinPool;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.RejectedExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
|
||||
public interface MatterGenerator {
|
||||
MultiBurst DISPATCHER = MultiBurst.burst;
|
||||
ConcurrentHashMap<MatterTaskKey, CompletableFuture<Void>> IN_FLIGHT_COMPONENTS = new ConcurrentHashMap<>();
|
||||
ConcurrentHashMap<MatterTaskKey, MatterComponentTask> IN_FLIGHT_COMPONENTS = new ConcurrentHashMap<>();
|
||||
long COMPONENT_TASK_POLL_MS = 1000L;
|
||||
long COMPONENT_TASK_TIMEOUT_MS = Long.getLong("iris.mantle.componentTimeout", 120000L);
|
||||
|
||||
@@ -368,38 +369,37 @@ public interface MatterGenerator {
|
||||
) {
|
||||
MantleFlag flag = component.getFlag();
|
||||
MatterTaskKey key = new MatterTaskKey(getMantle(), chunkX, chunkZ, flag);
|
||||
MatterComponentTask task = new MatterComponentTask(key);
|
||||
if (chunk.isFlagged(flag)) {
|
||||
return new MatterComponentTask(key, CompletableFuture.completedFuture(null), null);
|
||||
task.finish(null);
|
||||
return task;
|
||||
}
|
||||
|
||||
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||
CompletableFuture<Void> existing = IN_FLIGHT_COMPONENTS.putIfAbsent(key, future);
|
||||
MatterComponentTask existing = IN_FLIGHT_COMPONENTS.putIfAbsent(key, task);
|
||||
if (existing != null) {
|
||||
return new MatterComponentTask(key, existing, null);
|
||||
return existing;
|
||||
}
|
||||
|
||||
try {
|
||||
if (DISPATCHER.ownsCurrentThread()) {
|
||||
// A pool thread runs its claim inline: a task queued behind pool threads that block
|
||||
// on unmanaged waits (structure builds, cache locks) could starve the whole pool.
|
||||
completeComponentTask(future, key, chunk, component, writer, chunkX, chunkZ, context, IrisContext.get());
|
||||
return new MatterComponentTask(key, future, null);
|
||||
completeComponentTask(task, chunk, component, writer, chunkX, chunkZ, context, IrisContext.get());
|
||||
return task;
|
||||
}
|
||||
IrisContext callerContext = IrisContext.get();
|
||||
Future<?> submission = DISPATCHER.submit(() -> completeComponentTask(
|
||||
future,
|
||||
key,
|
||||
task.setSubmission(DISPATCHER.submit(() -> completeComponentTask(
|
||||
task,
|
||||
chunk,
|
||||
component,
|
||||
writer,
|
||||
chunkX,
|
||||
chunkZ,
|
||||
context,
|
||||
callerContext));
|
||||
return new MatterComponentTask(key, future, submission);
|
||||
callerContext)));
|
||||
return task;
|
||||
} catch (Throwable throwable) {
|
||||
IN_FLIGHT_COMPONENTS.remove(key, future);
|
||||
future.completeExceptionally(throwable);
|
||||
task.cancel(throwable);
|
||||
throw throwable;
|
||||
}
|
||||
}
|
||||
@@ -449,7 +449,7 @@ public interface MatterGenerator {
|
||||
}
|
||||
|
||||
private static void awaitComponentTask(MatterComponentTask task) {
|
||||
CompletableFuture<Void> future = task.future();
|
||||
CompletableFuture<Void> future = task.future;
|
||||
if (future.isDone() && !future.isCompletedExceptionally()) {
|
||||
return;
|
||||
}
|
||||
@@ -459,11 +459,17 @@ public interface MatterGenerator {
|
||||
try {
|
||||
ForkJoinPool.managedBlock(new ComponentTaskBlocker(task));
|
||||
} catch (InterruptedException interruption) {
|
||||
Thread.currentThread().interrupt();
|
||||
try {
|
||||
task.await(COMPONENT_TASK_TIMEOUT_MS);
|
||||
} finally {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
} catch (RejectedExecutionException exhaustedPool) {
|
||||
task.await(COMPONENT_TASK_TIMEOUT_MS);
|
||||
}
|
||||
return;
|
||||
}
|
||||
awaitComponentTaskBlocking(task);
|
||||
task.await(COMPONENT_TASK_TIMEOUT_MS);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -478,70 +484,22 @@ public interface MatterGenerator {
|
||||
|
||||
@Override
|
||||
public boolean block() {
|
||||
awaitComponentTaskBlocking(task);
|
||||
task.await(COMPONENT_TASK_TIMEOUT_MS);
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isReleasable() {
|
||||
if (!task.future().isDone()) {
|
||||
if (!task.future.isDone()) {
|
||||
return false;
|
||||
}
|
||||
task.future().join();
|
||||
task.future.join();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
private static void awaitComponentTaskBlocking(MatterComponentTask task) {
|
||||
CompletableFuture<Void> future = task.future();
|
||||
long start = System.currentTimeMillis();
|
||||
boolean interrupted = false;
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
try {
|
||||
future.get(COMPONENT_TASK_POLL_MS, TimeUnit.MILLISECONDS);
|
||||
return;
|
||||
} catch (InterruptedException interruption) {
|
||||
// The writer cannot close while this task still writes through it, so the wait is
|
||||
// uninterruptible and the flag is restored on the way out.
|
||||
interrupted = true;
|
||||
} catch (TimeoutException timeout) {
|
||||
Future<?> submission = task.submission();
|
||||
boolean dropped = submission != null && submission.isDone();
|
||||
if (future.isDone()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
long waited = System.currentTimeMillis() - start;
|
||||
if (!dropped && waited < COMPONENT_TASK_TIMEOUT_MS) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// The task can no longer complete (the dispatcher dropped it) or is wedged. Drop
|
||||
// the dedup entry so later generations do not inherit a future that never
|
||||
// completes, and release anything else already waiting on it.
|
||||
IllegalStateException failure = new IllegalStateException("Mantle component " + task.key()
|
||||
+ (dropped ? " was dropped by the dispatcher" : " did not complete in " + waited + "ms"));
|
||||
IN_FLIGHT_COMPONENTS.remove(task.key(), future);
|
||||
future.completeExceptionally(failure);
|
||||
IrisLogging.error(failure.getMessage());
|
||||
throw new CompletionException(failure);
|
||||
} catch (ExecutionException executionFailure) {
|
||||
Throwable cause = executionFailure.getCause();
|
||||
throw new CompletionException(cause != null ? cause : executionFailure);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (interrupted) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void completeComponentTask(
|
||||
CompletableFuture<Void> future,
|
||||
MatterTaskKey key,
|
||||
MatterComponentTask task,
|
||||
MantleChunk<Matter> chunk,
|
||||
MantleComponent component,
|
||||
MantleWriter writer,
|
||||
@@ -550,14 +508,17 @@ public interface MatterGenerator {
|
||||
ChunkContext context,
|
||||
IrisContext callerContext
|
||||
) {
|
||||
if (!task.start()) {
|
||||
return;
|
||||
}
|
||||
Throwable failure = null;
|
||||
try {
|
||||
runComponentWithContext(chunk, component, writer, chunkX, chunkZ, context, callerContext);
|
||||
future.complete(null);
|
||||
} catch (Throwable throwable) {
|
||||
future.completeExceptionally(throwable);
|
||||
failure = throwable;
|
||||
throw throwable;
|
||||
} finally {
|
||||
IN_FLIGHT_COMPONENTS.remove(key, future);
|
||||
task.finish(failure);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -597,11 +558,98 @@ public interface MatterGenerator {
|
||||
));
|
||||
}
|
||||
|
||||
/**
|
||||
* A launched component task. {@code submission} is the dispatcher handle when this generation
|
||||
* owns the task, and null when the future belongs to another generation or already completed.
|
||||
*/
|
||||
record MatterComponentTask(MatterTaskKey key, CompletableFuture<Void> future, Future<?> submission) {
|
||||
final class MatterComponentTask {
|
||||
private final MatterTaskKey key;
|
||||
private final CompletableFuture<Void> future = new CompletableFuture<>();
|
||||
private volatile Future<?> submission;
|
||||
private volatile Throwable failure;
|
||||
private boolean started;
|
||||
|
||||
MatterComponentTask(MatterTaskKey key) {
|
||||
this.key = key;
|
||||
}
|
||||
|
||||
synchronized boolean start() {
|
||||
if (started || future.isDone()) {
|
||||
return false;
|
||||
}
|
||||
started = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
synchronized void setSubmission(Future<?> submission) {
|
||||
this.submission = submission;
|
||||
if (failure != null) {
|
||||
submission.cancel(true);
|
||||
}
|
||||
}
|
||||
|
||||
synchronized boolean cancel(Throwable cause) {
|
||||
if (future.isDone() || failure != null) {
|
||||
return false;
|
||||
}
|
||||
failure = cause;
|
||||
if (!started) {
|
||||
finish(null);
|
||||
}
|
||||
if (submission != null) {
|
||||
submission.cancel(true);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
synchronized void finish(Throwable cause) {
|
||||
if (failure == null) {
|
||||
failure = cause;
|
||||
} else if (cause != null && cause != failure) {
|
||||
failure.addSuppressed(cause);
|
||||
}
|
||||
IN_FLIGHT_COMPONENTS.remove(key, this);
|
||||
if (failure == null) {
|
||||
future.complete(null);
|
||||
} else {
|
||||
future.completeExceptionally(failure);
|
||||
}
|
||||
}
|
||||
|
||||
void await(long timeoutMillis) {
|
||||
long start = System.nanoTime();
|
||||
boolean interrupted = false;
|
||||
try {
|
||||
while (true) {
|
||||
try {
|
||||
future.get(COMPONENT_TASK_POLL_MS, TimeUnit.MILLISECONDS);
|
||||
return;
|
||||
} catch (InterruptedException interruption) {
|
||||
// The writer cannot close while this task still writes through it, so the wait is
|
||||
// uninterruptible and the flag is restored on the way out.
|
||||
interrupted = true;
|
||||
} catch (TimeoutException timeout) {
|
||||
if (future.isDone() || failure != null) {
|
||||
continue;
|
||||
}
|
||||
Future<?> submitted = submission;
|
||||
boolean dropped = submitted != null && submitted.isDone();
|
||||
long waited = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
|
||||
if (!dropped && waited < timeoutMillis) {
|
||||
continue;
|
||||
}
|
||||
IllegalStateException timeoutFailure = new IllegalStateException("Mantle component " + key
|
||||
+ (dropped ? " was dropped by the dispatcher" : " did not complete in " + waited + "ms"));
|
||||
if (cancel(timeoutFailure)) {
|
||||
IrisLogging.error(timeoutFailure.getMessage());
|
||||
}
|
||||
} catch (ExecutionException executionFailure) {
|
||||
Throwable cause = executionFailure.getCause();
|
||||
throw new CompletionException(cause != null ? cause : executionFailure);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (interrupted) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
record MatterPassPlan(MantlePass pass, int passChunkRadius, int downstreamBlockRadius) {
|
||||
|
||||
@@ -33,9 +33,9 @@ public class EngineRuntimePublicationContractTest {
|
||||
int publishEnd = source.indexOf("private void scheduleRuntimeTasks", publishStart);
|
||||
String publish = source.substring(publishStart, publishEnd);
|
||||
|
||||
assertBefore(publish, "engine.getClosing().set(true)", "closeRuntime(next, e)");
|
||||
assertBefore(publish, "closeBackgroundTaskAdmission()", "closeRuntime(next, e)");
|
||||
assertBefore(publish, "sealAndAwait(", "closeRuntime(next, e)");
|
||||
assertBefore(publish, "engine.getClosing().set(true)", "cleanupFailure = engine.shutdownSequence.closeRuntime(next, null)");
|
||||
assertBefore(publish, "closeBackgroundTaskAdmission()", "cleanupFailure = engine.shutdownSequence.closeRuntime(next, null)");
|
||||
assertBefore(publish, "sealAndAwait(", "cleanupFailure = engine.shutdownSequence.closeRuntime(next, null)");
|
||||
}
|
||||
|
||||
@Test
|
||||
|
||||
@@ -1,6 +1,15 @@
|
||||
package art.arcane.iris.engine;
|
||||
|
||||
import art.arcane.iris.core.loader.IrisData;
|
||||
import art.arcane.iris.engine.EngineRuntimeBuilder.RuntimeAssembly;
|
||||
import art.arcane.iris.engine.framework.EngineEffects;
|
||||
import art.arcane.iris.engine.framework.EnginePlatformHooks;
|
||||
import art.arcane.iris.engine.framework.EngineTarget;
|
||||
import art.arcane.iris.engine.framework.EngineWorldManager;
|
||||
import art.arcane.iris.engine.framework.GenerationSessionException;
|
||||
import art.arcane.iris.engine.framework.GenerationSessionManager;
|
||||
import art.arcane.iris.engine.framework.NativeStructureOwnershipStore;
|
||||
import art.arcane.iris.engine.object.IrisWorld;
|
||||
import art.arcane.iris.engine.framework.EngineMode;
|
||||
import art.arcane.iris.engine.mantle.EngineMantle;
|
||||
import art.arcane.iris.spi.IrisPlatform;
|
||||
@@ -11,17 +20,28 @@ import org.junit.AfterClass;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
import org.mockito.InOrder;
|
||||
import org.mockito.MockedStatic;
|
||||
|
||||
import java.lang.reflect.Field;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertSame;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.junit.Assert.assertThrows;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.doNothing;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
import static org.mockito.Mockito.inOrder;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.mockStatic;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verifyNoInteractions;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
@@ -99,4 +119,238 @@ public class EngineShutdownDrainTest {
|
||||
assertTrue(assembly.ownsMantle);
|
||||
assertFalse(assembly.hash32.isCancelled());
|
||||
}
|
||||
@Test
|
||||
public void failedAssemblySaveKeepsOwnershipAndDoesNotCloseStorage() {
|
||||
RuntimeAssembly assembly = mock(RuntimeAssembly.class);
|
||||
assembly.mantle = mock(EngineMantle.class);
|
||||
assembly.ownsMantle = true;
|
||||
IllegalStateException failure = new IllegalStateException("Save failed");
|
||||
doThrow(failure).doNothing().when(assembly.mantle).saveAllNow();
|
||||
EngineShutdownSequence shutdown = new EngineShutdownSequence(null);
|
||||
|
||||
assertSame(failure, shutdown.closeAssembly(assembly, null));
|
||||
assertTrue(assembly.ownsMantle);
|
||||
verify(assembly.mantle, never()).close();
|
||||
assertNull(shutdown.closeAssembly(assembly, null));
|
||||
assertFalse(assembly.ownsMantle);
|
||||
assertNull(shutdown.closeAssembly(assembly, null));
|
||||
verify(assembly.mantle, times(2)).saveAllNow();
|
||||
verify(assembly.mantle).close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void failedAssemblyCloseKeepsOwnershipUntilRetrySucceeds() {
|
||||
RuntimeAssembly assembly = mock(RuntimeAssembly.class);
|
||||
assembly.mantle = mock(EngineMantle.class);
|
||||
assembly.ownsMantle = true;
|
||||
IllegalStateException failure = new IllegalStateException("Close failed");
|
||||
doThrow(failure).doNothing().when(assembly.mantle).close();
|
||||
EngineShutdownSequence shutdown = new EngineShutdownSequence(null);
|
||||
|
||||
assertSame(failure, shutdown.closeAssembly(assembly, null));
|
||||
assertTrue(assembly.ownsMantle);
|
||||
assertNull(shutdown.closeAssembly(assembly, null));
|
||||
assertFalse(assembly.ownsMantle);
|
||||
verify(assembly.mantle, times(2)).saveAllNow();
|
||||
verify(assembly.mantle, times(2)).close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void failedPartialAssemblyRetainsDataAndRetriesInsideItsAssemblyScope() throws Exception {
|
||||
ShutdownFixture fixture = new ShutdownFixture();
|
||||
RuntimeAssembly assembly = mock(RuntimeAssembly.class);
|
||||
EngineTarget detachedTarget = target();
|
||||
setField(assembly, "target", detachedTarget);
|
||||
assembly.complex = mock(IrisComplex.class);
|
||||
assembly.mantle = mock(EngineMantle.class);
|
||||
assembly.ownsMantle = true;
|
||||
RuntimeAssembly previous = mock(RuntimeAssembly.class);
|
||||
fixture.engine.runtimeAssembly.set(previous);
|
||||
IllegalStateException failure = new IllegalStateException("Planner still active");
|
||||
doAnswer(invocation -> {
|
||||
assertSame(assembly, fixture.engine.runtimeAssembly.get());
|
||||
throw failure;
|
||||
}).when(assembly.complex).close();
|
||||
|
||||
assertSame(failure, fixture.shutdown.closeAssembly(assembly, null));
|
||||
assertSame(previous, fixture.engine.runtimeAssembly.get());
|
||||
fixture.engine.runtimeAssembly.remove();
|
||||
assertTrue(fixture.shutdown.retainsData(detachedTarget.getData()));
|
||||
assertTrue(fixture.shutdown.closeDetachedTarget(detachedTarget, null) instanceof IllegalStateException);
|
||||
verify(detachedTarget, never()).close();
|
||||
try (MockedStatic<NativeStructureOwnershipStore> ignored = mockStatic(NativeStructureOwnershipStore.class)) {
|
||||
assertThrows(IllegalStateException.class, fixture.shutdown::close);
|
||||
verify(fixture.target, never()).close();
|
||||
verifyNoInteractions(assembly.mantle);
|
||||
doAnswer(invocation -> {
|
||||
assertSame(assembly, fixture.engine.runtimeAssembly.get());
|
||||
return null;
|
||||
}).when(assembly.complex).close();
|
||||
fixture.shutdown.close();
|
||||
}
|
||||
|
||||
assertNull(fixture.engine.runtimeAssembly.get());
|
||||
assertFalse(assembly.ownsMantle);
|
||||
assertFalse(fixture.shutdown.retainsData(detachedTarget.getData()));
|
||||
verify(assembly.mantle).close();
|
||||
verify(detachedTarget).close();
|
||||
verify(detachedTarget.getData()).close();
|
||||
assertTrue(fixture.engine.closed);
|
||||
assertNull(fixture.engine.runtime);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void failedManagerStartKeepsPublishedRuntimeWhenCleanupFails() throws Exception {
|
||||
ShutdownFixture fixture = new ShutdownFixture();
|
||||
fixture.engine.runtime = null;
|
||||
IllegalStateException startFailure = new IllegalStateException("Manager start failed");
|
||||
IllegalStateException closeFailure = new IllegalStateException("Planner still active");
|
||||
doThrow(startFailure).when(fixture.manager).start();
|
||||
doThrow(closeFailure).when(fixture.complex).close();
|
||||
|
||||
IllegalStateException failure = assertThrows(IllegalStateException.class,
|
||||
() -> new EngineRuntimeBuilder(fixture.engine).publishRuntime(fixture.runtime, null));
|
||||
|
||||
assertSame(startFailure, failure.getCause());
|
||||
assertSame(closeFailure, startFailure.getSuppressed()[0]);
|
||||
assertSame(fixture.runtime, fixture.engine.runtime);
|
||||
assertTrue(fixture.shutdown.retainsData(fixture.target.getData()));
|
||||
verify(fixture.mantle, never()).close();
|
||||
doNothing().when(fixture.complex).close();
|
||||
try (MockedStatic<NativeStructureOwnershipStore> ignored = mockStatic(NativeStructureOwnershipStore.class)) {
|
||||
fixture.shutdown.close();
|
||||
}
|
||||
assertNull(fixture.engine.runtime);
|
||||
assertTrue(fixture.engine.closed);
|
||||
verify(fixture.target).close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void failedStartDrainRetainsResourcesUntilGenerationStops() throws Exception {
|
||||
ShutdownFixture fixture = new ShutdownFixture();
|
||||
fixture.engine.runtime = null;
|
||||
doThrow(new IllegalStateException("Manager start failed")).when(fixture.manager).start();
|
||||
doThrow(new GenerationSessionException("Generation still active", false)).when(fixture.sessions)
|
||||
.sealAndAwait("failed world manager start", IrisEngine.SESSION_DRAIN_TIMEOUT_MILLIS, true);
|
||||
|
||||
assertThrows(IllegalStateException.class,
|
||||
() -> new EngineRuntimeBuilder(fixture.engine).publishRuntime(fixture.runtime, null));
|
||||
|
||||
assertSame(fixture.runtime, fixture.engine.runtime);
|
||||
verify(fixture.manager, never()).close();
|
||||
verifyNoInteractions(fixture.complex, fixture.mantle);
|
||||
try (MockedStatic<NativeStructureOwnershipStore> ignored = mockStatic(NativeStructureOwnershipStore.class)) {
|
||||
fixture.shutdown.close();
|
||||
}
|
||||
assertNull(fixture.engine.runtime);
|
||||
assertTrue(fixture.engine.closed);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void failedRetirementRetainsUnpublishedRuntimeWithoutClosingSharedMantle() throws Exception {
|
||||
ShutdownFixture fixture = new ShutdownFixture();
|
||||
EngineTarget nextTarget = target();
|
||||
GenerationRuntime nextGeneration = generation(nextTarget, fixture.mantle);
|
||||
EngineRuntime next = new EngineRuntime(nextGeneration, mock(EngineEffects.class), mock(EngineWorldManager.class));
|
||||
EngineMode nextMode = nextGeneration.mode();
|
||||
doThrow(new IllegalStateException("Previous planner active")).when(fixture.complex).close();
|
||||
doThrow(new IllegalStateException("Next mode active")).when(nextMode).close();
|
||||
|
||||
assertThrows(IllegalStateException.class,
|
||||
() -> new EngineRuntimeBuilder(fixture.engine).publishRuntime(next, fixture.runtime));
|
||||
|
||||
assertSame(fixture.runtime, fixture.engine.runtime);
|
||||
assertTrue(fixture.shutdown.retainsData(nextTarget.getData()));
|
||||
verifyNoInteractions(fixture.mantle);
|
||||
doNothing().when(fixture.complex).close();
|
||||
doNothing().when(nextMode).close();
|
||||
try (MockedStatic<NativeStructureOwnershipStore> ignored = mockStatic(NativeStructureOwnershipStore.class)) {
|
||||
fixture.shutdown.close();
|
||||
}
|
||||
assertFalse(fixture.shutdown.retainsData(nextTarget.getData()));
|
||||
verify(nextTarget).close();
|
||||
verify(fixture.target).close();
|
||||
verify(fixture.mantle).saveAllNow();
|
||||
verify(fixture.mantle).close();
|
||||
assertTrue(fixture.engine.closed);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void failedConstructionDrainPreventsAllRuntimeAndTargetRelease() throws Exception {
|
||||
ShutdownFixture fixture = new ShutdownFixture();
|
||||
doThrow(new GenerationSessionException("Generation still active", false)).doNothing().when(fixture.sessions)
|
||||
.sealAndAwait("failed initialization", 0L, true);
|
||||
IllegalStateException original = new IllegalStateException("Construction failed");
|
||||
try (MockedStatic<NativeStructureOwnershipStore> ignored = mockStatic(NativeStructureOwnershipStore.class)) {
|
||||
fixture.shutdown.cleanupFailedConstruction(original);
|
||||
assertSame(fixture.runtime, fixture.engine.runtime);
|
||||
assertFalse(fixture.engine.closed);
|
||||
verifyNoInteractions(fixture.complex, fixture.mantle);
|
||||
verify(fixture.target, never()).close();
|
||||
fixture.shutdown.cleanupFailedConstruction(original);
|
||||
}
|
||||
assertNull(fixture.engine.runtime);
|
||||
assertTrue(fixture.engine.closed);
|
||||
verify(fixture.mantle).close();
|
||||
verify(fixture.target).close();
|
||||
}
|
||||
|
||||
private static EngineTarget target() {
|
||||
EngineTarget target = mock(EngineTarget.class);
|
||||
when(target.getData()).thenReturn(mock(IrisData.class));
|
||||
when(target.getWorld()).thenReturn(mock(IrisWorld.class));
|
||||
return target;
|
||||
}
|
||||
|
||||
private static GenerationRuntime generation(EngineTarget target, EngineMantle mantle) {
|
||||
GenerationRuntime generation = mock(GenerationRuntime.class);
|
||||
IrisData data = target.getData();
|
||||
when(generation.target()).thenReturn(target);
|
||||
when(generation.data()).thenReturn(data);
|
||||
when(generation.mantle()).thenReturn(mantle);
|
||||
when(generation.complex()).thenReturn(mock(IrisComplex.class));
|
||||
when(generation.mode()).thenReturn(mock(EngineMode.class));
|
||||
when(generation.hash32()).thenReturn(new CompletableFuture<>());
|
||||
return generation;
|
||||
}
|
||||
|
||||
private static void setField(Object target, String name, Object value) throws Exception {
|
||||
Field field = target.getClass().getDeclaredField(name);
|
||||
field.setAccessible(true);
|
||||
field.set(target, value);
|
||||
}
|
||||
|
||||
private static final class ShutdownFixture {
|
||||
private final IrisEngine engine = mock(IrisEngine.class);
|
||||
private final EngineTarget target = target();
|
||||
private final EngineMantle mantle = mock(EngineMantle.class);
|
||||
private final GenerationRuntime generation = generation(target, mantle);
|
||||
private final IrisComplex complex = generation.complex();
|
||||
private final EngineWorldManager manager = mock(EngineWorldManager.class);
|
||||
private final EngineRuntime runtime = new EngineRuntime(generation, mock(EngineEffects.class), manager);
|
||||
private final GenerationSessionManager sessions = mock(GenerationSessionManager.class);
|
||||
private final EngineShutdownSequence shutdown = new EngineShutdownSequence(engine);
|
||||
|
||||
private ShutdownFixture() throws Exception {
|
||||
EngineBackgroundTasks background = mock(EngineBackgroundTasks.class);
|
||||
when(background.drainBackgroundTasks(anyString()))
|
||||
.thenReturn(new EngineBackgroundTasks.BackgroundTaskDrain(null, true));
|
||||
setField(engine, "backgroundTasks", background);
|
||||
setField(engine, "shutdownSequence", shutdown);
|
||||
setField(engine, "engineDataStore", mock(EngineDataStore.class));
|
||||
setField(engine, "lifecycleLock", new Object());
|
||||
setField(engine, "runtimeAssembly", new ThreadLocal<RuntimeAssembly>());
|
||||
setField(engine, "detachedGenerationRuntimes", ConcurrentHashMap.newKeySet());
|
||||
when(engine.getClosing()).thenReturn(new AtomicBoolean());
|
||||
when(engine.getGenerationSessions()).thenReturn(sessions);
|
||||
when(engine.getPlatformHooks()).thenReturn(mock(EnginePlatformHooks.class));
|
||||
when(engine.getMantle()).thenReturn(mantle);
|
||||
IrisWorld world = target.getWorld();
|
||||
when(engine.getWorld()).thenReturn(world);
|
||||
when(engine.beginShutdown()).thenReturn(true);
|
||||
engine.runtime = runtime;
|
||||
engine.publishedTarget = target;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -43,14 +43,14 @@ public class EngineShutdownOwnershipContractTest {
|
||||
String cleanup = source.substring(cleanupStart, cleanupEnd);
|
||||
int ownershipClose = cleanup.indexOf("NativeStructureOwnershipStore.close(engine)");
|
||||
int ownershipGate = cleanup.indexOf("if (ownershipFailure == null)", ownershipClose);
|
||||
int detachedRelease = cleanup.indexOf("closeDetachedGenerationRuntimes(null)", ownershipGate);
|
||||
int runtimeRelease = cleanup.indexOf("closeRuntime(engine.runtime, null)", ownershipGate);
|
||||
int runtimeRelease = cleanup.indexOf("releaseRuntime(cleanupFailure)", ownershipGate);
|
||||
int targetRelease = cleanup.indexOf("releaseTarget(cleanupFailure)", runtimeRelease);
|
||||
int closedPublication = cleanup.indexOf("engine.closed = true", ownershipGate);
|
||||
|
||||
assertTrue(ownershipClose >= 0);
|
||||
assertTrue(ownershipGate > ownershipClose);
|
||||
assertTrue(detachedRelease > ownershipGate);
|
||||
assertTrue(runtimeRelease > detachedRelease);
|
||||
assertTrue(runtimeRelease > ownershipGate);
|
||||
assertTrue(targetRelease > runtimeRelease);
|
||||
assertTrue(closedPublication > ownershipGate);
|
||||
}
|
||||
|
||||
|
||||
+142
-7
@@ -27,6 +27,8 @@ import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.CompletionException;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
@@ -44,6 +46,9 @@ import static org.mockito.ArgumentMatchers.anyInt;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
public class MatterGeneratorConcurrencyTest {
|
||||
@@ -221,23 +226,153 @@ public class MatterGeneratorConcurrencyTest {
|
||||
@Test
|
||||
public void dispatcherPropagatesAnAlreadyFailedSharedComponent() {
|
||||
IllegalStateException failure = new IllegalStateException("Shared component failed");
|
||||
ExecutionException observed = sharedComponentFailure(CompletableFuture.failedFuture(failure), ExecutionException.class);
|
||||
ExecutionException observed = sharedComponentFailure(failure, ExecutionException.class);
|
||||
assertSame(failure, observed.getCause());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void dispatcherPropagatesAnAlreadyCancelledSharedComponent() {
|
||||
CompletableFuture<Void> cancelled = new CompletableFuture<>();
|
||||
cancelled.cancel(false);
|
||||
sharedComponentFailure(cancelled, CancellationException.class);
|
||||
sharedComponentFailure(new CancellationException("Shared component cancelled"), CancellationException.class);
|
||||
}
|
||||
|
||||
private <T extends Throwable> T sharedComponentFailure(CompletableFuture<Void> completion, Class<T> expectedFailure) {
|
||||
@Test
|
||||
public void timedOutRunningComponentRetainsWriterAndSharedClaimUntilItExits() throws Exception {
|
||||
GeneratorFixture fixture = new GeneratorFixture(true);
|
||||
CountDownLatch started = new CountDownLatch(1);
|
||||
CountDownLatch release = new CountDownLatch(1);
|
||||
CountDownLatch writersAcquired = new CountDownLatch(2);
|
||||
CountDownLatch cancelled = new CountDownLatch(1);
|
||||
AtomicInteger runs = new AtomicInteger();
|
||||
MantleChunk<Matter> chunk = fixture.mantle.getChunk(0, 0);
|
||||
when(chunk.use()).thenAnswer(invocation -> {
|
||||
writersAcquired.countDown();
|
||||
return chunk;
|
||||
});
|
||||
RecordingComponent component = new RecordingComponent(ReservedFlag.OBJECT, 0, 0) {
|
||||
@Override
|
||||
public void generateLayer(MantleWriter writer, int x, int z, ChunkContext context) {
|
||||
runs.incrementAndGet();
|
||||
started.countDown();
|
||||
await(release);
|
||||
verify(chunk, never()).release();
|
||||
}
|
||||
};
|
||||
TestMatterGenerator generator = fixture.generator(List.of(new MantlePass(List.of(component), 0, 0)));
|
||||
CompletableFuture<Void> first = CompletableFuture.runAsync(
|
||||
() -> generator.generateMatter(0, 0, true, fixture.context));
|
||||
await(started);
|
||||
MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT);
|
||||
MatterGenerator.MatterComponentTask task = MatterGenerator.IN_FLIGHT_COMPONENTS.get(key);
|
||||
Future<?> submission = mock(Future.class);
|
||||
when(submission.cancel(true)).thenAnswer(invocation -> {
|
||||
cancelled.countDown();
|
||||
return true;
|
||||
});
|
||||
task.setSubmission(submission);
|
||||
CompletableFuture<Void> second = MultiBurst.burst.completeValueAsync(() -> {
|
||||
generator.generateMatter(0, 0, true, fixture.context);
|
||||
return null;
|
||||
});
|
||||
CompletableFuture<Void> timeout = CompletableFuture.runAsync(() -> task.await(0L));
|
||||
try {
|
||||
await(writersAcquired);
|
||||
await(cancelled);
|
||||
assertSame(task, MatterGenerator.IN_FLIGHT_COMPONENTS.get(key));
|
||||
assertFalse(first.isDone());
|
||||
assertFalse(second.isDone());
|
||||
assertFalse(timeout.isDone());
|
||||
verify(chunk, never()).release();
|
||||
assertEquals(1, runs.get());
|
||||
} finally {
|
||||
release.countDown();
|
||||
}
|
||||
assertThrows(ExecutionException.class, () -> timeout.get(5L, TimeUnit.SECONDS));
|
||||
assertThrows(ExecutionException.class, () -> first.get(5L, TimeUnit.SECONDS));
|
||||
assertThrows(ExecutionException.class, () -> second.get(5L, TimeUnit.SECONDS));
|
||||
verify(chunk, times(2)).release();
|
||||
assertFalse(MatterGenerator.IN_FLIGHT_COMPONENTS.containsKey(key));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void queuedTimeoutPreventsLateExecutionAndAllowsANewClaim() {
|
||||
GeneratorFixture fixture = new GeneratorFixture();
|
||||
MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT);
|
||||
MatterGenerator.MatterComponentTask task = new MatterGenerator.MatterComponentTask(key);
|
||||
MatterGenerator.IN_FLIGHT_COMPONENTS.put(key, task);
|
||||
Future<?> submission = mock(Future.class);
|
||||
task.setSubmission(submission);
|
||||
|
||||
CompletionException failure = assertThrows(CompletionException.class, () -> task.await(0L));
|
||||
|
||||
assertTrue(failure.getCause().getMessage().contains("did not complete"));
|
||||
assertFalse(task.start());
|
||||
assertFalse(MatterGenerator.IN_FLIGHT_COMPONENTS.containsKey(key));
|
||||
verify(submission).cancel(true);
|
||||
MatterGenerator.MatterComponentTask replacement = new MatterGenerator.MatterComponentTask(key);
|
||||
assertNull(MatterGenerator.IN_FLIGHT_COMPONENTS.putIfAbsent(key, replacement));
|
||||
task.finish(null);
|
||||
assertSame(replacement, MatterGenerator.IN_FLIGHT_COMPONENTS.get(key));
|
||||
replacement.finish(null);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void droppedQueuedSubmissionCannotRunAfterTheBarrier() {
|
||||
GeneratorFixture fixture = new GeneratorFixture();
|
||||
MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT);
|
||||
MatterGenerator.MatterComponentTask task = new MatterGenerator.MatterComponentTask(key);
|
||||
MatterGenerator.IN_FLIGHT_COMPONENTS.put(key, task);
|
||||
Future<?> submission = mock(Future.class);
|
||||
when(submission.isDone()).thenReturn(true);
|
||||
task.setSubmission(submission);
|
||||
|
||||
CompletionException failure = assertThrows(CompletionException.class,
|
||||
() -> task.await(MatterGenerator.COMPONENT_TASK_TIMEOUT_MS));
|
||||
|
||||
assertTrue(failure.getCause().getMessage().contains("was dropped"));
|
||||
assertFalse(task.start());
|
||||
assertFalse(MatterGenerator.IN_FLIGHT_COMPONENTS.containsKey(key));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void interruptedWaiterDrainsRunningTaskAndRestoresInterrupt() throws Exception {
|
||||
GeneratorFixture fixture = new GeneratorFixture();
|
||||
MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT);
|
||||
MatterGenerator.MatterComponentTask task = new MatterGenerator.MatterComponentTask(key);
|
||||
assertTrue(task.start());
|
||||
CountDownLatch waiting = new CountDownLatch(1);
|
||||
AtomicBoolean restored = new AtomicBoolean();
|
||||
CompletableFuture<Void> completion = new CompletableFuture<>();
|
||||
Thread waiter = new Thread(() -> {
|
||||
try {
|
||||
Thread.currentThread().interrupt();
|
||||
waiting.countDown();
|
||||
task.await(MatterGenerator.COMPONENT_TASK_TIMEOUT_MS);
|
||||
restored.set(Thread.currentThread().isInterrupted());
|
||||
completion.complete(null);
|
||||
} catch (Throwable failure) {
|
||||
completion.completeExceptionally(failure);
|
||||
}
|
||||
});
|
||||
waiter.start();
|
||||
await(waiting);
|
||||
try {
|
||||
assertFalse(completion.isDone());
|
||||
} finally {
|
||||
task.finish(null);
|
||||
}
|
||||
completion.get(5L, TimeUnit.SECONDS);
|
||||
waiter.join(5_000L);
|
||||
assertTrue(restored.get());
|
||||
}
|
||||
|
||||
private <T extends Throwable> T sharedComponentFailure(Throwable failure, Class<T> expectedFailure) {
|
||||
GeneratorFixture fixture = new GeneratorFixture(true);
|
||||
RecordingComponent component = new RecordingComponent(ReservedFlag.OBJECT, 0, 0);
|
||||
TestMatterGenerator generator = fixture.generator(List.of(new MantlePass(List.of(component), 0, 0)));
|
||||
MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT);
|
||||
MatterGenerator.IN_FLIGHT_COMPONENTS.put(key, completion);
|
||||
MatterGenerator.MatterComponentTask task = new MatterGenerator.MatterComponentTask(key);
|
||||
task.finish(failure);
|
||||
MatterGenerator.IN_FLIGHT_COMPONENTS.put(key, task);
|
||||
try {
|
||||
CompletableFuture<Void> generation = MultiBurst.burst.completeValueAsync(() -> {
|
||||
generator.generateMatter(0, 0, true, fixture.context);
|
||||
@@ -245,7 +380,7 @@ public class MatterGeneratorConcurrencyTest {
|
||||
});
|
||||
return assertThrows(expectedFailure, () -> generation.get(5L, TimeUnit.SECONDS));
|
||||
} finally {
|
||||
MatterGenerator.IN_FLIGHT_COMPONENTS.remove(key, completion);
|
||||
MatterGenerator.IN_FLIGHT_COMPONENTS.remove(key, task);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user