Adapter patches

(tests)

probe/src/main/java/art/arcane/iris/generation/runtime/ProbeBackgroundTasks.java

@@ -0,0 +1,17 @@
+package art.arcane.iris.generation.runtime;
+
+public final class ProbeBackgroundTasks {
+    private ProbeBackgroundTasks() {
+    }
+
+    public static void await(Engine engine) {
+        if (!(engine instanceof IrisEngine iris)) {
+            throw new IllegalArgumentException("Probe background barrier requires an Iris engine");
+        }
+        await(iris.backgroundTasks);
+    }
+
+    static void await(EngineBackgroundTasks tasks) {
+        tasks.drainBackgroundTasks("probe completion").requireComplete("probe completion");
+    }
+}

probe/src/main/java/art/arcane/iris/probe/GenerationOrderProbe.java

@@ -414,7 +414,7 @@ public final class GenerationOrderProbe {
                     ? generateParallel(session.engine(), scheduled, configuration)
                     : generateSequential(session.engine(), scheduled, configuration);
             long generationNanos = System.nanoTime() - started;
-            List<Throwable> reports = RealPackProbeSupport.settleAndDrain();
+            List<Throwable> reports = RealPackProbeSupport.settleAndDrain(session.engine());
             if (!reports.isEmpty()) {
                 RealPackProbeSupport.printReports(LOG_PREFIX, order + " generation reports", reports);
                 throw reportedFailure(order + " generation", reports);

probe/src/main/java/art/arcane/iris/probe/HydrologyPackProbe.java

@@ -3156,7 +3156,7 @@ public final class HydrologyPackProbe {
             }
             hydrologySeed = engine.getSeedManager().getBodies();
             engineReadyNanos = session.readyNanos();
-            List<Throwable> reports = RealPackProbeSupport.settleAndDrain();
+            List<Throwable> reports = RealPackProbeSupport.settleAndDrain(engine);
             if (!reports.isEmpty()) {
                 RealPackProbeSupport.printReports(LOG_PREFIX, "seed " + seed + " planning reports", reports);
                 throw reportedFailure("Hydrology coverage scan", reports);
@@ -4153,7 +4153,7 @@ public final class HydrologyPackProbe {
                     System.out.println(new GeneratedProcessResult(
                             witness.selector(), verification).machineLine());
                 }
-                List<Throwable> reports = RealPackProbeSupport.settleAndDrain();
+                List<Throwable> reports = RealPackProbeSupport.settleAndDrain(engine);
                 if (!reports.isEmpty()) {
                     RealPackProbeSupport.printReports(LOG_PREFIX, "generated chunk reports", reports);
                     throw reportedFailure("Hydrology generated chunk process", reports);

probe/src/main/java/art/arcane/iris/probe/RealPackProbeSupport.java

@@ -1,9 +1,11 @@
 package art.arcane.iris.probe;
 
 import art.arcane.iris.pack.loading.IrisData;
+import art.arcane.iris.generation.concurrent.MultiBurst;
 import art.arcane.iris.pack.PackValidationResult;
 import art.arcane.iris.pack.PackValidator;
 import art.arcane.iris.generation.runtime.IrisEngine;
+import art.arcane.iris.generation.runtime.ProbeBackgroundTasks;
 import art.arcane.iris.generation.runtime.Engine;
 import art.arcane.iris.generation.runtime.EngineAssignedComponent;
 import art.arcane.iris.generation.runtime.EngineEffects;
@@ -39,6 +41,7 @@ import java.util.EnumSet;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.LinkedHashMap;
@@ -46,6 +49,13 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.LockSupport;
 import java.util.stream.Stream;
 
 final class RealPackProbeSupport {
@@ -79,6 +89,7 @@ final class RealPackProbeSupport {
             validatePack(pack, logPrefix);
             return new Workspace(workRoot, pack, dimensionKey, logPrefix, false, bindings);
         } catch (Throwable failure) {
+            clearRuntime();
             Throwable cleanupFailure = deleteWorkRoot(workRoot);
             if (cleanupFailure != null) {
                 failure.addSuppressed(cleanupFailure);
@@ -103,6 +114,7 @@ final class RealPackProbeSupport {
             configureRuntime(new File(workRoot, "platform-validation"), hydrologyGeneratedVerification, null);
             return new Workspace(workRoot, pack, dimensionKey, logPrefix, hydrologyGeneratedVerification, null);
         } catch (Throwable failure) {
+            clearRuntime();
             Throwable cleanupFailure = deleteWorkRoot(workRoot);
             if (cleanupFailure != null) {
                 failure.addSuppressed(cleanupFailure);
@@ -120,7 +132,12 @@ final class RealPackProbeSupport {
         }
     }
 
-    static List<Throwable> settleAndDrain() throws InterruptedException {
+    static List<Throwable> settleAndDrain(Engine engine) throws InterruptedException {
+        if (IrisPlatforms.isBound() && IrisPlatforms.get() instanceof StubPlatform) {
+            ProbeBackgroundTasks.await(engine);
+            awaitGenerationWorkers();
+            return drainReported();
+        }
         List<Throwable> drained = new ArrayList<>();
         long quietSince = System.currentTimeMillis();
         long start = quietSince;
@@ -139,6 +156,54 @@ final class RealPackProbeSupport {
         return drained;
     }
 
+    private static void awaitGenerationWorkers() throws InterruptedException {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(15);
+        List<ForkJoinPool> pools = new ArrayList<>(3);
+        for (MultiBurst burst : List.of(MultiBurst.hydrology, MultiBurst.ioBurst, MultiBurst.burst)) {
+            try {
+                ForkJoinPool pool = burst.submit((Callable<ForkJoinPool>) ForkJoinTask::getPool)
+                        .get(Math.max(1L, deadline - System.nanoTime()), TimeUnit.NANOSECONDS);
+                if (pool != null) {
+                    pools.add(pool);
+                }
+            } catch (ExecutionException | TimeoutException failure) {
+                throw new IllegalStateException("Probe generation worker barrier failed", failure);
+            }
+        }
+        long[] previous = quietWorkerSnapshot(pools, deadline);
+        while (true) {
+            long[] current = quietWorkerSnapshot(pools, deadline);
+            if (Arrays.equals(previous, current)) {
+                return;
+            }
+            previous = current;
+        }
+    }
+
+    private static long[] quietWorkerSnapshot(List<ForkJoinPool> pools, long deadline) throws InterruptedException {
+        long[] counts = new long[pools.size()];
+        for (int index = 0; index < pools.size(); index++) {
+            ForkJoinPool pool = pools.get(index);
+            while (true) {
+                if (Thread.interrupted()) {
+                    throw new InterruptedException("Probe generation worker barrier interrupted");
+                }
+                if (System.nanoTime() >= deadline) {
+                    throw new IllegalStateException("Probe generation workers did not settle within 15 seconds");
+                }
+                long count = pool.getStealCount();
+                boolean quiet = pool.isQuiescent();
+                long settledCount = pool.getStealCount();
+                if (quiet && count == settledCount) {
+                    counts[index] = settledCount;
+                    break;
+                }
+                LockSupport.parkNanos(TimeUnit.MICROSECONDS.toNanos(100));
+            }
+        }
+        return counts;
+    }
+
     static void printReports(String logPrefix, String label, List<Throwable> reports) {
         if (reports.isEmpty()) {
             return;
@@ -305,7 +370,7 @@ final class RealPackProbeSupport {
                     router.preloadActiveRuntimes();
                 }
                 long readyNanos = System.nanoTime() - started;
-                List<Throwable> initializationReports = settleAndDrain();
+                List<Throwable> initializationReports = settleAndDrain(engine);
                 printReports(logPrefix, runLabel + " engine-init reports", initializationReports);
                 if (bindings != null && !initializationReports.isEmpty()) {
                     IllegalStateException failure = new IllegalStateException("Native engine initialization reported failures");
@@ -317,6 +382,7 @@ final class RealPackProbeSupport {
                 return new EngineSession(this, data, engine, readyNanos);
             } catch (Throwable failure) {
                 Throwable cleanupFailure = closeResources(engine, data);
+                clearRuntime();
                 sessionOpen = false;
                 if (cleanupFailure != null) {
                     failure.addSuppressed(cleanupFailure);
@@ -347,8 +413,7 @@ final class RealPackProbeSupport {
                     failure = bindingFailure;
                 }
             }
-            IrisPlatforms.unbind();
-            StubPlatform.errorSink(null);
+            clearRuntime();
             Throwable cleanupFailure = deleteWorkRoot(workRoot);
             if (cleanupFailure != null) {
                 failure = appendFailure(failure, cleanupFailure);
@@ -494,6 +559,15 @@ final class RealPackProbeSupport {
         }
     }
 
+    private static void clearRuntime() {
+        IrisPlatforms.unbind();
+        IrisServices.remove(PreservationRegistry.class);
+        IrisServices.remove(EngineWorldManagerProvider.class);
+        IrisServices.remove(EngineEffectsProvider.class);
+        IrisServices.remove(EnginePlatformHooks.class);
+        StubPlatform.errorSink(null);
+    }
+
     private static void configureRuntime(File platformRoot, boolean hydrologyGeneratedVerification, RuntimeBindings bindings) {
         REPORTED.clear();
         StubPlatform.bindGenerationStateHandlers();

probe/src/main/java/art/arcane/iris/probe/Terrain3DProbe.java

@@ -328,7 +328,7 @@ public final class Terrain3DProbe {
                 sections.addAll(renderer.write());
             }
             topology = terrainTopology.inspect();
-            failReported("generation", RealPackProbeSupport.settleAndDrain());
+            failReported("generation", RealPackProbeSupport.settleAndDrain(engine));
         }
         List<String> failures = coverageFailures(evidence, configuration.requiredBiomes(), configuration.strictGeometry());
         Result result = new Result(failures.isEmpty() ? "PASS" : "FAIL", configuration, generatedChunks,

probe/src/main/java/art/arcane/iris/probe/TransitionProbe.java

@@ -121,7 +121,7 @@ public final class TransitionProbe {
             }
             require(historicalBiomes.equals(router.biomes().snapshot(0, 0).orElseThrow()),
                     "Historical biome data changed after generation");
-            for (Throwable failure : RealPackProbeSupport.settleAndDrain()) {
+            for (Throwable failure : RealPackProbeSupport.settleAndDrain(engine)) {
                 throw new IllegalStateException("Generation reported a failure", failure);
             }
             require(bindings.captures > 0, "Promotion did not capture historical terrain");

probe/src/test/java/art/arcane/iris/generation/runtime/ProbeBackgroundTasksTest.java

@@ -0,0 +1,131 @@
+package art.arcane.iris.generation.runtime;
+
+import art.arcane.iris.probe.StubPlatform;
+import art.arcane.iris.spi.IrisPlatforms;
+import art.arcane.iris.testsupport.IrisRuntimeState;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
+
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+public final class ProbeBackgroundTasksTest {
+    @Rule
+    public final TemporaryFolder temporary = new TemporaryFolder();
+
+    @Before
+    public void bindRuntime() {
+        IrisPlatforms.bind(new StubPlatform(temporary.getRoot()));
+    }
+
+    @After
+    public void resetRuntime() {
+        IrisRuntimeState.reset();
+        StubPlatform.errorSink(null);
+    }
+
+    @Test(timeout = 15_000L)
+    public void waitsForTrackedContinuationBeforeReturning() throws Exception {
+        EngineBackgroundTasks tasks = new EngineBackgroundTasks();
+        tasks.openBackgroundTaskAdmission();
+        CountDownLatch firstEntered = new CountDownLatch(1);
+        CountDownLatch firstRelease = new CountDownLatch(1);
+        CountDownLatch continuationEntered = new CountDownLatch(1);
+        CountDownLatch continuationRelease = new CountDownLatch(1);
+        try (ExecutorService closer = Executors.newSingleThreadExecutor()) {
+            try {
+                tasks.scheduleTrackedTask(() -> {
+                    awaitRelease(firstEntered, firstRelease);
+                    tasks.scheduleTrackedTask(() -> awaitRelease(continuationEntered, continuationRelease));
+                });
+                assertTrue(firstEntered.await(5, TimeUnit.SECONDS));
+                Future<?> completion = startDrain(closer, tasks);
+                assertThrows(TimeoutException.class, () -> completion.get(100, TimeUnit.MILLISECONDS));
+                firstRelease.countDown();
+                assertTrue(continuationEntered.await(5, TimeUnit.SECONDS));
+                assertThrows(TimeoutException.class, () -> completion.get(100, TimeUnit.MILLISECONDS));
+                continuationRelease.countDown();
+                completion.get(5, TimeUnit.SECONDS);
+            } finally {
+                firstRelease.countDown();
+                continuationRelease.countDown();
+            }
+        }
+    }
+
+    @Test(timeout = 15_000L)
+    public void propagatesAnAsynchronousTrackedFailure() throws Exception {
+        EngineBackgroundTasks tasks = new EngineBackgroundTasks();
+        tasks.openBackgroundTaskAdmission();
+        CountDownLatch entered = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        IllegalStateException failure = new IllegalStateException("tracked generation failure");
+        CountDownLatch reported = new CountDownLatch(1);
+        StubPlatform.errorSink(error -> reported.countDown());
+        try (ExecutorService closer = Executors.newSingleThreadExecutor()) {
+            try {
+                tasks.scheduleTrackedTask(() -> {
+                    awaitRelease(entered, release);
+                    throw failure;
+                });
+                assertTrue(entered.await(5, TimeUnit.SECONDS));
+                Future<?> completion = startDrain(closer, tasks);
+                assertThrows(TimeoutException.class, () -> completion.get(100, TimeUnit.MILLISECONDS));
+                release.countDown();
+                ExecutionException observed = assertThrows(ExecutionException.class,
+                        () -> completion.get(5, TimeUnit.SECONDS));
+                assertSame(failure, observed.getCause().getCause().getCause());
+                assertTrue(reported.await(5, TimeUnit.SECONDS));
+            } finally {
+                release.countDown();
+            }
+        }
+    }
+
+    private static Future<?> startDrain(ExecutorService closer, EngineBackgroundTasks tasks) {
+        AtomicReference<Thread> drainingThread = new AtomicReference<>();
+        Future<?> completion = closer.submit(() -> {
+            drainingThread.set(Thread.currentThread());
+            ProbeBackgroundTasks.await(tasks);
+        });
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        while (System.nanoTime() < deadline) {
+            Thread thread = drainingThread.get();
+            if (thread != null && thread.getState() == Thread.State.TIMED_WAITING) {
+                for (StackTraceElement frame : thread.getStackTrace()) {
+                    if (frame.getClassName().equals(EngineBackgroundTasks.class.getName())
+                            && frame.getMethodName().equals("drainBackgroundTasks")) {
+                        return completion;
+                    }
+                }
+            }
+            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+        }
+        throw new AssertionError("Probe barrier did not await its initial task");
+    }
+
+    private static void awaitRelease(CountDownLatch entered, CountDownLatch release) {
+        entered.countDown();
+        try {
+            assertTrue(release.await(5, TimeUnit.SECONDS));
+        } catch (InterruptedException interruption) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException(interruption);
+        }
+    }
+
+}

probe/src/test/java/art/arcane/iris/probe/ObjectContinuationProbeTest.java

@@ -211,7 +211,7 @@ public final class ObjectContinuationProbeTest {
                     assertArrayEquals(oldReceipt, SavedTerrainChunk.readReceipt(world.worldFolder().toPath(), 0, 0));
                     assertEquals(originalActivation, reopened.resolveActivation(0, 0).activationId());
                     assertTrue(reopened.semantics(1, 0).orElseThrow().objectKeys().contains(fixture.objectKey()));
-                    assertTrue("Generation must not report swallowed errors", RealPackProbeSupport.settleAndDrain().isEmpty());
+                    assertTrue("Generation must not report swallowed errors", RealPackProbeSupport.settleAndDrain(engine).isEmpty());
                 } finally {
                     engine.close();
                 }

probe/src/test/java/art/arcane/iris/probe/RealPackProbeSupportBarrierTest.java

@@ -0,0 +1,101 @@
+package art.arcane.iris.probe;
+
+import art.arcane.iris.generation.concurrent.MultiBurst;
+import art.arcane.iris.generation.runtime.EngineEffectsProvider;
+import art.arcane.iris.generation.runtime.EnginePlatformHooks;
+import art.arcane.iris.generation.runtime.EngineWorldManagerProvider;
+import art.arcane.iris.generation.runtime.PreservationRegistry;
+import art.arcane.iris.spi.IrisPlatforms;
+import art.arcane.iris.spi.IrisServices;
+import art.arcane.iris.testsupport.IrisRuntimeState;
+import art.arcane.iris.testsupport.PlatformLeakGuard;
+import art.arcane.iris.world.task.J;
+import org.junit.AfterClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.net.URL;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+public final class RealPackProbeSupportBarrierTest {
+    @AfterClass
+    public static void resetRuntime() {
+        IrisRuntimeState.reset();
+    }
+
+    @Test
+    public void closesOwnedRuntimeServicesAndPreservesUnrelatedServices() throws Exception {
+        URL resource = RealPackProbeSupportBarrierTest.class.getResource("/seed-cohesion");
+        assertNotNull(resource);
+        IrisServices.register(String.class, "unrelated");
+        try {
+            try (RealPackProbeSupport.Workspace workspace = RealPackProbeSupport.openWorkspace(
+                    new File(resource.toURI()), "cohesion", "[workspace-cleanup]")) {
+                assertNotNull(IrisServices.getOrNull(EngineEffectsProvider.class));
+            }
+            assertFalse(IrisPlatforms.isBound());
+            assertNull(IrisServices.getOrNull(PreservationRegistry.class));
+            assertNull(IrisServices.getOrNull(EngineWorldManagerProvider.class));
+            assertNull(IrisServices.getOrNull(EngineEffectsProvider.class));
+            assertNull(IrisServices.getOrNull(EnginePlatformHooks.class));
+            assertEquals("unrelated", IrisServices.get(String.class));
+        } finally {
+            IrisServices.remove(String.class);
+        }
+    }
+
+    @Test(timeout = 30_000L)
+    public void waitsForCrossPoolContinuationsAndCollectsTheirErrors() throws Exception {
+        URL resource = RealPackProbeSupportBarrierTest.class.getResource("/seed-cohesion");
+        assertNotNull(resource);
+        try (RealPackProbeSupport.Workspace workspace = RealPackProbeSupport.openWorkspace(
+                new File(resource.toURI()), "cohesion", "[worker-barrier]");
+             RealPackProbeSupport.EngineSession session = workspace.openEngine(1337L, false, "barrier")) {
+            for (int repeat = 0; repeat < 40; repeat++) {
+                verifyCrossPoolContinuation(session, MultiBurst.ioBurst, MultiBurst.hydrology);
+                verifyCrossPoolContinuation(session, MultiBurst.hydrology, MultiBurst.ioBurst);
+            }
+        }
+        PlatformLeakGuard.requireClean("closed real-pack workspace", "end");
+    }
+
+    private static void verifyCrossPoolContinuation(RealPackProbeSupport.EngineSession session,
+                                                     MultiBurst first, MultiBurst second) throws Exception {
+        IllegalStateException failure = new IllegalStateException("nested generation failure");
+        CountDownLatch entered = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        try {
+            first.lazy(() -> {
+                entered.countDown();
+                try {
+                    assertTrue(release.await(5, TimeUnit.SECONDS));
+                } catch (InterruptedException interruption) {
+                    Thread.currentThread().interrupt();
+                    throw new IllegalStateException(interruption);
+                }
+                Runnable failedTask = () -> {
+                    throw failure;
+                };
+                second.lazy(() -> first.lazy(() -> second.lazy(() -> J.a(failedTask))));
+            });
+            assertTrue(entered.await(5, TimeUnit.SECONDS));
+            release.countDown();
+            List<Throwable> reports = RealPackProbeSupport.settleAndDrain(session.engine());
+            assertEquals(1, reports.size());
+            assertSame(failure, reports.getFirst());
+            assertTrue(RealPackProbeSupport.drainReported().isEmpty());
+        } finally {
+            release.countDown();
+        }
+    }
+
+}