From 6ddbac6d1ac6cff5d80bd34fd12e4b04e1df11a3 Mon Sep 17 00:00:00 2001 From: james-mcsherry Date: Thu, 2 Jul 2026 16:18:02 +0100 Subject: [PATCH 1/5] QUEUE-130 Release parked store after StoreAppender construction --- .../queue/impl/single/StoreAppender.java | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java b/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java index e05be62a76..94c643f9c0 100644 --- a/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java +++ b/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java @@ -48,6 +48,16 @@ class StoreAppender extends AbstractCloseable * This is the key in the table-store where we store that information */ private static final String NORMALISED_EOFS_TO_TABLESTORE_KEY = "normalisedEOFsTo"; + + /** + * When set, a freshly-constructed appender releases the store its EOF back-scan parked on, so it + * holds no roll-cycle file open until its first write re-acquires the current cycle. This prevents + * an appender on a thread that then goes idle (or, in async/BUFFERED mode, never writes to the file + * directly at all) from pinning an old {@code .cq4} open for the life of the queue - which defeats + * {@code FileUtil.removableRollFileCandidates()}. See QUEUE-130. + */ + private static final boolean RELEASE_PARKED_STORE_ON_CONSTRUCTION = + Jvm.getBoolean("queue.appender.releaseParkedStoreOnConstruction"); @NotNull private final SingleChronicleQueue queue; @NotNull @@ -120,6 +130,14 @@ class StoreAppender extends AbstractCloseable } if (wire != null) resetPosition(); + + // The back-scan above parks this appender on a store (the newest EOF-ed cycle, or + // the first non-EOF cycle after it). If we hold on to that store and this thread + // then never writes, its file stays mapped/open forever. Release it so we start in + // the same state as an appender created on an empty queue (null store, no FD); the + // first write re-acquires the current cycle via setWireIfNull/rollCycleTo. + if (RELEASE_PARKED_STORE_ON_CONSTRUCTION) + releaseParkedStore(); } } finally { writeLock.unlock(); @@ -399,6 +417,26 @@ private void setCycle2(final int cycle, final WireStoreSupplier.CreateStrategy c queue.onRoll(cycle); } + /** + * Releases the store (and its wires) that the construction-time EOF back-scan left this appender + * parked on, returning the appender to the same "never written" state as one created on an empty + * queue. The next write re-acquires the current cycle. Only the mapped-file reservation and open + * FD are dropped; nothing on disk changes. + */ + private void releaseParkedStore() { + if (store == null) + return; + + releaseBytesFor(wireForIndex); + releaseBytesFor(wire); + wireForIndex = null; + wire = null; + + storePool.closeStore(store); + store = null; + cycle = Integer.MIN_VALUE; + } + /** * Resets the wires (primary and indexing) for this appender based on the store. * Releases any existing wire resources before creating new ones. From f33d6730e4dcc2148cce52f485b09f5338498e8d Mon Sep 17 00:00:00 2001 From: james-mcsherry Date: Thu, 2 Jul 2026 16:18:02 +0100 Subject: [PATCH 2/5] QUEUE-130 Updated tests --- .../openhft/chronicle/queue/CreateAtIndexTest.java | 14 +++++++++++++- .../queue/impl/single/NotCompleteTest.java | 8 +++++++- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java b/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java index 051d7239d1..34c9f00a82 100644 --- a/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java +++ b/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java @@ -43,7 +43,12 @@ public class CreateAtIndexTest extends QueueTestCommon { String before = queue.dump(); appender.writeBytes(0x421d00000000L, HELLO_WORLD); String after = queue.dump(); - assertEquals(before, after); + // The duplicate-index write appends no data. With queue.appender.releaseParkedStoreOnConstruction + // (QUEUE-130) the fresh appender first re-acquires the current cycle's store and re-runs EOF + // normalisation. That leaves only non-semantic metadata deltas - a modCount tick, the + // normalisedEOFsTo watermark record, and store free-space counts - none of which are the data + // this test guards, so normalise them out before asserting the contents are unchanged. + assertEquals(cleanDump(before), cleanDump(after)); } /* @@ -92,6 +97,13 @@ public class CreateAtIndexTest extends QueueTestCommon { } } + private static String cleanDump(String dump) { + return dump + .replaceAll("# \\d+ bytes remaining", "# NN bytes remaining") + .replaceAll("modCount: (\\d+)", "modCount: 00") + .replaceAll("# position: \\d+, header: \\d+\\R--- !!data #binary\\RnormalisedEOFsTo: \\d+\\R", ""); + } + @Test public void testWrittenAndReadIndexesAreTheSameOfTheFirstExcerpt() { File tmp = getTmpDir(); diff --git a/src/test/java/net/openhft/chronicle/queue/impl/single/NotCompleteTest.java b/src/test/java/net/openhft/chronicle/queue/impl/single/NotCompleteTest.java index a59d07e75f..bb9d254807 100644 --- a/src/test/java/net/openhft/chronicle/queue/impl/single/NotCompleteTest.java +++ b/src/test/java/net/openhft/chronicle/queue/impl/single/NotCompleteTest.java @@ -152,7 +152,13 @@ private SingleChronicleQueue createQueue(File tmpDir) { // the last line of the dump changes - haven't spent the time to get to the bottom of this private String cleanQueueDump(String from) { - return from.replaceAll("# [0-9]+ bytes remaining$", "").replaceAll("modCount: (\\d+)", "modCount: 00"); + return from.replaceAll("# [0-9]+ bytes remaining", "# NN bytes remaining") + .replaceAll("modCount: (\\d+)", "modCount: 00") + // With queue.appender.releaseParkedStoreOnConstruction (QUEUE-130) a fresh appender's + // first write re-runs EOF normalisation, which records the normalisedEOFsTo watermark + // (and shifts store free-space counts). That is EOF bookkeeping (covered by the dedicated + // EOF tests), not the interrupted-write data this test guards, so strip it before comparing. + .replaceAll("# position: \\d+, header: \\d+\\R--- !!data #binary\\RnormalisedEOFsTo: \\d+\\R", ""); } private void doWrite(ChronicleQueue queue, BiConsumer action) { From 844502d6824ebd75f43d4c05e041a34f4b077b71 Mon Sep 17 00:00:00 2001 From: james-mcsherry Date: Fri, 3 Jul 2026 12:18:11 +0100 Subject: [PATCH 3/5] QUEUE-130 Add test that a parked appender does not pin old roll files --- .../queue/impl/single/StoreAppender.java | 24 ++---- .../chronicle/queue/CreateAtIndexTest.java | 6 +- .../AppenderReleasesParkedStoreTest.java | 86 +++++++++++++++++++ .../queue/impl/single/IndexingTestCommon.java | 2 +- .../queue/impl/single/NotCompleteTest.java | 5 +- 5 files changed, 95 insertions(+), 28 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java diff --git a/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java b/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java index 94c643f9c0..abbe003eb9 100644 --- a/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java +++ b/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java @@ -49,15 +49,6 @@ class StoreAppender extends AbstractCloseable */ private static final String NORMALISED_EOFS_TO_TABLESTORE_KEY = "normalisedEOFsTo"; - /** - * When set, a freshly-constructed appender releases the store its EOF back-scan parked on, so it - * holds no roll-cycle file open until its first write re-acquires the current cycle. This prevents - * an appender on a thread that then goes idle (or, in async/BUFFERED mode, never writes to the file - * directly at all) from pinning an old {@code .cq4} open for the life of the queue - which defeats - * {@code FileUtil.removableRollFileCandidates()}. See QUEUE-130. - */ - private static final boolean RELEASE_PARKED_STORE_ON_CONSTRUCTION = - Jvm.getBoolean("queue.appender.releaseParkedStoreOnConstruction"); @NotNull private final SingleChronicleQueue queue; @NotNull @@ -109,6 +100,7 @@ class StoreAppender extends AbstractCloseable int lastExistingCycle = queue.lastCycle(); int firstCycle = queue.firstCycle(); long start = System.nanoTime(); + int scannedCycle = Integer.MIN_VALUE; final WriteLock writeLock = this.queue.writeLock(); writeLock.lock(); try { @@ -130,20 +122,16 @@ class StoreAppender extends AbstractCloseable } if (wire != null) resetPosition(); + scannedCycle = cycle; - // The back-scan above parks this appender on a store (the newest EOF-ed cycle, or - // the first non-EOF cycle after it). If we hold on to that store and this thread - // then never writes, its file stays mapped/open forever. Release it so we start in - // the same state as an appender created on an empty queue (null store, no FD); the - // first write re-acquires the current cycle via setWireIfNull/rollCycleTo. - if (RELEASE_PARKED_STORE_ON_CONSTRUCTION) - releaseParkedStore(); + // Don't hold the back-scan's store open; the first write re-acquires it + releaseParkedStore(); } } finally { writeLock.unlock(); long tookMillis = (System.nanoTime() - start) / 1_000_000; - if (tookMillis > WARN_SLOW_APPENDER_MS || (lastExistingCycle >= 0 && cycle != lastExistingCycle)) - Jvm.perf().on(getClass(), "Took " + tookMillis + "ms to find first open cycle " + cycle); + if (tookMillis > WARN_SLOW_APPENDER_MS || (lastExistingCycle >= 0 && scannedCycle != lastExistingCycle)) + Jvm.perf().on(getClass(), "Took " + tookMillis + "ms to find first open cycle " + scannedCycle); } } catch (RuntimeException ex) { // Perhaps initialization code needs to be moved away from constructor diff --git a/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java b/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java index 34c9f00a82..0277068811 100644 --- a/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java +++ b/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java @@ -43,11 +43,7 @@ public class CreateAtIndexTest extends QueueTestCommon { String before = queue.dump(); appender.writeBytes(0x421d00000000L, HELLO_WORLD); String after = queue.dump(); - // The duplicate-index write appends no data. With queue.appender.releaseParkedStoreOnConstruction - // (QUEUE-130) the fresh appender first re-acquires the current cycle's store and re-runs EOF - // normalisation. That leaves only non-semantic metadata deltas - a modCount tick, the - // normalisedEOFsTo watermark record, and store free-space counts - none of which are the data - // this test guards, so normalise them out before asserting the contents are unchanged. + // ignore benign metadata deltas from the appender's first-write EOF normalisation assertEquals(cleanDump(before), cleanDump(after)); } diff --git a/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java b/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java new file mode 100644 index 0000000000..980b0802c3 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java @@ -0,0 +1,86 @@ +/* + * Copyright 2013-2025 chronicle.software; SPDX-License-Identifier: Apache-2.0 + */ +package net.openhft.chronicle.queue.impl.single; + +import net.openhft.chronicle.core.Jvm; +import net.openhft.chronicle.queue.util.FileUtil; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Stream; + +import static java.util.stream.Collectors.toList; +import static net.openhft.chronicle.queue.internal.util.InternalFileUtil.getAllOpenFilesIsSupportedOnOS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/** + * Verifies through {@link FileUtil#removableRollFileCandidates} that an appender does not keep an old + * roll-cycle file open once it has rolled into the past. Removal stops at the first still-open file + * (oldest first), so a pinned old file blocks everything after it. + *

+ * These tests inspect open descriptors via {@code /proc/self/fd} and are skipped where unsupported. + */ +class AppenderReleasesParkedStoreTest extends IndexingTestCommon { + + private static final Comparator EARLIEST_FIRST = Comparator.comparing(File::getName); + + // Control: with only the writer active, the current cycle stays open and earlier cycles are removable. + @Test + void writerAloneLeavesEveryEarlierCycleRemovable() { + assumeTrue(getAllOpenFilesIsSupportedOnOS()); + + appender.writeText("cycle-0"); + timeProvider.advanceMillis(1_001); + appender.writeText("cycle-1"); + timeProvider.advanceMillis(1_001); + appender.writeText("cycle-2"); + + final List created = cycleFiles(); + assertEquals(3, created.size(), "expected three roll-cycle files, got " + created); + assertEquals(created.subList(0, 2), removableCandidates()); + } + + // A fresh appender parks on the current cycle at construction and never writes; after the writer rolls + // past it, it must not keep the now-old file open. + @Test + void parkedAppenderDoesNotPinAnOldCycle() { + assumeTrue(getAllOpenFilesIsSupportedOnOS()); + + appender.writeText("cycle-0"); + timeProvider.advanceMillis(1_001); + appender.writeText("cycle-1"); + + final StoreAppender parked = (StoreAppender) queue.createAppender(); + try { + timeProvider.advanceMillis(1_001); + appender.writeText("cycle-2"); + + final List created = cycleFiles(); + assertEquals(3, created.size(), "expected three roll-cycle files, got " + created); + assertEquals(created.subList(0, 2), removableCandidates()); + } finally { + parked.close(); + } + } + + // All roll-cycle files, earliest first; the pause lets freshly rolled files appear. + private List cycleFiles() { + Jvm.pause(300); + final File[] files = queue.file().listFiles(FileUtil::hasQueueSuffix); + assertNotNull(files, "no queue files found in " + queue.file()); + return Stream.of(files).sorted(EARLIEST_FIRST).collect(toList()); + } + + // Removable candidates, earliest first; the pause lets rolled-off files be released first. + private List removableCandidates() { + Jvm.pause(300); + final List candidates = FileUtil.removableRollFileCandidates(queue.file()).collect(toList()); + assertEquals(candidates.stream().sorted(EARLIEST_FIRST).collect(toList()), candidates); + return candidates; + } +} diff --git a/src/test/java/net/openhft/chronicle/queue/impl/single/IndexingTestCommon.java b/src/test/java/net/openhft/chronicle/queue/impl/single/IndexingTestCommon.java index 6c67ced599..adf8cbce88 100644 --- a/src/test/java/net/openhft/chronicle/queue/impl/single/IndexingTestCommon.java +++ b/src/test/java/net/openhft/chronicle/queue/impl/single/IndexingTestCommon.java @@ -24,7 +24,7 @@ */ public class IndexingTestCommon extends QueueTestCommon { - SetTimeProvider timeProvider; + protected SetTimeProvider timeProvider; SingleChronicleQueue queue; StoreAppender appender; private List closeables; diff --git a/src/test/java/net/openhft/chronicle/queue/impl/single/NotCompleteTest.java b/src/test/java/net/openhft/chronicle/queue/impl/single/NotCompleteTest.java index bb9d254807..2c365344f2 100644 --- a/src/test/java/net/openhft/chronicle/queue/impl/single/NotCompleteTest.java +++ b/src/test/java/net/openhft/chronicle/queue/impl/single/NotCompleteTest.java @@ -154,10 +154,7 @@ private SingleChronicleQueue createQueue(File tmpDir) { private String cleanQueueDump(String from) { return from.replaceAll("# [0-9]+ bytes remaining", "# NN bytes remaining") .replaceAll("modCount: (\\d+)", "modCount: 00") - // With queue.appender.releaseParkedStoreOnConstruction (QUEUE-130) a fresh appender's - // first write re-runs EOF normalisation, which records the normalisedEOFsTo watermark - // (and shifts store free-space counts). That is EOF bookkeeping (covered by the dedicated - // EOF tests), not the interrupted-write data this test guards, so strip it before comparing. + // strip the normalisedEOFsTo record written by first-write EOF normalisation .replaceAll("# position: \\d+, header: \\d+\\R--- !!data #binary\\RnormalisedEOFsTo: \\d+\\R", ""); } From 0ccc51bac9505d384f96778b4887e18803dba683 Mon Sep 17 00:00:00 2001 From: james-mcsherry Date: Mon, 6 Jul 2026 11:02:11 +0100 Subject: [PATCH 4/5] QUEUE-130: removed pause from test --- .../queue/impl/single/AppenderReleasesParkedStoreTest.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java b/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java index 980b0802c3..cf44d791f7 100644 --- a/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java +++ b/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java @@ -70,7 +70,6 @@ void parkedAppenderDoesNotPinAnOldCycle() { // All roll-cycle files, earliest first; the pause lets freshly rolled files appear. private List cycleFiles() { - Jvm.pause(300); final File[] files = queue.file().listFiles(FileUtil::hasQueueSuffix); assertNotNull(files, "no queue files found in " + queue.file()); return Stream.of(files).sorted(EARLIEST_FIRST).collect(toList()); @@ -78,7 +77,6 @@ private List cycleFiles() { // Removable candidates, earliest first; the pause lets rolled-off files be released first. private List removableCandidates() { - Jvm.pause(300); final List candidates = FileUtil.removableRollFileCandidates(queue.file()).collect(toList()); assertEquals(candidates.stream().sorted(EARLIEST_FIRST).collect(toList()), candidates); return candidates; From d1b0e7de5e8cfe8deed74aa994ab3be430fbb34e Mon Sep 17 00:00:00 2001 From: james-mcsherry Date: Fri, 10 Jul 2026 13:55:32 +0100 Subject: [PATCH 5/5] QUEUE-130: Addressed review comments --- .../queue/impl/single/StoreAppender.java | 4 ++- .../AppenderReleasesParkedStoreTest.java | 8 ++++-- .../queue/impl/single/NormaliseEOFsTest.java | 28 +++++++++++++++++++ 3 files changed, 36 insertions(+), 4 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java b/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java index abbe003eb9..fa518341bc 100644 --- a/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java +++ b/src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java @@ -629,7 +629,9 @@ public void normaliseEOFs() { final WriteLock writeLock = queue.writeLock(); writeLock.lock(); try { - normaliseEOFs0(cycle); + // use the getter, not the raw field: after the construction-time back-scan releases its + // parked store the field is Integer.MIN_VALUE, and the getter resolves that to lastCycle + normaliseEOFs0(cycle()); } finally { writeLock.unlock(); long tookMillis = (System.nanoTime() - start) / 1_000_000; diff --git a/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java b/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java index cf44d791f7..35ec8378a6 100644 --- a/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java +++ b/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java @@ -3,7 +3,7 @@ */ package net.openhft.chronicle.queue.impl.single; -import net.openhft.chronicle.core.Jvm; +import net.openhft.chronicle.core.io.BackgroundResourceReleaser; import net.openhft.chronicle.queue.util.FileUtil; import org.junit.jupiter.api.Test; @@ -68,15 +68,17 @@ void parkedAppenderDoesNotPinAnOldCycle() { } } - // All roll-cycle files, earliest first; the pause lets freshly rolled files appear. + // All roll-cycle files, earliest first. private List cycleFiles() { final File[] files = queue.file().listFiles(FileUtil::hasQueueSuffix); assertNotNull(files, "no queue files found in " + queue.file()); return Stream.of(files).sorted(EARLIEST_FIRST).collect(toList()); } - // Removable candidates, earliest first; the pause lets rolled-off files be released first. + // Removable candidates, earliest first. The parked store's file descriptor is dropped on a + // background thread, so drain the releaser before inspecting open descriptors. private List removableCandidates() { + BackgroundResourceReleaser.releasePendingResources(); final List candidates = FileUtil.removableRollFileCandidates(queue.file()).collect(toList()); assertEquals(candidates.stream().sorted(EARLIEST_FIRST).collect(toList()), candidates); return candidates; diff --git a/src/test/java/net/openhft/chronicle/queue/impl/single/NormaliseEOFsTest.java b/src/test/java/net/openhft/chronicle/queue/impl/single/NormaliseEOFsTest.java index 94a4e8876b..389c599f02 100644 --- a/src/test/java/net/openhft/chronicle/queue/impl/single/NormaliseEOFsTest.java +++ b/src/test/java/net/openhft/chronicle/queue/impl/single/NormaliseEOFsTest.java @@ -98,6 +98,34 @@ public void normaliseShouldResumeFromPreviousNormalisation() { } } + // A freshly constructed appender parks on the current cycle during its EOF back-scan and then + // releases that store, leaving its cycle field unset. normaliseEOFs() must still finalise the older + // cycles rather than silently no-op - the watermark it advances is "normalisedEOFsTo" in the table store. + @Test + public void freshlyConstructedAppenderNormalisesWithoutWriting() { + SetTimeProvider setTimeProvider = new SetTimeProvider(); + try (final SingleChronicleQueue queue = createQueue(setTimeProvider); + final ExcerptAppender writer = queue.createAppender()) { + writer.writeText("cycle-0"); + setTimeProvider.advanceMillis(1_001); + writer.writeText("cycle-1"); + setTimeProvider.advanceMillis(1_001); + writer.writeText("cycle-2"); + + final int firstCycle = queue.firstCycle(); + try (final ExcerptAppender fresh = queue.createAppender()) { + final long normalisedBefore = queue.tableStoreAcquire("normalisedEOFsTo", firstCycle).getVolatileValue(); + + fresh.normaliseEOFs(); + + final long normalisedAfter = queue.tableStoreAcquire("normalisedEOFsTo", firstCycle).getVolatileValue(); + assertTrue(normalisedAfter > normalisedBefore, + "normaliseEOFs on a never-written appender should have advanced the watermark past " + + normalisedBefore + " but it stayed at " + normalisedAfter); + } + } + } + private void createNewRollCycles(ExcerptAppender appender, SetTimeProvider timeProvider) { for (int i = 0; i < 10; i++) { timeProvider.advanceMillis(3_000);