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..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 @@ -48,6 +48,7 @@ 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"; + @NotNull private final SingleChronicleQueue queue; @NotNull @@ -99,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 { @@ -120,12 +122,16 @@ class StoreAppender extends AbstractCloseable } if (wire != null) resetPosition(); + scannedCycle = cycle; + + // 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 @@ -399,6 +405,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. @@ -603,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/CreateAtIndexTest.java b/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java index 051d7239d1..0277068811 100644 --- a/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java +++ b/src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java @@ -43,7 +43,8 @@ public class CreateAtIndexTest extends QueueTestCommon { String before = queue.dump(); appender.writeBytes(0x421d00000000L, HELLO_WORLD); String after = queue.dump(); - assertEquals(before, after); + // ignore benign metadata deltas from the appender's first-write EOF normalisation + assertEquals(cleanDump(before), cleanDump(after)); } /* @@ -92,6 +93,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/AppenderReleasesParkedStoreTest.java b/src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java new file mode 100644 index 0000000000..35ec8378a6 --- /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.io.BackgroundResourceReleaser; +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. + 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 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/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/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); 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..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 @@ -152,7 +152,10 @@ 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") + // strip the normalisedEOFsTo record written by first-write EOF normalisation + .replaceAll("# position: \\d+, header: \\d+\\R--- !!data #binary\\RnormalisedEOFsTo: \\d+\\R", ""); } private void doWrite(ChronicleQueue queue, BiConsumer action) {