From 66a55852aec5f8ab4dcd2d0a803eb9fbb74ebb52 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 18 Sep 2026 08:51:43 +0100 Subject: [PATCH 1/4] Bound event-loop termination waits and preserve close ownership The secondary shutdown waiter logged indefinitely after its deadline. Fail with one diagnostic using monotonic elapsed time, keep STOPPING truthful, and surface stop failure before AbstractCloseable can mark a live loop closed. Regression controls exercise expiry, interruption, callback failure and concurrent stop. --- src/main/docs/bounded-termination.adoc | 27 +++ src/main/docs/decision-log.adoc | 10 + .../threads/AbstractLifecycleEventLoop.java | 106 ++++++++-- .../threads/TerminationWaitTest.java | 196 ++++++++++++++++++ 4 files changed, 323 insertions(+), 16 deletions(-) create mode 100644 src/main/docs/bounded-termination.adoc create mode 100644 src/test/java/net/openhft/chronicle/threads/TerminationWaitTest.java diff --git a/src/main/docs/bounded-termination.adoc b/src/main/docs/bounded-termination.adoc new file mode 100644 index 000000000..46f1802ce --- /dev/null +++ b/src/main/docs/bounded-termination.adoc @@ -0,0 +1,27 @@ += Bounded termination waits +:sectnums: +:lang: en-GB + +== Contract + +The thread that wins the transition to `STOPPING` owns the stop callback. Other +callers wait at most five minutes of monotonic elapsed time. Expiry, interruption, +a failed callback or reentrant stop throws `IllegalStateException`; interruption +retains the interrupted flag. Only a successful callback sets `STOPPED`. + +The first failed wait records the loop name, state, elapsed time, stopper stack +and up to eight owned event-loop thread stacks, limited to 64 frames each. +Subsequent failures still throw but do not repeat the diagnostic. The timeout +bounds secondary waits; it cannot forcibly complete an application stop callback. + +`close()` performs this stop check before entering `AbstractCloseable`'s close +machinery, which otherwise catches exceptions and marks the object closed. +An unsuccessful stop therefore leaves the close state and handlers untouched. +After a blocked callback eventually finishes, the caller can close normally. + +== Regression coverage + +`TerminationWaitTest` controls elapsed time independently of wall-clock changes +and exercises normal/repeated stop, concurrent callers, expiry through stop and +close, interruption, callback failure, one diagnostic and retained ownership. +All controlled worker threads are released and joined after the assertion. diff --git a/src/main/docs/decision-log.adoc b/src/main/docs/decision-log.adoc index 3d7ace841..7329d99a7 100644 --- a/src/main/docs/decision-log.adoc +++ b/src/main/docs/decision-log.adoc @@ -246,6 +246,16 @@ Impact & Consequences:: * Explicitly recorded, bounded repeat runs remain useful for diagnosis. * CI must still handle reported failures when `maven.test.failure.ignore` allows later build steps to continue. +[[THR-NF-O-038]] +=== THR-NF-O-038 Bound unsuccessful termination waits + +Date:: 2026-09-18 +Context:: A shutdown deadline logged on every poll without ending the wait; close also swallowed stop failures and could release live resources. +Decision Statement:: Use monotonic elapsed time, throw on unsuccessful waits, emit one bounded diagnostic and stop before committing the close state. +Impact & Consequences:: A failed wait retains resource ownership and does not claim completion. The original stop callback remains responsible for finishing. Interrupted callers keep their interrupted status. +Validation:: `TerminationWaitTest` covers timeout through stop and close, interruption, callback failure, normal and concurrent stop, and eventual cleanup. +Notes/Links:: link:bounded-termination.adoc[Full termination contract and diagnostic limits]. + [[THR-FN-037]] === THR-FN-037 Preserve legacy registration with explicit checked admission diff --git a/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java b/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java index 4f75029a8..9e20d6d9c 100644 --- a/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java +++ b/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java @@ -12,7 +12,9 @@ import org.jetbrains.annotations.NotNull; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.LongSupplier; /** * Base implementation that manages the life-cycle of an {@link EventLoop}. @@ -30,17 +32,22 @@ * Transitions are linear in that order. Invoking {@code stop()} while in * {@code NEW} skips {@code STARTED} entirely. Both {@code start()} and * {@code stop()} are idempotent and {@code stop()} blocks until the loop is - * {@code STOPPED}. + * {@code STOPPED}. A failed or interrupted termination wait throws without + * claiming that shutdown completed. */ @SuppressWarnings("this-escape") public abstract class AbstractLifecycleEventLoop extends AbstractCloseable implements EventLoop { /** - * After this time, awaitTermination will log an error and return, this is really only so - * tests don't block forever. This time should be kept as "effectively forever". + * Bound a secondary caller's wait for the thread already stopping the loop. */ private static final long AWAIT_TERMINATION_TIMEOUT_MS = TimeUnit.MINUTES.toMillis(5); private final AtomicReference lifecycle = new AtomicReference<>(EventLoopLifecycle.NEW); + private final AtomicBoolean terminationFailureReported = new AtomicBoolean(); + private final long terminationTimeoutNs; + private final LongSupplier nanoClock; + private volatile Thread stoppingThread; + private volatile Throwable stopFailure; protected final String name; volatile boolean privateGroup; @@ -54,6 +61,15 @@ public abstract class AbstractLifecycleEventLoop extends AbstractCloseable imple * @param name descriptive name for the loop */ protected AbstractLifecycleEventLoop(@NotNull String name) { + this(name, TimeUnit.MILLISECONDS.toNanos(AWAIT_TERMINATION_TIMEOUT_MS), System::nanoTime); + } + + // Package-local seam: tests advance elapsed time without changing the production deadline. + AbstractLifecycleEventLoop(@NotNull String name, long terminationTimeoutNs, LongSupplier nanoClock) { + if (terminationTimeoutNs <= 0) + throw new IllegalArgumentException("Termination timeout must be positive"); + this.terminationTimeoutNs = terminationTimeoutNs; + this.nanoClock = nanoClock; this.name = name.replaceAll("/$", ""); // event loops operate on dedicated threads but may be closed elsewhere @@ -120,16 +136,30 @@ public final String name() { @Override public final void stop() { if (lifecycle.compareAndSet(EventLoopLifecycle.NEW, EventLoopLifecycle.STOPPING)) { - performStopFromNew(); - lifecycle.set(EventLoopLifecycle.STOPPED); + performStop(false); } else if (lifecycle.compareAndSet(EventLoopLifecycle.STARTED, EventLoopLifecycle.STOPPING)) { - performStopFromStarted(); - lifecycle.set(EventLoopLifecycle.STOPPED); + performStop(true); } else { awaitTermination(); } } + private void performStop(boolean started) { + stoppingThread = Thread.currentThread(); + try { + if (started) + performStopFromStarted(); + else + performStopFromNew(); + lifecycle.set(EventLoopLifecycle.STOPPED); + } catch (RuntimeException | Error failure) { + stopFailure = failure; + throw failure; + } finally { + stoppingThread = null; + } + } + /** * Stop the loop when {@link #stop()} is invoked before it has started. * Implementations should block until every handler has received @@ -149,22 +179,63 @@ public final void stop() { * *

If the state does not change within * {@link #AWAIT_TERMINATION_TIMEOUT_MS} milliseconds an error is logged and - * the method returns. The timeout is primarily to avoid tests hanging - * indefinitely.

+ * an {@link IllegalStateException} is thrown. Interruption preserves the + * interrupted status and also fails the wait. Neither case completes the + * lifecycle or transfers ownership of resources.

*/ protected final void awaitTermination() { - long endTime = System.currentTimeMillis() + AWAIT_TERMINATION_TIMEOUT_MS; - while (!Thread.currentThread().isInterrupted()) { + long start = nanoClock.getAsLong(); + while (true) { if (lifecycle.get() == EventLoopLifecycle.STOPPED) return; - if (System.currentTimeMillis() > endTime) { - Jvm.error().on(getClass(), "awaitTermination() timed out, continuing. This probably represents a bug."); - } + long elapsed = nanoClock.getAsLong() - start; + if (stopFailure != null) + throw terminationFailure("stop callback failed", elapsed); + if (stoppingThread == Thread.currentThread()) + throw terminationFailure("reentrant stop", elapsed); + if (Thread.currentThread().isInterrupted()) + throw terminationFailure("interrupted", elapsed); + if (elapsed >= terminationTimeoutNs) + throw terminationFailure("timed out", elapsed); Jvm.pause(1); } - if (lifecycle.get() != EventLoopLifecycle.STOPPED) { - Jvm.warn().on(getClass(), "awaitTermination() interrupted, returning in state " + lifecycle.get()); + } + + private IllegalStateException terminationFailure(String reason, long elapsedNs) { + StringBuilder diagnostic = new StringBuilder("awaitTermination() ").append(reason) + .append(": loop=").append(name).append(", lifecycle=").append(lifecycle.get()) + .append(", elapsedMs=").append(TimeUnit.NANOSECONDS.toMillis(elapsedNs)); + Thread stopper = stoppingThread; + appendThread(diagnostic, "stopper", stopper); + int remaining = 8; + for (Thread thread : Thread.getAllStackTraces().keySet()) { + if (thread != stopper && isRunningOnThread(thread)) { + appendThread(diagnostic, "event loop", thread); + if (--remaining == 0) { + diagnostic.append("\nFurther event-loop threads omitted"); + break; + } + } + } + IllegalStateException failure = new IllegalStateException(diagnostic.toString(), stopFailure); + // A timeout must not produce one error per millisecond, or per subsequent close call. + if (terminationFailureReported.compareAndSet(false, true)) + Jvm.error().on(getClass(), diagnostic.toString(), failure); + return failure; + } + + @SuppressWarnings("deprecation") // Thread.threadId() is unavailable on the supported Java 8 baseline. + private static void appendThread(StringBuilder diagnostic, String role, Thread thread) { + diagnostic.append('\n').append(role).append('='); + if (thread == null) { + diagnostic.append("none"); + return; } + diagnostic.append(thread.getName()).append(" id=").append(thread.getId()) + .append(" state=").append(thread.getState()); + StackTraceElement[] stack = thread.getStackTrace(); + for (int i = 0; i < Math.min(stack.length, 64); i++) + diagnostic.append("\n at ").append(stack[i]); } @Override @@ -177,6 +248,9 @@ protected void assertCloseable() { if (!privateGroup && isRunningOnThread(Thread.currentThread())) { throw new ThreadingIllegalStateException(getClass() + ": Attempting to close " + name + " from within!", createdHere()); } + // AbstractCloseable swallows performClose failures and marks the object closed. + // Stop before entering that path so failure cannot release a live loop's handlers. + stop(); } public abstract boolean isRunningOnThread(Thread thread); diff --git a/src/test/java/net/openhft/chronicle/threads/TerminationWaitTest.java b/src/test/java/net/openhft/chronicle/threads/TerminationWaitTest.java new file mode 100644 index 000000000..851a2f629 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/threads/TerminationWaitTest.java @@ -0,0 +1,196 @@ +/* + * Copyright 2013-2025 chronicle.software; SPDX-License-Identifier: Apache-2.0 + */ +package net.openhft.chronicle.threads; + +import net.openhft.chronicle.core.Jvm; +import net.openhft.chronicle.core.onoes.ExceptionKey; +import net.openhft.chronicle.core.threads.EventHandler; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.*; + +@Timeout(15) +class TerminationWaitTest extends ThreadsTestCommon { + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void normalAndRepeatedStop(boolean started) { + try (ControlledLoop loop = new ControlledLoop()) { + if (started) + loop.start(); + loop.release.countDown(); + loop.stop(); + loop.stop(); + assertEquals(1, loop.stopCalls.get()); + assertEquals(started, loop.stoppedFromStarted); + } + } + + @Test + void concurrentStopCompletesBeforeDeadline() throws Exception { + ControlledLoop loop = new ControlledLoop(); + AtomicReference failure = new AtomicReference<>(); + Thread stopper = startStopper(loop, failure); + Thread waiter = new Thread(() -> { + try { + loop.stop(); + } catch (Throwable t) { + failure.set(t); + } + }, "termination-waiter"); + try { + waiter.start(); + loop.release.countDown(); + join(waiter); + join(stopper); + assertNull(failure.get()); + assertEquals(1, loop.stopCalls.get()); + } finally { + loop.release.countDown(); + join(stopper); + join(waiter); + loop.close(); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void timeoutDoesNotCompleteStopOrClose(boolean close) throws Exception { + ControlledLoop loop = new ControlledLoop(); + AtomicReference workerFailure = new AtomicReference<>(); + Thread stopper = startStopper(loop, workerFailure); + expectException("awaitTermination() timed out"); + try { + loop.clockStep.set(100); + IllegalStateException failure = assertThrows(IllegalStateException.class, + () -> { if (close) loop.close(); else loop.stop(); }); + assertTrue(failure.getMessage().contains("loop=controlled")); + assertTrue(failure.getMessage().contains("lifecycle=STOPPING")); + assertTrue(failure.getMessage().contains("stopper=termination-stopper")); + assertTrue(failure.getMessage().contains("CountDownLatch.await")); + assertFalse(loop.isClosed()); + assertEquals(0, loop.resourcesClosed.get()); + assertTrue(stopper.isAlive()); + assertThrows(IllegalStateException.class, loop::close); + Map exceptions = Jvm.getValue(this, "exceptions"); + int diagnostics = exceptions.entrySet().stream() + .filter(e -> e.getKey().message.contains("awaitTermination() timed out")) + .mapToInt(Map.Entry::getValue).sum(); + assertEquals(1, diagnostics, "Repeated waits must not flood the log"); + } finally { + loop.release.countDown(); + join(stopper); + loop.close(); + } + assertNull(workerFailure.get()); + assertTrue(loop.isClosed()); + assertEquals(1, loop.resourcesClosed.get()); + } + + @Test + void interruptionPreservesStatusAndResourceOwnership() throws Exception { + ControlledLoop loop = new ControlledLoop(); + AtomicReference workerFailure = new AtomicReference<>(); + Thread stopper = startStopper(loop, workerFailure); + expectException("awaitTermination() interrupted"); + try { + Thread.currentThread().interrupt(); + assertThrows(IllegalStateException.class, loop::close); + assertTrue(Thread.currentThread().isInterrupted()); + assertFalse(loop.isClosed()); + assertEquals(0, loop.resourcesClosed.get()); + } finally { + Thread.interrupted(); + loop.release.countDown(); + join(stopper); + loop.close(); + } + assertNull(workerFailure.get()); + } + + @Test + void failedStopRemainsVisibleThroughClose() { + ControlledLoop loop = new ControlledLoop(); + IllegalArgumentException original = new IllegalArgumentException("stop callback failed deliberately"); + loop.stopFailure = original; + expectException("awaitTermination() stop callback failed"); + try { + assertSame(original, assertThrows(IllegalArgumentException.class, loop::close)); + assertFalse(loop.isClosed()); + IllegalStateException repeated = assertThrows(IllegalStateException.class, loop::close); + assertSame(original, repeated.getCause()); + assertTrue(repeated.getMessage().contains("lifecycle=STOPPING")); + assertEquals(0, loop.resourcesClosed.get()); + } finally { + // This deliberately failed fake owns no native resources or worker threads. + loop.unmonitor(); + } + } + + private static Thread startStopper(ControlledLoop loop, AtomicReference failure) throws Exception { + Thread thread = new Thread(() -> { + try { + loop.stop(); + } catch (Throwable t) { + failure.set(t); + } + }, "termination-stopper"); + thread.start(); + assertTrue(loop.stopping.await(5, TimeUnit.SECONDS)); + return thread; + } + + private static void join(Thread thread) throws InterruptedException { + thread.join(5000); + assertFalse(thread.isAlive(), () -> thread.getName() + " did not terminate"); + } + + private static final class ControlledLoop extends AbstractLifecycleEventLoop { + final AtomicLong clockStep; + final CountDownLatch stopping = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + final AtomicInteger stopCalls = new AtomicInteger(); + final AtomicInteger resourcesClosed = new AtomicInteger(); + boolean stoppedFromStarted; + RuntimeException stopFailure; + + ControlledLoop() { + this(new AtomicLong(), new AtomicLong()); + } + + private ControlledLoop(AtomicLong clock, AtomicLong step) { + super("controlled", 100, () -> clock.getAndAdd(step.get())); + this.clockStep = step; + } + + @Override protected void performStart() { } + @Override protected void performStopFromStarted() { stoppedFromStarted = true; performStopFromNew(); } + @Override protected void performStopFromNew() { + stopCalls.incrementAndGet(); + if (stopFailure != null) + throw stopFailure; + stopping.countDown(); + try { + assertTrue(release.await(10, TimeUnit.SECONDS), "Stop gate not released"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + @Override protected void performClose() { super.performClose(); resourcesClosed.incrementAndGet(); } + @Override public boolean isRunningOnThread(Thread thread) { return false; } + @Override public void addHandler(EventHandler handler) { throw new UnsupportedOperationException(); } + @Override public void unpause() { } + @Override public boolean isAlive() { return stopping.getCount() == 0 && release.getCount() != 0; } + } +} From 5bd9e7741591e753586d4e4ccf9efb63c1fdf05c Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 18 Sep 2026 10:30:40 +0100 Subject: [PATCH 2/4] Keep termination bookkeeping out of Wire configuration --- src/main/docs/bounded-termination.adoc | 5 +++++ .../threads/AbstractLifecycleEventLoop.java | 13 ++++++++----- 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/src/main/docs/bounded-termination.adoc b/src/main/docs/bounded-termination.adoc index 46f1802ce..08f8a56f1 100644 --- a/src/main/docs/bounded-termination.adoc +++ b/src/main/docs/bounded-termination.adoc @@ -19,6 +19,11 @@ machinery, which otherwise catches exceptions and marks the object closed. An unsuccessful stop therefore leaves the close state and handlers untouched. After a blocked callback eventually finishes, the caller can close normally. +The clock, deadline and failure/owner bookkeeping are transient runtime state. +They are excluded from reflective Wire configuration marshalling, preserving the +existing event-group representation. Chronicle-Wire's `MarshallingEventGroupTest` +provides the downstream compatibility control. + == Regression coverage `TerminationWaitTest` controls elapsed time independently of wall-clock changes diff --git a/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java b/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java index 9e20d6d9c..8724c1a4e 100644 --- a/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java +++ b/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java @@ -43,11 +43,14 @@ public abstract class AbstractLifecycleEventLoop extends AbstractCloseable imple */ private static final long AWAIT_TERMINATION_TIMEOUT_MS = TimeUnit.MINUTES.toMillis(5); private final AtomicReference lifecycle = new AtomicReference<>(EventLoopLifecycle.NEW); - private final AtomicBoolean terminationFailureReported = new AtomicBoolean(); - private final long terminationTimeoutNs; - private final LongSupplier nanoClock; - private volatile Thread stoppingThread; - private volatile Throwable stopFailure; + //! Termination bookkeeping belongs to the live loop, not its marshalled configuration. + //! In particular a clock lambda and a stopping thread cannot be portable wire data. + //! Integration control: Chronicle-Wire's MarshallingEventGroupTest.test. + private transient final AtomicBoolean terminationFailureReported = new AtomicBoolean(); + private transient final long terminationTimeoutNs; + private transient final LongSupplier nanoClock; + private transient volatile Thread stoppingThread; + private transient volatile Throwable stopFailure; protected final String name; volatile boolean privateGroup; From e3f53b7735ec47b11d5f4ae073717b02edd50d48 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 18 Sep 2026 11:10:22 +0100 Subject: [PATCH 3/4] Keep handler shutdown bookkeeping out of Wire configuration --- src/main/docs/bounded-termination.adoc | 4 +++- .../java/net/openhft/chronicle/threads/BlockingEventLoop.java | 4 +++- .../java/net/openhft/chronicle/threads/MediumEventLoop.java | 4 +++- .../java/net/openhft/chronicle/threads/MonitorEventLoop.java | 4 +++- 4 files changed, 12 insertions(+), 4 deletions(-) diff --git a/src/main/docs/bounded-termination.adoc b/src/main/docs/bounded-termination.adoc index 08f8a56f1..5ac27b845 100644 --- a/src/main/docs/bounded-termination.adoc +++ b/src/main/docs/bounded-termination.adoc @@ -19,7 +19,9 @@ machinery, which otherwise catches exceptions and marks the object closed. An unsuccessful stop therefore leaves the close state and handlers untouched. After a blocked callback eventually finishes, the caller can close normally. -The clock, deadline and failure/owner bookkeeping are transient runtime state. +The clock, deadline, failure/owner bookkeeping, handler-finished flags and pending +handler admission list are transient runtime state. The pending list is a subset +of the existing handler collection, rather than separate configuration. They are excluded from reflective Wire configuration marshalling, preserving the existing event-group representation. Chronicle-Wire's `MarshallingEventGroupTest` provides the downstream compatibility control. diff --git a/src/main/java/net/openhft/chronicle/threads/BlockingEventLoop.java b/src/main/java/net/openhft/chronicle/threads/BlockingEventLoop.java index 53fd9605b..28457aaf3 100644 --- a/src/main/java/net/openhft/chronicle/threads/BlockingEventLoop.java +++ b/src/main/java/net/openhft/chronicle/threads/BlockingEventLoop.java @@ -43,7 +43,9 @@ public class BlockingEventLoop extends AbstractLifecycleEventLoop implements Eve @NotNull private transient final ExecutorService service; private final List handlers = new CopyOnWriteArrayList<>(); - private final List pendingHandlers = new ArrayList<>(); + //! Pending admission tracks a runtime subset of handlers, which are already represented above. + //! Keep this ownership bookkeeping out of Wire: MarshallingEventGroupTest.test in Chronicle-Wire. + private transient final List pendingHandlers = new ArrayList<>(); private final List runners = new CopyOnWriteArrayList<>(); private final NamedThreadFactory threadFactory; private final Supplier pauserSupplier; diff --git a/src/main/java/net/openhft/chronicle/threads/MediumEventLoop.java b/src/main/java/net/openhft/chronicle/threads/MediumEventLoop.java index 11f07fe25..7e7dede7e 100644 --- a/src/main/java/net/openhft/chronicle/threads/MediumEventLoop.java +++ b/src/main/java/net/openhft/chronicle/threads/MediumEventLoop.java @@ -47,7 +47,9 @@ public class MediumEventLoop extends AbstractLifecycleEventLoop implements CoreE */ private final transient Object addHandlerMutex = new Object(); private final transient Object startStopMutex = new Object(); - private volatile boolean handlersFinished; + //! Shutdown callback bookkeeping is runtime state, not Wire configuration. + //! Compatibility control: Chronicle-Wire's MarshallingEventGroupTest.test. + private transient volatile boolean handlersFinished; @Nullable protected final transient EventLoop parent; diff --git a/src/main/java/net/openhft/chronicle/threads/MonitorEventLoop.java b/src/main/java/net/openhft/chronicle/threads/MonitorEventLoop.java index 61e87770c..93cb15f98 100644 --- a/src/main/java/net/openhft/chronicle/threads/MonitorEventLoop.java +++ b/src/main/java/net/openhft/chronicle/threads/MonitorEventLoop.java @@ -42,7 +42,9 @@ public class MonitorEventLoop extends AbstractLifecycleEventLoop implements Runn private final List handlers = new CopyOnWriteArrayList<>(); private final Pauser pauser; private transient volatile Thread thread = null; - private boolean handlersFinished; + //! Shutdown callback bookkeeping is runtime state, not Wire configuration. + //! Compatibility control: Chronicle-Wire's MarshallingEventGroupTest.test. + private transient boolean handlersFinished; public MonitorEventLoop(final EventLoop parent, final Pauser pauser) { this(parent, "", pauser); From d7ff5bc039a4780881c15ac818287328779fb5a9 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 18 Sep 2026 12:36:01 +0100 Subject: [PATCH 4/4] Fix termination completion and fixture shutdown races --- src/main/docs/bounded-termination.adoc | 12 +++++ .../threads/AbstractLifecycleEventLoop.java | 12 +++-- .../BlockingEventLoopShutdownTest.java | 16 +++++- .../threads/TerminationWaitTest.java | 51 ++++++++++++++++++- 4 files changed, 83 insertions(+), 8 deletions(-) diff --git a/src/main/docs/bounded-termination.adoc b/src/main/docs/bounded-termination.adoc index 5ac27b845..9c15aa65b 100644 --- a/src/main/docs/bounded-termination.adoc +++ b/src/main/docs/bounded-termination.adoc @@ -8,6 +8,9 @@ The thread that wins the transition to `STOPPING` owns the stop callback. Other callers wait at most five minutes of monotonic elapsed time. Expiry, interruption, a failed callback or reentrant stop throws `IllegalStateException`; interruption retains the interrupted flag. Only a successful callback sets `STOPPED`. +If the callback completes between a waiter's initial state read and its expiry +or interruption check, completed shutdown takes precedence; the interrupted flag +is still preserved. An unfinished callback retains the failure behaviour. The first failed wait records the loop name, state, elapsed time, stopper stack and up to eight owned event-loop thread stacks, limited to 64 frames each. @@ -32,3 +35,12 @@ provides the downstream compatibility control. and exercises normal/repeated stop, concurrent callers, expiry through stop and close, interruption, callback failure, one diagnostic and retained ownership. All controlled worker threads are released and joined after the assertion. +`completedStopWinsRaceWithWaitFailure` completes the real stop callback during +the controlled clock read, covering both interruption and deadline expiry. +Stack assertions identify the owned stop callback without requiring it to have +reached a particular instruction after releasing the observer's latch. + +The surviving-runner lookup fixture keeps both handlers live for the lookup +assertions, then waits for their actual removal before closing the loop. Merely +releasing their latches did not guarantee that teardown could no longer interrupt +the second handler's wait. diff --git a/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java b/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java index 8724c1a4e..f62dadef8 100644 --- a/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java +++ b/src/main/java/net/openhft/chronicle/threads/AbstractLifecycleEventLoop.java @@ -196,10 +196,14 @@ protected final void awaitTermination() { throw terminationFailure("stop callback failed", elapsed); if (stoppingThread == Thread.currentThread()) throw terminationFailure("reentrant stop", elapsed); - if (Thread.currentThread().isInterrupted()) - throw terminationFailure("interrupted", elapsed); - if (elapsed >= terminationTimeoutNs) - throw terminationFailure("timed out", elapsed); + if (Thread.currentThread().isInterrupted() || elapsed >= terminationTimeoutNs) { + //! Shutdown can complete after the loop's first state read, including while + //! interrupting its workers. Honour completed ownership transfer before failing. + //! Control: TerminationWaitTest.completedStopWinsRaceWithWaitFailure. + if (lifecycle.get() == EventLoopLifecycle.STOPPED) + return; + throw terminationFailure(Thread.currentThread().isInterrupted() ? "interrupted" : "timed out", elapsed); + } Jvm.pause(1); } } diff --git a/src/test/java/net/openhft/chronicle/threads/BlockingEventLoopShutdownTest.java b/src/test/java/net/openhft/chronicle/threads/BlockingEventLoopShutdownTest.java index 89f2f1d18..1568f288c 100644 --- a/src/test/java/net/openhft/chronicle/threads/BlockingEventLoopShutdownTest.java +++ b/src/test/java/net/openhft/chronicle/threads/BlockingEventLoopShutdownTest.java @@ -68,7 +68,7 @@ void lookupFindsSurvivingRunnerWhenEarlierRunnerIsRemoved() throws IllegalAccess final CountDownLatch finishFirst = new CountDownLatch(1); final CountDownLatch finishSecond = new CountDownLatch(1); final AtomicReference survivor = new AtomicReference<>(); - final RemovingRunnerList runners = new RemovingRunnerList(finishFirst); + final RemovingRunnerList runners = new RemovingRunnerList(finishFirst, 2); runners.finishAfterGet = true; try (BlockingEventLoop loop = new BlockingEventLoop("surviving-runner")) { Jvm.getField(BlockingEventLoop.class, "runners").set(loop, runners); @@ -95,6 +95,10 @@ void lookupFindsSurvivingRunnerWhenEarlierRunnerIsRemoved() throws IllegalAccess } finally { finishFirst.countDown(); finishSecond.countDown(); + // Releasing the latch only makes the survivor runnable. Closing immediately + // can interrupt await before it returns and turn normal teardown into a WARN. + // Keep the live-runner assertions above, then await both actual removals. + await(runners.allRemoved); } } } @@ -112,12 +116,18 @@ private static final class RemovingRunnerList extends CopyOnWriteArrayList controlled = new AtomicReference<>(); + AtomicReference stopperRef = new AtomicReference<>(); + ControlledLoop loop = new ControlledLoop(new AtomicLong(), () -> { + if (clockReads.incrementAndGet() == 2) { + controlled.get().release.countDown(); + try { + join(stopperRef.get()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + if (interrupt) + Thread.currentThread().interrupt(); + return 100; + } + return 0; + }); + controlled.set(loop); + AtomicReference workerFailure = new AtomicReference<>(); + Thread stopper = startStopper(loop, workerFailure); + stopperRef.set(stopper); + try { + // Complete the real stopper between the initial state read and failure check. + assertDoesNotThrow(loop::close); + assertEquals(interrupt, Thread.currentThread().isInterrupted()); + assertTrue(loop.isClosed()); + assertEquals(1, loop.resourcesClosed.get()); + assertNull(workerFailure.get()); + } finally { + Thread.interrupted(); + loop.release.countDown(); + join(stopper); + loop.close(); + } + } + private static Thread startStopper(ControlledLoop loop, AtomicReference failure) throws Exception { Thread thread = new Thread(() -> { try { @@ -169,7 +212,11 @@ private static final class ControlledLoop extends AbstractLifecycleEventLoop { } private ControlledLoop(AtomicLong clock, AtomicLong step) { - super("controlled", 100, () -> clock.getAndAdd(step.get())); + this(step, () -> clock.getAndAdd(step.get())); + } + + private ControlledLoop(AtomicLong step, LongSupplier clock) { + super("controlled", 100, clock); this.clockStep = step; }