mirrors/Iris
1
0
Fork 0
mirror of https://github.com/VolmitSoftware/Iris.git synced 2026-10-20 07:24:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Extended probe worker drain

Browse Source 
- Probe workers share a bounded 120-second settling deadline

- Timeout failures include the unsettled pool state

- Tests cover active work, deadline expiry, and interruption
This commit is contained in:
Brian Neumann-Fopiano
2026-10-13 12:29:53 -04:00
parent 2e3a41f262
commit 65894df632
2 changed files with 95 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
probe/src/main/java/art/arcane/iris/probe/RealPackProbeSupport.java
+8 -2
View File
@@ -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<Throwable> 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<ForkJoinPool> 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<ForkJoinPool> 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();
probe/src/test/java/art/arcane/iris/probe/RealPackProbeSupportBarrierTest.java
+87
View File
@@ -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<Throwable> 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");