From 65894df63222a520a7991d6e04db03bcdbbe8fce Mon Sep 17 00:00:00 2001 From: Brian Neumann-Fopiano Date: Tue, 13 Oct 2026 12:29:53 -0400 Subject: [PATCH] Extended probe worker drain - Probe workers share a bounded 120-second settling deadline - Timeout failures include the unsettled pool state - Tests cover active work, deadline expiry, and interruption --- .../iris/probe/RealPackProbeSupport.java | 10 ++- .../RealPackProbeSupportBarrierTest.java | 87 +++++++++++++++++++ 2 files changed, 95 insertions(+), 2 deletions(-) diff --git a/probe/src/main/java/art/arcane/iris/probe/RealPackProbeSupport.java b/probe/src/main/java/art/arcane/iris/probe/RealPackProbeSupport.java index 41a23172d..3890f5c38 100644 --- a/probe/src/main/java/art/arcane/iris/probe/RealPackProbeSupport.java +++ b/probe/src/main/java/art/arcane/iris/probe/RealPackProbeSupport.java @@ -59,6 +59,7 @@ import java.util.concurrent.locks.LockSupport; import java.util.stream.Stream; final class RealPackProbeSupport { + private static final long WORKER_DRAIN_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(120); private static final List REPORTED = Collections.synchronizedList(new ArrayList<>()); private RealPackProbeSupport() { @@ -157,7 +158,7 @@ final class RealPackProbeSupport { } private static void awaitGenerationWorkers() throws InterruptedException { - long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(15); + long deadline = System.nanoTime() + WORKER_DRAIN_TIMEOUT_NANOS; List pools = new ArrayList<>(3); for (MultiBurst burst : List.of(MultiBurst.hydrology, MultiBurst.ioBurst, MultiBurst.burst)) { try { @@ -170,6 +171,10 @@ final class RealPackProbeSupport { throw new IllegalStateException("Probe generation worker barrier failed", failure); } } + awaitGenerationWorkers(pools, deadline); + } + + static void awaitGenerationWorkers(List pools, long deadline) throws InterruptedException { long[] previous = quietWorkerSnapshot(pools, deadline); while (true) { long[] current = quietWorkerSnapshot(pools, deadline); @@ -189,7 +194,8 @@ final class RealPackProbeSupport { throw new InterruptedException("Probe generation worker barrier interrupted"); } if (System.nanoTime() >= deadline) { - throw new IllegalStateException("Probe generation workers did not settle within 15 seconds"); + throw new IllegalStateException("Probe generation workers did not settle before the deadline: pool " + + (index + 1) + "/" + pools.size() + " " + pool); } long count = pool.getStealCount(); boolean quiet = pool.isQuiescent(); diff --git a/probe/src/test/java/art/arcane/iris/probe/RealPackProbeSupportBarrierTest.java b/probe/src/test/java/art/arcane/iris/probe/RealPackProbeSupportBarrierTest.java index a920f8cb7..712205da8 100644 --- a/probe/src/test/java/art/arcane/iris/probe/RealPackProbeSupportBarrierTest.java +++ b/probe/src/test/java/art/arcane/iris/probe/RealPackProbeSupportBarrierTest.java @@ -17,13 +17,17 @@ import java.io.File; import java.net.URL; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; 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.assertThrows; import static org.junit.Assert.assertTrue; public final class RealPackProbeSupportBarrierTest { @@ -68,6 +72,89 @@ public final class RealPackProbeSupportBarrierTest { PlatformLeakGuard.requireClean("closed real-pack workspace", "end"); } + @Test(timeout = 10_000L) + public void waitsForActiveWorkWithinItsDeadline() throws Exception { + try (ForkJoinPool pool = new ForkJoinPool(1)) { + CountDownLatch entered = new CountDownLatch(1); + Future task = pool.submit(() -> { + entered.countDown(); + Thread.sleep(100L); + return null; + }); + assertTrue(entered.await(5L, TimeUnit.SECONDS)); + RealPackProbeSupport.awaitGenerationWorkers(List.of(pool), + System.nanoTime() + TimeUnit.SECONDS.toNanos(5)); + assertTrue(task.isDone()); + task.get(); + } + } + + @Test(timeout = 10_000L) + public void rejectsUnsettledWorkAtItsDeadlineWithoutCancellingIt() throws Exception { + CountDownLatch release = new CountDownLatch(1); + try (ForkJoinPool pool = new ForkJoinPool(1)) { + CountDownLatch entered = new CountDownLatch(1); + Future task = pool.submit(() -> { + entered.countDown(); + release.await(); + return null; + }); + try { + assertTrue(entered.await(5L, TimeUnit.SECONDS)); + IllegalStateException failure = assertThrows(IllegalStateException.class, + () -> RealPackProbeSupport.awaitGenerationWorkers(List.of(pool), + System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(50))); + assertTrue(failure.getMessage().contains("pool 1/1")); + assertTrue(failure.getMessage().contains("active =")); + assertFalse(task.isDone()); + } finally { + release.countDown(); + } + RealPackProbeSupport.awaitGenerationWorkers(List.of(pool), + System.nanoTime() + TimeUnit.SECONDS.toNanos(5)); + task.get(); + } + } + + @Test(timeout = 10_000L) + public void interruptionStopsAnActiveWorkerWait() throws Exception { + CountDownLatch release = new CountDownLatch(1); + try (ForkJoinPool pool = new ForkJoinPool(1)) { + CountDownLatch entered = new CountDownLatch(1); + Future task = pool.submit(() -> { + entered.countDown(); + release.await(); + return null; + }); + AtomicReference failure = new AtomicReference<>(); + CountDownLatch waiting = new CountDownLatch(1); + Thread waiter = new Thread(() -> { + waiting.countDown(); + try { + RealPackProbeSupport.awaitGenerationWorkers(List.of(pool), + System.nanoTime() + TimeUnit.SECONDS.toNanos(5)); + } catch (Throwable interrupted) { + failure.set(interrupted); + } + }); + try { + assertTrue(entered.await(5L, TimeUnit.SECONDS)); + waiter.start(); + assertTrue(waiting.await(5L, TimeUnit.SECONDS)); + waiter.interrupt(); + waiter.join(5_000L); + assertFalse(waiter.isAlive()); + assertTrue(failure.get() instanceof InterruptedException); + assertFalse(task.isDone()); + } finally { + release.countDown(); + waiter.interrupt(); + waiter.join(5_000L); + } + task.get(); + } + } + private static void verifyCrossPoolContinuation(RealPackProbeSupport.EngineSession session, MultiBurst first, MultiBurst second) throws Exception { IllegalStateException failure = new IllegalStateException("nested generation failure");