From 1092c3f822b52a5224e1c166bb29c2ec64cda90e Mon Sep 17 00:00:00 2001 From: "Christopher L. Shannon" Date: Thu, 6 Feb 2025 16:48:12 -0500 Subject: [PATCH] Add support for PageFile free page truncation This commit adds support for index compaction using a truncation strategy as described in #2232 Compaction works by reclaiming space by truncating free pages at the end of the file. A large amount of free pages can be allocated during certain use cases (such as a large message backlogs) and this strategy helps by removing the excess space when possible. The amount of space reclaimed is configurable using a min/max free page ratio. The min free page ratio defines the amount of free pages to leave and the max free page ratio defines the max free pages to allow before attemping to reclaim space. The compaction check will be performed during the normal checkpoint cycle. --- .../kahadb/KahaDBPersistenceAdapter.java | 25 ++ .../store/kahadb/MessageDatabase.java | 48 +++ .../store/kahadb/disk/page/PageFile.java | 231 ++++++++++- .../kahadb/KahaDBIndexCompactionTest.java | 140 +++++++ .../disk/page/PageFileCompactionTest.java | 385 ++++++++++++++++++ 5 files changed, 827 insertions(+), 2 deletions(-) create mode 100644 activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/KahaDBIndexCompactionTest.java create mode 100644 activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/page/PageFileCompactionTest.java diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java index 2725eb08a9a..c280096f999 100644 --- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java +++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java @@ -46,6 +46,7 @@ import org.apache.activemq.store.kahadb.data.KahaXATransactionId; import org.apache.activemq.store.kahadb.disk.journal.DataFileFactory; import org.apache.activemq.store.kahadb.disk.journal.Journal.JournalDiskSyncStrategy; +import org.apache.activemq.store.kahadb.disk.page.PageFile.PageFileCompactionStrategy; import org.apache.activemq.usage.SystemUsage; import org.apache.activemq.util.ServiceStopper; @@ -774,6 +775,30 @@ public void setEnableSubscriptionStatistics(boolean enableSubscriptionStatistics letter.setEnableSubscriptionStatistics(enableSubscriptionStatistics); } + public float getMinFreePageCompactionRatio() { + return letter.getMinFreePageCompactionRatio(); + } + + public void setMinFreePageCompactionRatio(float minFreePageCompactionRatio) { + letter.setMinFreePageCompactionRatio(minFreePageCompactionRatio); + } + + public float getMaxFreePageCompactionRatio() { + return letter.getMaxFreePageCompactionRatio(); + } + + public void setMaxFreePageCompactionRatio(float maxFreePageCompactionRatio) { + letter.setMaxFreePageCompactionRatio(maxFreePageCompactionRatio); + } + + public PageFileCompactionStrategy getIndexCompactionStrategy() { + return letter.getIndexCompactionStrategy(); + } + + public void setIndexCompactionStrategy(PageFileCompactionStrategy indexCompactionStrategy) { + letter.setIndexCompactionStrategy(indexCompactionStrategy); + } + public KahaDBStore getStore() { return letter; } diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java index 08751b9c3bd..b8982b2941c 100644 --- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java +++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java @@ -97,6 +97,7 @@ import org.apache.activemq.store.kahadb.disk.journal.TargetedDataFileAppender; import org.apache.activemq.store.kahadb.disk.page.Page; import org.apache.activemq.store.kahadb.disk.page.PageFile; +import org.apache.activemq.store.kahadb.disk.page.PageFile.PageFileCompactionStrategy; import org.apache.activemq.store.kahadb.disk.page.Transaction; import org.apache.activemq.store.kahadb.disk.util.LocationMarshaller; import org.apache.activemq.store.kahadb.disk.util.LongMarshaller; @@ -291,6 +292,9 @@ public enum PurgeRecoveredXATransactionStrategy { private boolean enableIndexDiskSyncs = true; private boolean enableIndexRecoveryFile = true; private boolean enableIndexPageCaching = true; + private PageFileCompactionStrategy indexCompactionStrategy = PageFileCompactionStrategy.NEVER; + private float minFreePageCompactionRatio = .1F; + private float maxFreePageCompactionRatio = .3F; ReentrantReadWriteLock checkpointLock = new ReentrantReadWriteLock(); private boolean enableAckCompaction = true; @@ -1699,6 +1703,8 @@ private void checkpointUpdate(final boolean cleanup) throws IOException { Set filesToGc = pageFile.tx().execute((Transaction.CallableClosure, IOException>) tx -> checkpointUpdate(tx, cleanup)); pageFile.flush(); + pageFile.compact(); + // after the index update such that partial removal does not leave dangling references in the index. journal.removeDataFiles(filesToGc); } finally { @@ -3263,6 +3269,9 @@ private PageFile createPageFile() throws IOException { index.setEnableDiskSyncs(isEnableIndexDiskSyncs()); index.setEnableRecoveryFile(isEnableIndexRecoveryFile()); index.setEnablePageCaching(isEnableIndexPageCaching()); + index.setCompactionStrategy(getIndexCompactionStrategy()); + index.setMaxFreePageCompactionRatio(getMaxFreePageCompactionRatio()); + index.setMinFreePageCompactionRatio(getMinFreePageCompactionRatio()); return index; } @@ -4119,6 +4128,45 @@ public void setEnableSubscriptionStatistics(boolean enableSubscriptionStatistics this.enableSubscriptionStatistics = enableSubscriptionStatistics; } + public float getMinFreePageCompactionRatio() { + return minFreePageCompactionRatio; + } + + /** + * The ratio of the minimum amount of free pages to keep. + * The default will keep a minimum of 10% of free pages, relative to the + * current size of the file when compaction is started. + * + * @param minFreePageCompactionRatio + */ + public void setMinFreePageCompactionRatio(float minFreePageCompactionRatio) { + this.minFreePageCompactionRatio = minFreePageCompactionRatio; + } + + public float getMaxFreePageCompactionRatio() { + return maxFreePageCompactionRatio; + } + + /** + * The ratio of the maximum amount of free pages to allow before triggering compaction. + * The default will trigger a compaction attempt if the percentage of free pages + * hits 30% of the total file size. + * + * @param maxFreePageCompactionRatio + */ + public void setMaxFreePageCompactionRatio(float maxFreePageCompactionRatio) { + this.maxFreePageCompactionRatio = maxFreePageCompactionRatio; + } + + public PageFileCompactionStrategy getIndexCompactionStrategy() { + return indexCompactionStrategy; + } + + public void setIndexCompactionStrategy( + PageFileCompactionStrategy indexCompactionStrategy) { + this.indexCompactionStrategy = indexCompactionStrategy; + } + private static class MessageDatabaseObjectInputStream extends ObjectInputStream { public MessageDatabaseObjectInputStream(InputStream is) throws IOException { diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/page/PageFile.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/page/PageFile.java index cfb86d613de..af5944b4797 100644 --- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/page/PageFile.java +++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/page/PageFile.java @@ -35,6 +35,7 @@ import java.util.LinkedHashMap; import java.util.Map; import java.util.Map.Entry; +import java.util.Objects; import java.util.Properties; import java.util.TreeMap; import java.util.concurrent.CountDownLatch; @@ -137,8 +138,9 @@ public class PageFile { private final AtomicLong nextFreePageId = new AtomicLong(); private SequenceSet freeList = new SequenceSet(); - private AtomicReference recoveredFreeList = new AtomicReference(); - private AtomicReference trackingFreeDuringRecovery = new AtomicReference(); + // package visibility for testing + final AtomicReference recoveredFreeList = new AtomicReference<>(); + private final AtomicReference trackingFreeDuringRecovery = new AtomicReference<>(); private final AtomicLong nextTxid = new AtomicLong(); @@ -150,6 +152,33 @@ public class PageFile { private boolean useLFRUEviction = false; private float LFUEvictionFactor = 0.2f; + // Compaction config + private PageFileCompactionStrategy compactionStrategy = PageFileCompactionStrategy.NEVER; + + // The ratio of the minimum amount of free pages to keep + // The default will keep a minimum of 10% of free pages, relative to the + // current size of the file when compaction is started. + // For example, if the file has 1000 pages and 500 are free at the end of the file, + // 400 would be removed and 100 would be kept (100 is 10% of 1000). This would leave + // a file with 500 pages used + 100 free + private float minFreePageCompactionRatio = .1F; + + // The ratio of the maximum amount of free pages to allow before triggering compaction. + // The default will trigger a compaction attempt if the percentage of free pages + // hits 30% of the total file size. + // For example, if the file has 1000 pages and 350 are free at the end of the file, this + // is greater than 30% so we'd truncate the file. If minFreePageCompactionRatio is kept + // as the default of 10%, 250 pages would be removed leaving 100. + private float maxFreePageCompactionRatio = .3F; + + public enum PageFileCompactionStrategy { + TRUNCATION, NEVER; + + public boolean isNever() { + return this == NEVER; + } + } + /** * Use to keep track of updated pages which have not yet been committed. */ @@ -372,6 +401,9 @@ private void archive(File file, String suffix) throws IOException { */ public void load() throws IOException, IllegalStateException { if (loaded.compareAndSet(false, true)) { + if (maxFreePageCompactionRatio < minFreePageCompactionRatio) { + throw new IllegalStateException("minFreePageCompactionRatio is greater than maxFreePageCompactionRatio"); + } if (enablePageCaching) { if (isUseLFRUEviction()) { @@ -613,6 +645,150 @@ public void flush() throws IOException { } } + /** + * Compact the PageFile if needed. The only supported strategy is truncation, + * but future strategies may be added. + *

+ * PageFile is a contiguous block of pages and new pages get allocated + * as space is needed. If pages are no longer needed anymore, they are + * marked as free so they can be re-used. This normally works well + * for workflows that are consistent. + *

+ * However, sometimes it's possible to end up with a large number of free + * pages and wasted space. An example is an unusual backlog of data on + * a destination. This can cause a huge increase in the PageFile size to + * track the messages. Once the backlog is gone, the PageFile has a lot + * of free pages and is huge and will never shrink. + *

+ * Truncation helps in this scenario by simply truncating the file and + * removing the excess free pages. This works well because all we need + * to do is remove the free pages from the tracking and then simply + * set the length of the file to the new size which is nearly instant time, + * so there is no noticeable performance impact. + *

+ * One major limitation to note is that this strategy only works if the free pages are + * at the end of the file. If the free pages are in the middle of the file + * his won't work so we can't shrink the file. This could happen, for example, if a new + * destination was created while thre was a big backlog and it was written + * to the index. This situation will hopefully be addressed in a future + * compaction update as it is a more complex to handle and will likely require + * a different compaction strategy such as defragging first or rewriting the file. + * + * @throws IOException + */ + public void compact() throws IOException { + if (compactionStrategy.isNever()) { + LOG.debug("Skipping PageFile compaction check, compaction is disabled."); + return; + } + + LOG.debug("Beginning PageFile compaction check."); + + // Don't compact if we have not finished free page recovery + if (trackingFreeDuringRecovery.get() != null) { + LOG.debug("Skipping compaction, async recovery not finished"); + return; + } + + // diskSize computed by checking nextFreePageId + long diskSize = getDiskSize(); + + // Disk size of the free pages in the page file + long freePageCount = getFreePageCount(); + long totalPageCount = getPageCount(); + + // Percentage of pages that are free vs in use + double freePageRatio = Math.round((double)freePageCount / totalPageCount * 100d) / 100d; + + // Only attempt to compact if we have reached the maximum ratio of free pages + // that is configured. + var tolerance = .001; + if (freePageRatio < maxFreePageCompactionRatio - tolerance) { + var formatted = Math.round((double)freePageCount / totalPageCount * 100d) / 100d; + LOG.debug("Skipping compaction, page file freePageRatio {} is less than " + + "configured maxFreePageCompactionRatio {}", String.format("%,.2f", formatted), maxFreePageCompactionRatio); + return; + } + + // Get the last sequence of contiguous set of free pages in the file + // This block could be either in the middle of the file somewhere or + // at the end of the file + final Sequence lastFreeSeq = freeList.getTail(); + + // Sanity check, should not happen if we have a free page ratio + if (lastFreeSeq == null) { + LOG.warn("Skipping compaction, no free pages"); + return; + } + + // If we have a block of free pages then check if it is + // at the end of the file. + // + // If the offset of the last free page + the size of a page + // (to account for the nextFreePage that was allocated already) + // equals the disk size then there are no in use pages after this block. + if (toOffset(lastFreeSeq.getLast()) + pageSize != diskSize) { + LOG.info("Unable to compact, last free page block is not at the end of the file"); + return; + } + + long minFreePages = Math.round(totalPageCount * minFreePageCompactionRatio); + + // we must keep a minimum number of free pages so find the point + // where we can truncate without removing too many pages + long maxPagesToTruncate = freePageCount - minFreePages; + + final Sequence freePagesToDelete; + // If the block of free pages is larger than the max we need to truncate + // then we can split the sequence to only delete the max + if (lastFreeSeq.range() > maxPagesToTruncate) { + var last = lastFreeSeq.getLast(); + var updatedFirst = (last - maxPagesToTruncate) + 1; + freePagesToDelete = new Sequence(updatedFirst, last); + } else { + freePagesToDelete = lastFreeSeq; + } + + // sync on writes to block the async thread from trying to write while + // we are updating the file and truncating + synchronized (writes) { + // Generally this should be empty but need to verify + if (!writes.isEmpty()) { + LOG.warn("Skipping compaction, writes are in flight."); + return; + } + + LOG.debug("Number of free pages to be deleted: {}", freePagesToDelete.range()); + LOG.debug("Disk size of end of file free pages: {}", pageSize * freePagesToDelete.range()); + + if (enablePageCaching) { + pageCache.keySet().removeIf(freePagesToDelete::contains); + } + + long newDiskSize = toOffset(freePagesToDelete.getFirst()); + LOG.debug("Truncating page file to length: {}", newDiskSize); + + // Remove the free pages from the freeList tracking + if (freePagesToDelete == lastFreeSeq) { + freeList.removeLastSequence(); + } else { + freeList.remove(freePagesToDelete); + } + nextFreePageId.getAndAdd(-freePagesToDelete.range()); + + // Now that metadata is all updated, we can truncate the file + // This should be the last step to make sure there is no issue + // updating any of the metadata/tracking before truncation + writeFile.getRaf().setLength(newDiskSize); + if (enableDiskSyncs) { + writeFile.sync(); + } + + LOG.debug("Page file was compacted, new length: {}, old length:{}", newDiskSize, diskSize); + LOG.debug("New page count: {}, free page count: {}", getPageCount(), getFreePageCount()); + } + } + @Override public String toString() { @@ -889,6 +1065,57 @@ public void setUseLFRUEviction(boolean useLFRUEviction) { this.useLFRUEviction = useLFRUEviction; } + public PageFileCompactionStrategy getCompactionStrategy() { + return compactionStrategy; + } + + public void setCompactionStrategy(PageFileCompactionStrategy compactionStrategy) { + assertNotLoaded(); + this.compactionStrategy = Objects.requireNonNull(compactionStrategy); + } + + public float getMinFreePageCompactionRatio() { + return minFreePageCompactionRatio; + } + + /** + * The ratio of the minimum amount of free pages to keep. + * The default will keep a minimum of 10% of free pages, relative to the + * current size of the file when compaction is started. + * + * @param minFreePageCompactionRatio + */ + public void setMinFreePageCompactionRatio(float minFreePageCompactionRatio) { + validateCompactionRatio(minFreePageCompactionRatio, "minFreePageCompactionRatio"); + this.minFreePageCompactionRatio = minFreePageCompactionRatio; + } + + public float getMaxFreePageCompactionRatio() { + return maxFreePageCompactionRatio; + } + + /** + * The ratio of the maximum amount of free pages to allow before triggering compaction. + * The default will trigger a compaction attempt if the percentage of free pages + * hits 30% of the total file size. + * + * @param maxFreePageCompactionRatio + */ + public void setMaxFreePageCompactionRatio(float maxFreePageCompactionRatio) { + validateCompactionRatio(maxFreePageCompactionRatio, "maxFreePageCompactionRatio"); + this.maxFreePageCompactionRatio = maxFreePageCompactionRatio; + } + + private void validateCompactionRatio(float ratio, String name) { + assertNotLoaded(); + if (ratio < 0) { + throw new IllegalArgumentException(name + " must not be negative"); + } + if (ratio > 1) { + throw new IllegalArgumentException(name + " must not be greater than 1"); + } + } + /////////////////////////////////////////////////////////////////// // Package Protected Methods exposed to Transaction /////////////////////////////////////////////////////////////////// diff --git a/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/KahaDBIndexCompactionTest.java b/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/KahaDBIndexCompactionTest.java new file mode 100644 index 00000000000..0f2750b2ccc --- /dev/null +++ b/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/KahaDBIndexCompactionTest.java @@ -0,0 +1,140 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.store.kahadb; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.IOException; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.activemq.command.ActiveMQTextMessage; +import org.apache.activemq.command.MessageAck; +import org.apache.activemq.command.MessageId; +import org.apache.activemq.store.MessageStore; +import org.apache.activemq.store.kahadb.MessageDatabase.StoredDestination; +import org.apache.activemq.store.kahadb.data.KahaDestination; +import org.apache.activemq.store.kahadb.disk.journal.Journal.JournalDiskSyncStrategy; +import org.apache.activemq.store.kahadb.disk.journal.Location; +import org.apache.activemq.store.kahadb.disk.page.PageFile.PageFileCompactionStrategy; +import org.apache.activemq.store.kahadb.disk.page.Transaction; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class KahaDBIndexCompactionTest { + + private static final Logger LOG = LoggerFactory.getLogger(KahaDBIndexCompactionTest.class); + + @Rule + public TemporaryFolder dataDir = new TemporaryFolder(new File("target")); + private final String payload = new String(new byte[1024]); + + private BrokerService broker = null; + private final ActiveMQQueue destination = new ActiveMQQueue("Test"); + private KahaDBPersistenceAdapter adapter; + + + protected void startBroker() throws Exception { + broker = new BrokerService(); + broker.setDeleteAllMessagesOnStartup(true); + broker.setPersistent(true); + broker.setUseJmx(true); + broker.setDataDirectory(dataDir.getRoot().getAbsolutePath()); + adapter = (KahaDBPersistenceAdapter) broker.getPersistenceAdapter(); + adapter.setJournalDiskSyncStrategy(JournalDiskSyncStrategy.NEVER.name()); + adapter.setEnableIndexDiskSyncs(true); + adapter.setIndexCompactionStrategy(PageFileCompactionStrategy.TRUNCATION); + + // disable cleanup task so we can control it int ests + adapter.setCheckpointInterval(0); + adapter.setCleanupInterval(0); + broker.start(); + LOG.info("Starting broker.."); + } + + @Before + public void start() throws Exception { + startBroker(); + } + + @After + public void tearDown() throws Exception { + if (broker != null) { + broker.stop(); + broker.waitUntilStopped(); + } + } + + @Test + public void testTruncate() throws Exception { + final KahaDBStore store = adapter.getStore(); + var pageFile = store.getPageFile(); + + //Add a single message and update once so we can compare the size consistently + MessageStore messageStore = store.createQueueMessageStore(destination); + messageStore.start(); + + for (int i = 0; i < 1000; i++) { + ActiveMQTextMessage textMessage = getMessage(new MessageId("111:222:" + i)); + messageStore.addMessage(broker.getAdminConnectionContext(), textMessage); + } + + long diskSize = pageFile.getDiskSize(); + store.checkpointCleanup(true); + // should not be able to truncate as we added a bunch of data to the index + // but have not cleared it yet + assertEquals(diskSize, store.getPageFile().getDiskSize()); + + for (int i = 0; i < 1000; i++) { + MessageId messageId = new MessageId("111:222:" + i); + MessageAck ack = new MessageAck(); + ack.setMessageID(messageId); + ack.setDestination(messageStore.getDestination()); + ack.setAckType(MessageAck.INDIVIDUAL_ACK_TYPE); + ack.setMessageCount(1); + messageStore.removeMessage(broker.getAdminConnectionContext(), ack); + } + + // get free page size before cleanup + long freePages = pageFile.getFreePageCount(); + long totalPages = pageFile.getPageCount(); + + // After consuming all messages cleanup again, we should be able to truncate + // as there are > 30% free pages in the page file + store.checkpointCleanup(true); + assertTrue(store.getPageFile().getDiskSize() < diskSize); + assertTrue(store.getPageFile().getFreePageCount() < freePages); + assertEquals(totalPages * store.getPageFile().getMinFreePageCompactionRatio(), + pageFile.getFreePageCount(), 1); + } + + + private ActiveMQTextMessage getMessage(final MessageId messageId) throws Exception { + ActiveMQTextMessage textMessage = new ActiveMQTextMessage(); + textMessage.setMessageId(messageId); + textMessage.setText(payload); + + return textMessage; + } + +} diff --git a/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/page/PageFileCompactionTest.java b/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/page/PageFileCompactionTest.java new file mode 100644 index 00000000000..25f52655546 --- /dev/null +++ b/activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/disk/page/PageFileCompactionTest.java @@ -0,0 +1,385 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.store.kahadb.disk.page; + + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import org.apache.activemq.store.kahadb.disk.page.PageFile.PageFileCompactionStrategy; +import org.apache.activemq.store.kahadb.disk.util.StringMarshaller; +import org.apache.activemq.util.Wait; +import org.junit.After; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +public class PageFileCompactionTest { + + @Rule + public TemporaryFolder tempFolder = new TemporaryFolder(); + + private PageFile pf; + + @After + public void tearDown() throws Exception { + if (pf != null) { + try { + pf.unload(); + } catch (Exception ignored) { + // ignore + } + } + } + + // min .1 + // max .3 + @Test + public void testFreeAll() throws IOException { + pf = new PageFile(tempFolder.newFolder(), "pagefile"); + pf.setCompactionStrategy(PageFileCompactionStrategy.TRUNCATION); + pf.load(); + + // write 100 pages with some test data + writePages(pf, 100, true); + + // verify page file properly allocated for all 100 pages + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + + // compaction shouldn't do anything + pf.compact(); + + // file length shouldn't have changed + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + + // free the first 50 pages, so 50% will now be marked as free + freePages(pf, 0, 50, false); + + // Compaction is not able to truncate because the free pages + // are only at the front + pf.compact(); + + // file length shouldn't have changed + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + + // free the remaining pages, but do not flush + freePages(pf, 50, 100, false); + // compaction should not do anything becuase the free pages + // were not yet flushed + pf.compact(); + + // file length shouldn't have changed + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + + // now we flush, compaction should finally truncate + pf.flush(); + pf.compact(); + + // All 100 pages were free, 10 should be left because of the 10% min ratio + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(10), pf.getDiskSize()); + assertEquals(10, pf.getFreePageCount()); + } + + @Test + public void testUnCleanShutdownRecoveryFile() throws Exception { + var tmp = tempFolder.newFolder(); + pf = new PageFile(tmp, "pagefile"); + pf.setCompactionStrategy(PageFileCompactionStrategy.TRUNCATION); + pf.load(); + + // write 100 pages with test data + writePages(pf, 100, true); + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + + // free 30 pages (the threshold) so we compact + freePages(pf, 70, 100, true); + assertEquals(30, pf.getFreePageCount()); + assertEquals(100, pf.getPageCount()); + + pf.compact(); + + // we should have truncated 20 pages + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(80), pf.getDiskSize()); + assertEquals(10, pf.getFreePageCount()); + assertEquals(80, pf.getPageCount()); + + // At this point the PageFile was not closed and shut down properly + // so the free page list wasn't persisted, etc. + + // The recovery file contains the last batch of writes if enabled, so + // in this case the last thing written was to free 30 pages so + // loading will bring back and reallocate the 30 free pages bringing + // the file back to 100 + PageFile pf2 = new PageFile(tmp, "pagefile"); + pf2.load(); + // free page recovery is async + assertTrue(Wait.waitFor(() -> pf2.recoveredFreeList.get() != null, 1000, 1)); + + // free page pages won't be recovered until we flush so all 100 + // pages should still be in use + assertEquals(0, pf2.getFreePageCount()); + assertEquals(100, pf2.getPageCount()); + assertEquals(pf2.toOffset(100), pf2.getDiskSize()); + + // flush merges recovered free pages and writes them again + pf2.flush(); + + // restored back to before compaction + assertEquals(pf2.getFile().length(), pf2.getDiskSize()); + assertEquals(pf2.toOffset(100), pf2.getDiskSize()); + assertEquals(30, pf2.getFreePageCount()); + assertEquals(100, pf2.getPageCount()); + } + + @Test + public void testUnCleanShutdownNoRecoveryFile() throws Exception { + var tmp = tempFolder.newFolder(); + pf = new PageFile(tmp, "pagefile"); + pf.setCompactionStrategy(PageFileCompactionStrategy.TRUNCATION); + // disable recovery file + pf.setEnableRecoveryFile(false); + pf.load(); + + // write 100 pages with test data + writePages(pf, 100, true); + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + + // free 30 pages (the threshold) so we compact + freePages(pf, 70, 100, true); + assertEquals(30, pf.getFreePageCount()); + assertEquals(100, pf.getPageCount()); + + pf.compact(); + + // we should have truncated 20 pages + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(80), pf.getDiskSize()); + assertEquals(10, pf.getFreePageCount()); + assertEquals(80, pf.getPageCount()); + + // At this point the PageFile was not closed and shut down properly + // so the free page list wasn't persisted, etc. On load + // 80 pages come back (none free) because we truncated 20 + + // The recovery file is not active + PageFile pf2 = new PageFile(tmp, "pagefile"); + pf2.load(); + pf2.flush(); + + // restored back to before compaction - the 20 pages were truncated + // and won't come back and are lost without a recovery file + assertEquals(pf2.getFile().length(), pf2.getDiskSize()); + assertEquals(pf2.toOffset(80), pf2.getDiskSize()); + assertEquals(0, pf2.getFreePageCount()); + assertEquals(80, pf2.getPageCount()); + + + + // TODO we also want to test with recovery file still enabled, but + // writing something after compaction and then unclean shutdown + // In that case the recovery file would contain the new writes + // and not the free pages so the compaction should hopefully hold + // Ie if we compacted from 30 to 10 free pages, then write 1 page + // and unclean shutdown, we would recover with 9 free pages left + // as it would replay the 1 new write and not recover the previous + // free pages. + +// assertEquals(pf2.getFile().length(), pf2.getDiskSize()); +// assertEquals(pf2.toOffset(80), pf2.getDiskSize()); +// assertEquals(10, pf2.getFreePageCount()); +// assertEquals(80, pf2.getPageCount()); + + } + + @Test + public void testMinMaxFreePageRatio() throws IOException { + pf = new PageFile(tempFolder.newFolder(), "pagefile"); + pf.setCompactionStrategy(PageFileCompactionStrategy.TRUNCATION); + // keep 15% when compacting + pf.setMinFreePageCompactionRatio(.15f); + // don't compact until we hit 40% free + pf.setMaxFreePageCompactionRatio(.4f); + pf.load(); + + // write 100 pages of test data + writePages(pf, 100, true); + + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + + // free 39 pages, 1 less than the threshold + freePages(pf, 61, 100, true); + pf.compact(); + + // No compaction should have happened + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + assertEquals(39, pf.getFreePageCount()); + assertEquals(100, pf.getPageCount()); + + // free 1 more page, we should now be at the 30% marker + freePages(pf, 60, 61, true); + pf.compact(); + + // We keep 15% of the total size (so 15 pages) which means + // 25 pages get truncated + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(75), pf.getDiskSize()); + assertEquals(15, pf.getFreePageCount()); + assertEquals(75, pf.getPageCount()); + + } + + @Test + public void testNoMinMax() throws IOException { + pf = new PageFile(tempFolder.newFolder(), "pagefile"); + pf.setCompactionStrategy(PageFileCompactionStrategy.TRUNCATION); + // Setting min to 0 means there is no minimum free pages to retain + // so we will drop them all + pf.setMinFreePageCompactionRatio(0); + // Setting max to 0 means there's no threshold to start truncation so + // it will try and compact as soon as possible + pf.setMaxFreePageCompactionRatio(0); + pf.load(); + + // allocate 50 pages with data, and 50 free + writePages(pf, 50, true); + allocateFree(pf, 50, true); + + assertEquals(100, pf.getPageCount()); + assertEquals(50, pf.getFreePageCount()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + + // this should truncate all 50 free pages + pf.compact(); + + assertEquals(50, pf.getPageCount()); + assertEquals(0, pf.getFreePageCount()); + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(50), pf.getDiskSize()); + + // all should be removed now + freePages(pf, 0, 50, true); + pf.compact(); + + assertEquals(pf.getFile().length(), pf.getDiskSize()); + assertEquals(pf.toOffset(0), pf.getDiskSize()); + } + + @Test + public void testNever() throws IOException { + // Default is to never compact + pf = new PageFile(tempFolder.newFolder(), "pagefile"); + pf.load(); + + // allocate all free pages + allocateFree(pf, 100, true); + + assertEquals(100, pf.getFreePageCount()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + + pf.compact(); + + // compaction is disabled, should not compact + assertEquals(100, pf.getFreePageCount()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + + // enable but set min/max to 100 so will never compact + pf.unload(); + pf.setCompactionStrategy(PageFileCompactionStrategy.TRUNCATION); + pf.setMinFreePageCompactionRatio(1); + pf.setMaxFreePageCompactionRatio(1); + pf.load(); + pf.compact(); + + assertEquals(100, pf.getFreePageCount()); + assertEquals(pf.toOffset(100), pf.getDiskSize()); + } + + @Test + public void testConfigValidation() throws IOException { + // Default is to never compact + pf = new PageFile(tempFolder.newFolder(), "pagefile"); + + assertThrows(NullPointerException.class, () -> pf.setCompactionStrategy(null)); + assertThrows(IllegalArgumentException.class, () -> pf.setMinFreePageCompactionRatio(-.01f)); + assertThrows(IllegalArgumentException.class, () -> pf.setMinFreePageCompactionRatio(1.00001f)); + assertThrows(IllegalArgumentException.class, () -> pf.setMaxFreePageCompactionRatio(-.01f)); + assertThrows(IllegalArgumentException.class, () -> pf.setMaxFreePageCompactionRatio(1.00001f)); + + // should be no errors + pf.setMinFreePageCompactionRatio(0f); + pf.setMinFreePageCompactionRatio(1f); + pf.setMaxFreePageCompactionRatio(0f); + pf.setMaxFreePageCompactionRatio(1f); + + // max can't be less than min + pf.setMaxFreePageCompactionRatio(.2f); + pf.setMinFreePageCompactionRatio(.3f); + assertThrows(IllegalStateException.class, () -> pf.load()); + } + + + private void allocateFree(PageFile pf, int pageCount, boolean flush) throws IOException { + // Insert some data into the page file. + Transaction tx = pf.tx(); + for (int i = 0; i < pageCount; i++) { + Page page = tx.allocate(); + assertEquals(Page.PAGE_FREE_TYPE, page.getType()); + } + tx.commit(); + if (flush) { + pf.flush(); + } + } + + private void writePages(PageFile pf, int pageCount, boolean flush) throws IOException { + // Insert some data into the page file. + Transaction tx = pf.tx(); + for (int i = 0; i < pageCount; i++) { + Page page = tx.allocate(); + assertEquals(Page.PAGE_FREE_TYPE, page.getType()); + page.set("page:" + i); + tx.store(page, StringMarshaller.INSTANCE, false); + } + tx.commit(); + if (flush) { + pf.flush(); + } + } + + private void freePages(PageFile pf, int start, int end, boolean flush) throws IOException { + Transaction tx = pf.tx(); + for (int i = start; i < end; i++) { + tx.free(i); + } + tx.commit(); + if (flush) { + pf.flush(); + } + } +}