From 55d8f3f4edc6fbc22e3794bad5d466499a3b4183 Mon Sep 17 00:00:00 2001 From: victor Date: Wed, 8 Jul 2026 23:00:19 +0800 Subject: [PATCH] Reap SegmentManager worker on interrupted close SegmentManager.close() joined the worker thread with a bounded t.join(5_000). When close() ran on a thread that already carried a pending interrupt -- the interrupted-teardown path that BackgroundDrainer deliberately leaves the status set on -- join() threw InterruptedException immediately, so the worker got no time to observe running=false and exit. close() then logged the "worker did not stop" warning and returned, leaking a live worker that still owned segment files. The leaked worker kept creating spares and trimming/unlinking sealed segments. A subsequent CursorSendEngine that mmapped one of those files raced the concurrent truncation/unlink and took a SIGBUS, surfacing as "java.lang.InternalError: a fault occurred in an unsafe memory access operation". This showed up as a flaky failure in BackgroundDrainerInterruptIsStopSignalTest. close() now clears the caller's pending interrupt before the join, waits the worker out to a deadline, and restores the interrupt on the way out so the rest of the interrupted-teardown protocol still observes it. The loop.close() latch await elsewhere stays interrupt-sensitive; only the worker join, which must reap a passive daemon that owns files, is made robust against a pending interrupt. Add testInterruptedCallerDoesNotAbandonReapableWorker, a deterministic regression that fails against the old code, and give SegmentManager a @TestOnly worker-join timeout seam so the existing timeout-branch test reaches its branch without the interrupt shortcut the fix removes. --- .../qwp/client/sf/cursor/SegmentManager.java | 31 ++++++- .../cursor/SegmentManagerCloseRaceTest.java | 85 ++++++++++++++++++- 2 files changed, 109 insertions(+), 7 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentManager.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentManager.java index d96b8627..3b7cff60 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentManager.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/SegmentManager.java @@ -67,6 +67,7 @@ public final class SegmentManager implements QuietCloseable { public static final long DISK_FULL_LOG_THROTTLE_NANOS = 30_000_000_000L; // 30 s public static final long UNLIMITED_TOTAL_BYTES = Long.MAX_VALUE; private static final Logger LOG = LoggerFactory.getLogger(SegmentManager.class); + private static final long WORKER_JOIN_TIMEOUT_MILLIS = 5_000L; private final AtomicLong fileGeneration = new AtomicLong(); private final Object lock = new Object(); @@ -107,6 +108,7 @@ public final class SegmentManager implements QuietCloseable { // {@link #lock}; the lock covers both paths so the counter stays // consistent across registration boundaries. private long totalBytes; + private long workerJoinTimeoutMillis = WORKER_JOIN_TIMEOUT_MILLIS; // volatile because wakeWorker() reads workerThread without holding the // monitor; the synchronized start()/close() pair handles the // start-vs-close ordering. @@ -168,10 +170,28 @@ public synchronized void close() { Thread t = workerThread; if (t != null) { LockSupport.unpark(t); + // A pending interrupt on the caller makes Thread.join() throw at + // once; clear it so the join actually reaps the worker (which + // still owns segment files), then restore it for the rest of the + // interrupted-teardown protocol. + boolean interrupted = Thread.interrupted(); + long deadlineNanos = System.nanoTime() + workerJoinTimeoutMillis * 1_000_000L; try { - t.join(5_000); - } catch (InterruptedException ignored) { - Thread.currentThread().interrupt(); + while (t.isAlive()) { + long remainingMillis = (deadlineNanos - System.nanoTime()) / 1_000_000L; + if (remainingMillis <= 0) { + break; + } + try { + t.join(remainingMillis); + } catch (InterruptedException ignored) { + interrupted = true; + } + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } } if (t.isAlive()) { LOG.warn("SegmentManager worker did not stop before close wait completed; " @@ -287,6 +307,11 @@ public void setBeforeTrimSyncHook(Runnable hook) { this.beforeTrimSyncHook = hook; } + @TestOnly + public void setWorkerJoinTimeoutMillis(long millis) { + this.workerJoinTimeoutMillis = millis; + } + public synchronized void start() { if (workerThread != null) { throw new IllegalStateException("already started"); diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentManagerCloseRaceTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentManagerCloseRaceTest.java index 384cd7ad..3d7c6a7c 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentManagerCloseRaceTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/SegmentManagerCloseRaceTest.java @@ -170,10 +170,7 @@ public void testCloseDoesNotFreePathScratchWhenWorkerStillAlive() throws Excepti Assert.assertTrue("precondition: path scratch should be allocated", readPathScratchImpl(manager) != 0L); - // Exercise the same branch as a timed-out join without making - // the test sleep for 5 seconds: join() returns while the worker - // is still alive. close() must leave worker-owned native memory - // alone so the worker can resume safely. + manager.setWorkerJoinTimeoutMillis(50L); Thread.currentThread().interrupt(); manager.close(); Assert.assertTrue("close should preserve interrupted status", @@ -185,6 +182,7 @@ public void testCloseDoesNotFreePathScratchWhenWorkerStillAlive() throws Excepti readPathScratchImpl(manager) != 0L); releaseWorker.countDown(); + manager.setWorkerJoinTimeoutMillis(TimeUnit.SECONDS.toMillis(60)); manager.close(); managerClosed = true; Assert.assertNull("successful close should clear workerThread", @@ -206,6 +204,85 @@ public void testCloseDoesNotFreePathScratchWhenWorkerStillAlive() throws Excepti }); } + @Test(timeout = 15_000L) + public void testInterruptedCallerDoesNotAbandonReapableWorker() throws Exception { + TestUtils.assertMemoryLeak(() -> { + long segSize = MmapSegment.HEADER_SIZE + (MmapSegment.FRAME_HEADER_SIZE + 32); + String slot = tmpDir + "/interrupt-slot"; + Assert.assertEquals(0, Files.mkdir(slot, Files.DIR_MODE_DEFAULT)); + MmapSegment initial = MmapSegment.create(slot + "/sf-initial.sfa", 0L, segSize); + SegmentRing ring = new SegmentRing(initial, segSize); + SegmentManager manager = new SegmentManager(segSize, TimeUnit.SECONDS.toNanos(60)); + CountDownLatch workerBlocked = new CountDownLatch(1); + CountDownLatch releaseWorker = new CountDownLatch(1); + CountDownLatch closeReturned = new CountDownLatch(1); + AtomicBoolean fired = new AtomicBoolean(); + AtomicBoolean interruptPreserved = new AtomicBoolean(); + AtomicReference hookErr = new AtomicReference<>(); + AtomicReference closeErr = new AtomicReference<>(); + boolean managerClosed = false; + try { + manager.register(ring, slot); + manager.setBeforeInstallSyncHook(() -> { + if (!fired.compareAndSet(false, true)) return; + workerBlocked.countDown(); + try { + if (!releaseWorker.await(10, TimeUnit.SECONDS)) { + hookErr.compareAndSet(null, + new AssertionError("timed out waiting for test to release worker")); + } + } catch (Throwable t) { + hookErr.compareAndSet(null, t); + } + }); + manager.start(); + Assert.assertTrue("worker did not reach install hook", + workerBlocked.await(5, TimeUnit.SECONDS)); + + Thread closer = new Thread(() -> { + Thread.currentThread().interrupt(); + try { + manager.close(); + } catch (Throwable t) { + closeErr.compareAndSet(null, t); + } finally { + interruptPreserved.set(Thread.currentThread().isInterrupted()); + closeReturned.countDown(); + } + }, "interrupted-closer"); + closer.start(); + + Assert.assertFalse("interrupted close() abandoned a live worker instead of waiting", + closeReturned.await(300, TimeUnit.MILLISECONDS)); + + releaseWorker.countDown(); + Assert.assertTrue("close() never returned after the worker was released", + closeReturned.await(10, TimeUnit.SECONDS)); + closer.join(TimeUnit.SECONDS.toMillis(5)); + managerClosed = readWorkerThread(manager) == null; + + if (closeErr.get() != null) { + throw new AssertionError("close() threw", closeErr.get()); + } + Assert.assertNull("close() must reap the worker despite the pending interrupt", + readWorkerThread(manager)); + Assert.assertTrue("close() must restore the caller's interrupt status", + interruptPreserved.get()); + if (hookErr.get() != null) { + throw new AssertionError("install hook failed", hookErr.get()); + } + } finally { + manager.setBeforeInstallSyncHook(null); + releaseWorker.countDown(); + if (!managerClosed) { + Thread.interrupted(); + manager.close(); + } + ring.close(); + } + }); + } + private static void cleanupRecursively(String dir) { if (!Files.exists(dir)) return; long find = Files.findFirst(dir);