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(); + } + } +}