| /* |
| * 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.jackrabbit.oak.segment; |
| |
| import static org.apache.jackrabbit.guava.common.collect.Lists.newArrayList; |
| import static org.apache.jackrabbit.guava.common.collect.Sets.newHashSet; |
| import static org.apache.jackrabbit.guava.common.util.concurrent.Uninterruptibles.awaitUninterruptibly; |
| import static org.apache.jackrabbit.guava.common.util.concurrent.Uninterruptibles.sleepUninterruptibly; |
| import static java.lang.Integer.getInteger; |
| import static java.lang.String.valueOf; |
| import static java.util.concurrent.Executors.newFixedThreadPool; |
| import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor; |
| import static java.util.concurrent.TimeUnit.MILLISECONDS; |
| import static java.util.concurrent.TimeUnit.SECONDS; |
| import static org.apache.commons.io.FileUtils.byteCountToDisplaySize; |
| import static org.apache.jackrabbit.oak.api.Type.STRING; |
| import static org.apache.jackrabbit.oak.commons.PathUtils.concat; |
| import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE; |
| import static org.apache.jackrabbit.oak.segment.ClassicCompactor.UPDATE_LIMIT; |
| import static org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions.defaultGCOptions; |
| import static org.apache.jackrabbit.oak.segment.file.FileStoreBuilder.fileStoreBuilder; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| import static org.junit.Assume.assumeTrue; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.File; |
| import java.io.IOException; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Random; |
| import java.util.Set; |
| import java.util.UUID; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.FutureTask; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.function.Consumer; |
| |
| import org.apache.jackrabbit.guava.common.io.ByteStreams; |
| import org.apache.jackrabbit.oak.api.Blob; |
| import org.apache.jackrabbit.oak.api.CommitFailedException; |
| import org.apache.jackrabbit.oak.api.PropertyState; |
| import org.apache.jackrabbit.oak.api.Type; |
| import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser; |
| import org.apache.jackrabbit.oak.plugins.blob.datastore.DataStoreBlobStore; |
| import org.apache.jackrabbit.oak.plugins.blob.datastore.OakFileDataStore; |
| import org.apache.jackrabbit.oak.plugins.memory.StringPropertyState; |
| import org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions; |
| import org.apache.jackrabbit.oak.segment.file.FileStore; |
| import org.apache.jackrabbit.oak.segment.file.FileStoreGCMonitor; |
| import org.apache.jackrabbit.oak.segment.file.InvalidFileStoreVersionException; |
| import org.apache.jackrabbit.oak.segment.tool.Compact; |
| import org.apache.jackrabbit.oak.spi.blob.BlobStore; |
| import org.apache.jackrabbit.oak.spi.commit.CommitInfo; |
| import org.apache.jackrabbit.oak.spi.commit.EmptyHook; |
| import org.apache.jackrabbit.oak.spi.state.ChildNodeEntry; |
| import org.apache.jackrabbit.oak.spi.state.NodeBuilder; |
| import org.apache.jackrabbit.oak.spi.state.NodeState; |
| import org.apache.jackrabbit.oak.spi.state.NodeStore; |
| import org.apache.jackrabbit.oak.stats.Clock; |
| import org.apache.jackrabbit.oak.stats.DefaultStatisticsProvider; |
| import org.apache.jackrabbit.oak.stats.StatisticsProvider; |
| import org.junit.BeforeClass; |
| import org.junit.Rule; |
| import org.junit.Test; |
| import org.junit.rules.TemporaryFolder; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| public class CompactionAndCleanupIT { |
| |
| private static final Logger log = LoggerFactory |
| .getLogger(CompactionAndCleanupIT.class); |
| |
| @Rule |
| public TemporaryFolder folder = new TemporaryFolder(new File("target")); |
| |
| @BeforeClass |
| public static void init() { |
| // Allow running gc without backoff. Needed for testCancelCompactionSNFE. |
| // See FileStore.GC_BACKOFF. |
| System.setProperty("oak.gc.backoff", "0"); |
| } |
| |
| private File getFileStoreFolder() { |
| return folder.getRoot(); |
| } |
| |
| @Test |
| public void compactPersistsHead() throws Exception { |
| FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withGCOptions(defaultGCOptions().setRetainedGenerations(2)) |
| .withMaxFileSize(1) |
| .build(); |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| |
| try { |
| // Create ~2MB of data |
| NodeBuilder extra = nodeStore.getRoot().builder(); |
| NodeBuilder content = extra.child("content"); |
| for (int i = 0; i < 10000; i++) { |
| NodeBuilder c = content.child("c" + i); |
| for (int j = 0; j < 1000; j++) { |
| c.setProperty("p" + i, "v" + i); |
| } |
| } |
| nodeStore.merge(extra, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| fileStore.compactFull(); |
| assertEquals(fileStore.getRevisions().getHead(), fileStore.getRevisions().getPersistedHead()); |
| } finally { |
| fileStore.close(); |
| } |
| } |
| |
| @Test |
| public void compactionNoBinaryClone() throws Exception { |
| ScheduledExecutorService executor = newSingleThreadScheduledExecutor(); |
| FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withGCOptions(defaultGCOptions().setRetainedGenerations(2)) |
| .withStatisticsProvider(new DefaultStatisticsProvider(executor)) |
| .withMaxFileSize(1) |
| .build(); |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| |
| try { |
| // 5MB blob |
| int blobSize = 5 * 1024 * 1024; |
| |
| // Create ~2MB of data |
| NodeBuilder extra = nodeStore.getRoot().builder(); |
| NodeBuilder content = extra.child("content"); |
| for (int i = 0; i < 10000; i++) { |
| NodeBuilder c = content.child("c" + i); |
| for (int j = 0; j < 1000; j++) { |
| c.setProperty("p" + i, "v" + i); |
| } |
| } |
| nodeStore.merge(extra, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| long size1 = fileStore.getStats().getApproximateSize(); |
| log.debug("File store size {}", byteCountToDisplaySize(size1)); |
| |
| // Create a property with 5 MB blob |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setProperty("blob1", createBlob(nodeStore, blobSize)); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| long size2 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the store should grow", size2 > size1); |
| assertTrue("the store should grow of at least the size of the blob", size2 - size1 >= blobSize); |
| |
| // Now remove the property. No gc yet -> size doesn't shrink |
| builder = nodeStore.getRoot().builder(); |
| builder.removeProperty("blob1"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| long size3 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the store should grow", size3 > size2); |
| |
| // 1st gc cycle -> no reclaimable garbage... |
| fileStore.compactFull(); |
| fileStore.cleanup(); |
| |
| long size4 = fileStore.getStats().getApproximateSize(); |
| |
| // Add another 5MB binary doubling the blob size |
| builder = nodeStore.getRoot().builder(); |
| builder.setProperty("blob2", createBlob(nodeStore, blobSize)); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| long size5 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the store should grow", size5 > size4); |
| assertTrue("the store should grow of at least the size of the blob", size5 - size4 >= blobSize); |
| |
| // 2st gc cycle -> 1st blob should get collected |
| fileStore.compactFull(); |
| fileStore.cleanup(); |
| |
| long size6 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the store should shrink", size6 < size5); |
| assertTrue("the store should shrink of at least the size of the blob", size5 - size6 >= blobSize); |
| |
| // 3rtd gc cycle -> no significant change |
| fileStore.compactFull(); |
| fileStore.cleanup(); |
| |
| long size7 = fileStore.getStats().getApproximateSize(); |
| |
| // No data loss |
| byte[] blob = ByteStreams.toByteArray(nodeStore.getRoot() |
| .getProperty("blob2").getValue(Type.BINARY).getNewStream()); |
| assertEquals(blobSize, blob.length); |
| } finally { |
| fileStore.close(); |
| } |
| } |
| |
| @Test |
| public void offlineCompaction() throws Exception { |
| SegmentGCOptions gcOptions = defaultGCOptions().setOffline(); |
| ScheduledExecutorService executor = newSingleThreadScheduledExecutor(); |
| FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withMaxFileSize(1) |
| .withGCOptions(gcOptions) |
| .withStatisticsProvider(new DefaultStatisticsProvider(executor)) |
| .build(); |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| |
| try { |
| // 5MB blob |
| int blobSize = 5 * 1024 * 1024; |
| |
| // Create ~2MB of data |
| NodeBuilder extra = nodeStore.getRoot().builder(); |
| NodeBuilder content = extra.child("content"); |
| for (int i = 0; i < 10000; i++) { |
| NodeBuilder c = content.child("c" + i); |
| for (int j = 0; j < 1000; j++) { |
| c.setProperty("p" + i, "v" + i); |
| } |
| } |
| nodeStore.merge(extra, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| long size1 = fileStore.getStats().getApproximateSize(); |
| log.debug("File store size {}", byteCountToDisplaySize(size1)); |
| |
| // Create a property with 5 MB blob |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setProperty("blob1", createBlob(nodeStore, blobSize)); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| long size2 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the store should grow", size2 > size1); |
| assertTrue("the store should grow of at least the size of the blob", size2 - size1 > blobSize); |
| |
| // Now remove the property. No gc yet -> size doesn't shrink |
| builder = nodeStore.getRoot().builder(); |
| builder.removeProperty("blob1"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| long size3 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the size should grow", size3 > size2); |
| |
| // 1st gc cycle -> 1st blob should get collected |
| fileStore.compactFull(); |
| fileStore.cleanup(); |
| |
| long size4 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the store should shrink", size4 < size3); |
| assertTrue("the store should shrink of at least the size of the blob", size3 - size4 >= blobSize); |
| |
| // Add another 5MB binary |
| builder = nodeStore.getRoot().builder(); |
| builder.setProperty("blob2", createBlob(nodeStore, blobSize)); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| long size5 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the store should grow", size5 > size4); |
| assertTrue("the store should grow of at least the size of the blob", size5 - size4 > blobSize); |
| |
| // 2st gc cycle -> 2nd blob should *not* be collected |
| fileStore.compactFull(); |
| fileStore.cleanup(); |
| |
| long size6 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the blob should not be collected", size6 > blobSize); |
| |
| // 3rd gc cycle -> no significant change |
| fileStore.compactFull(); |
| fileStore.cleanup(); |
| |
| long size7 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the blob should not be collected", size7 > blobSize); |
| |
| // No data loss |
| byte[] blob = ByteStreams.toByteArray(nodeStore.getRoot() |
| .getProperty("blob2").getValue(Type.BINARY).getNewStream()); |
| assertEquals(blobSize, blob.length); |
| } finally { |
| fileStore.close(); |
| } |
| } |
| |
| @Test |
| public void cancelOfflineCompaction() throws Exception { |
| final AtomicBoolean cancelCompaction = new AtomicBoolean(true); |
| try (FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withGCOptions(defaultGCOptions().setOffline()) |
| .build()) { |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| // Create ~2MB of data |
| NodeBuilder extra = nodeStore.getRoot().builder(); |
| NodeBuilder content = extra.child("content"); |
| for (int i = 0; i < 10000; i++) { |
| NodeBuilder c = content.child("c" + i); |
| for (int j = 0; j < 1000; j++) { |
| c.setProperty("p" + i, "v" + i); |
| } |
| } |
| nodeStore.merge(extra, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| NodeState uncompactedRoot = nodeStore.getRoot(); |
| |
| // Keep cancelling compaction |
| new Thread(() -> { |
| while (cancelCompaction.get()) { |
| fileStore.cancelGC(); |
| } |
| }).start(); |
| |
| fileStore.compactFull(); |
| |
| // Cancelling compaction must not corrupt the repository. See OAK-7050. |
| NodeState compactedRoot = nodeStore.getRoot(); |
| assertTrue(compactedRoot.exists()); |
| assertEquals(uncompactedRoot, compactedRoot); |
| } finally { |
| cancelCompaction.set(false); |
| } |
| } |
| |
| /** |
| * Create a lot of data nodes (no binaries) and a few checkpoints, verify |
| * that compacting checkpoints will not cause the size to explode |
| */ |
| @Test |
| public void offlineCompactionCps() throws Exception { |
| SegmentGCOptions gcOptions = defaultGCOptions().setOffline(); |
| ScheduledExecutorService executor = newSingleThreadScheduledExecutor(); |
| FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withMaxFileSize(1) |
| .withGCOptions(gcOptions) |
| .withStatisticsProvider(new DefaultStatisticsProvider(executor)) |
| .build(); |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| try { |
| // Create ~2MB of data |
| NodeBuilder extra = nodeStore.getRoot().builder(); |
| NodeBuilder content = extra.child("content"); |
| for (int i = 0; i < 10000; i++) { |
| NodeBuilder c = content.child("c" + i); |
| for (int j = 0; j < 1000; j++) { |
| c.setProperty("p" + i, "v" + i); |
| } |
| } |
| nodeStore.merge(extra, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| fileStore.compactFull(); |
| fileStore.cleanup(); |
| // Compacts to 548Kb |
| long size0 = fileStore.getStats().getApproximateSize(); |
| |
| int cpNo = 4; |
| Set<String> cps = new HashSet<String>(); |
| for (int i = 0; i < cpNo; i++) { |
| cps.add(nodeStore.checkpoint(60000)); |
| } |
| assertEquals(cpNo, cps.size()); |
| for (String cp : cps) { |
| assertTrue(nodeStore.retrieve(cp) != null); |
| } |
| |
| long size1 = fileStore.getStats().getApproximateSize(); |
| assertTrue("the size should grow or stay the same", size1 >= size0); |
| |
| // TODO the following assertion doesn't say anything useful. The |
| // conveyed message is "the repository can shrink, grow or stay the |
| // same, as long as it remains in a 10% margin of the previous size |
| // that I took out of thin air". It has to be fixed or removed. |
| |
| // fileStore.compact(); |
| // fileStore.cleanup(); |
| // long size2 = fileStore.getStats().getApproximateSize(); |
| // assertSize("with checkpoints compacted", size2, size1 * 9/10, size1 * 11 / 10); |
| } finally { |
| fileStore.close(); |
| } |
| } |
| |
| @Test |
| public void equalContentAfterOC() throws Exception { |
| SegmentGCOptions gcOptions = defaultGCOptions().setOffline(); |
| ScheduledExecutorService executor = newSingleThreadScheduledExecutor(); |
| |
| try (FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withGCOptions(gcOptions) |
| .build()) { |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| |
| // Add initial content |
| NodeBuilder rootBuilder = nodeStore.getRoot().builder(); |
| addNodes(rootBuilder, 8, "p"); |
| addProperties(rootBuilder, 3); |
| nodeStore.merge(rootBuilder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| |
| NodeState initialRoot = nodeStore.getRoot(); |
| assertTrue(fileStore.compactFull()); |
| NodeState compactedRoot = nodeStore.getRoot(); |
| |
| assertTrue(initialRoot != compactedRoot); |
| assertEquals(initialRoot, compactedRoot); |
| } |
| } |
| |
| /** |
| * Create 2 binary nodes with same content and same reference. Verify |
| * de-duplication capabilities of compaction |
| */ |
| @Test |
| public void offlineCompactionBinR1() throws Exception { |
| SegmentGCOptions gcOptions = defaultGCOptions().setOffline(); |
| ScheduledExecutorService executor = newSingleThreadScheduledExecutor(); |
| FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withMaxFileSize(1) |
| .withGCOptions(gcOptions) |
| .withStatisticsProvider(new DefaultStatisticsProvider(executor)) |
| .build(); |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders |
| .builder(fileStore).build(); |
| |
| try { |
| NodeBuilder extra = nodeStore.getRoot().builder(); |
| NodeBuilder content = extra.child("content"); |
| |
| int blobSize = 5 * 1024 * 1024; |
| byte[] data = new byte[blobSize]; |
| new Random().nextBytes(data); |
| Blob b = nodeStore.createBlob(new ByteArrayInputStream(data)); |
| |
| NodeBuilder c1 = content.child("c1"); |
| c1.setProperty("blob1", b); |
| NodeBuilder c2 = content.child("c2"); |
| c2.setProperty("blob2", b); |
| nodeStore.merge(extra, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| int cpNo = 4; |
| Set<String> cps = new HashSet<String>(); |
| for (int i = 0; i < cpNo; i++) { |
| cps.add(nodeStore.checkpoint(60000)); |
| } |
| assertEquals(cpNo, cps.size()); |
| for (String cp : cps) { |
| assertTrue(nodeStore.retrieve(cp) != null); |
| } |
| |
| // 5Mb, de-duplication by the SegmentWriter |
| long size1 = fileStore.getStats().getApproximateSize(); |
| fileStore.compactFull(); |
| fileStore.cleanup(); |
| long size2 = fileStore.getStats().getApproximateSize(); |
| assertSize("with compacted binaries", size2, 0, size1 * 11 / 10); |
| } finally { |
| fileStore.close(); |
| } |
| } |
| |
| /** |
| * Test for the Offline compaction tool (OAK-5971) |
| */ |
| @Test |
| public void offlineCompactionTool() throws Exception { |
| SegmentGCOptions gcOptions = defaultGCOptions().setOffline(); |
| ScheduledExecutorService executor = newSingleThreadScheduledExecutor(); |
| FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withMaxFileSize(1) |
| .withGCOptions(gcOptions) |
| .withStatisticsProvider(new DefaultStatisticsProvider(executor)) |
| .build(); |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| try { |
| NodeBuilder root = nodeStore.getRoot().builder(); |
| root.child("content"); |
| nodeStore.merge(root, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| } finally { |
| fileStore.close(); |
| } |
| |
| Compact.builder().withPath(getFileStoreFolder()).build().run(); |
| |
| fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withMaxFileSize(1) |
| .withGCOptions(gcOptions) |
| .withStatisticsProvider(new DefaultStatisticsProvider(executor)) |
| .build(); |
| nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| try { |
| assertTrue(nodeStore.getRoot().hasChildNode("content")); |
| } finally { |
| fileStore.close(); |
| } |
| } |
| |
| private static void assertSize(String info, long size, long lower, long upper) { |
| log.debug("File Store {} size {}, expected in interval [{},{}]", |
| info, size, lower, upper); |
| assertTrue("File Store " + info + " size expected in interval " + |
| "[" + (lower) + "," + (upper) + "] but was: " + (size), |
| size >= lower && size <= (upper)); |
| } |
| |
| private static Blob createBlob(NodeStore nodeStore, int size) throws IOException { |
| byte[] data = new byte[size]; |
| new Random().nextBytes(data); |
| return nodeStore.createBlob(new ByteArrayInputStream(data)); |
| } |
| |
| @Test |
| public void testCancelCompaction() |
| throws Throwable { |
| final FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withGCOptions(defaultGCOptions().setRetainedGenerations(2)) |
| .withMaxFileSize(1) |
| .build(); |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| addNodes(builder, 10, ""); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| FutureTask<Boolean> async = runAsync(new Callable<Boolean>() { |
| @Override |
| public Boolean call() throws IOException { |
| boolean cancelled = false; |
| for (int k = 0; !cancelled && k < 1000; k++) { |
| cancelled = !fileStore.compactFull(); |
| } |
| return cancelled; |
| } |
| }); |
| |
| // Give the compaction thread a head start |
| sleepUninterruptibly(1, SECONDS); |
| |
| fileStore.close(); |
| try { |
| assertTrue(async.get()); |
| } catch (ExecutionException e) { |
| if (!(e.getCause() instanceof IllegalStateException)) { |
| // Throw cause unless this is an ISE thrown by the |
| // store being already closed, which is kinda expected |
| throw e.getCause(); |
| } |
| } |
| } |
| |
| /** |
| * See OAK-5517: SNFE when running compaction after a cancelled gc |
| */ |
| @Test |
| public void testCancelCompactionSNFE() |
| throws Throwable { |
| final FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withGCOptions(defaultGCOptions() |
| .setRetainedGenerations(2) |
| .setEstimationDisabled(true)) |
| .withMaxFileSize(1) |
| .build(); |
| try { |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| |
| final Callable<Void> cancel = new Callable<Void>() { |
| @Override |
| public Void call() throws Exception { |
| // Give the compaction thread a head start |
| sleepUninterruptibly(1000, MILLISECONDS); |
| fileStore.cancelGC(); |
| return null; |
| } |
| }; |
| |
| for (int k = 0; k < 100; k++) { |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| addNodes(builder, 10, k + "-"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| // Cancelling gc should not cause a SNFE on subsequent gc runs |
| runAsync(cancel); |
| fileStore.fullGC(); |
| } |
| } finally { |
| fileStore.close(); |
| } |
| } |
| |
| private static void addNodes(NodeBuilder builder, int depth, String prefix) { |
| if (depth > 0) { |
| NodeBuilder child1 = builder.setChildNode(prefix + "1"); |
| addNodes(child1, depth - 1, prefix); |
| NodeBuilder child2 = builder.setChildNode(prefix + "2"); |
| addNodes(child2, depth - 1, prefix); |
| } |
| } |
| |
| private static void addProperties(NodeBuilder builder, int count) { |
| for (int c = 0; c < count; c++) { |
| builder.setProperty("p-" + c, "v-" + c); |
| } |
| for (String child : builder.getChildNodeNames()) { |
| addProperties(builder.getChildNode(child), count); |
| } |
| } |
| |
| /** |
| * Regression test for OAK-2192 testing for mixed segments. This test does not |
| * cover OAK-3348. I.e. it does not assert the segment graph is free of cross |
| * gc generation references. |
| */ |
| @Test |
| public void testMixedSegments() throws Exception { |
| FileStore store = fileStoreBuilder(getFileStoreFolder()) |
| .withMaxFileSize(2) |
| .withMemoryMapping(true) |
| .build(); |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(store).build(); |
| AtomicBoolean compactionSuccess = new AtomicBoolean(true); |
| |
| NodeBuilder root = nodeStore.getRoot().builder(); |
| createNodes(root.setChildNode("test"), 10, 3); |
| nodeStore.merge(root, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| |
| Set<UUID> beforeSegments = new HashSet<UUID>(); |
| collectSegments(store.getReader(), store.getHead().getRecordId(), |
| segmentId -> beforeSegments.add(segmentId.asUUID())); |
| |
| |
| final AtomicReference<Boolean> run = new AtomicReference<Boolean>(true); |
| final List<Exception> failedCommits = newArrayList(); |
| Thread[] threads = new Thread[10]; |
| for (int k = 0; k < threads.length; k++) { |
| final int threadId = k; |
| threads[k] = new Thread(() -> { |
| for (int j = 0; run.get(); j++) { |
| String nodeName = "b-" + threadId + "," + j; |
| try { |
| NodeBuilder changes = nodeStore.getRoot().builder(); |
| changes.setChildNode(nodeName); |
| nodeStore.merge(changes, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| Thread.sleep(5); |
| } catch (InterruptedException e) { |
| Thread.interrupted(); |
| break; |
| } catch (Exception e) { |
| failedCommits.add(new ExecutionException("Failed commit " + nodeName, e)); |
| } |
| } |
| }); |
| threads[k].start(); |
| } |
| store.compactFull(); |
| run.set(false); |
| for (Thread t : threads) { |
| t.join(); |
| } |
| store.flush(); |
| |
| assumeTrue("Failed to acquire compaction lock", compactionSuccess.get()); |
| for (Exception failedCommit : failedCommits) { |
| throw new Exception("A background commit failed", failedCommit); |
| } |
| |
| Set<UUID> afterSegments = new HashSet<UUID>(); |
| collectSegments(store.getReader(), store.getHead().getRecordId(), |
| segmentId -> afterSegments.add(segmentId.asUUID())); |
| try { |
| for (UUID u : beforeSegments) { |
| assertFalse("Mixed segments found: " + u, afterSegments.contains(u)); |
| } |
| } finally { |
| store.close(); |
| } |
| } |
| |
| @Test |
| public void testMixedSegmentsGCGeneration() throws Exception { |
| try (FileStore store = fileStoreBuilder(getFileStoreFolder()) |
| .withMaxFileSize(2) |
| .withMemoryMapping(true) |
| .build()) { |
| |
| CountDownLatch readyToCompact = new CountDownLatch(1); |
| CountDownLatch compactionCompleted = new CountDownLatch(1); |
| SegmentNodeBuilder changes = store.getHead().builder(); |
| changes.setProperty("a", "a"); |
| changes.setProperty(new StringPropertyState("b", "b") { |
| @Override |
| public String getValue() { |
| readyToCompact.countDown(); |
| awaitUninterruptibly(compactionCompleted); |
| return super.getValue(); |
| } |
| }); |
| |
| // Overlap an ongoing write operation triggered by the call to getNodeState |
| // with a full compaction. This should not cause the written node state |
| // to reference segments from multiple generations |
| FutureTask<SegmentNodeState> futureNodeState = runAsync(changes::getNodeState); |
| readyToCompact.await(); |
| |
| store.compactFull(); |
| compactionCompleted.countDown(); |
| |
| // The node state from the write operation that started before |
| // compaction completed should reference only segments from generation 0. |
| SegmentNodeState gen0NodeState = futureNodeState.get(); |
| collectSegments(store.getReader(), gen0NodeState.getRecordId(), |
| segmentId -> assertEquals("Full generation should be 0", |
| 0, segmentId.getGcGeneration().getFullGeneration())); |
| |
| // Retrieving the node state again should trigger a rewriting to |
| // the next generation (1). |
| SegmentNodeState gen1NodeState = changes.getNodeState(); |
| collectSegments(store.getReader(), gen1NodeState.getRecordId(), |
| segmentId -> assertEquals("Full generation should be 1", |
| 1, segmentId.getGcGeneration().getFullGeneration())); |
| } |
| } |
| |
| /** |
| * Set a root node referring to a child node that lives in a different segments. Depending |
| * on the order how the SegmentBufferWriters associated with the threads used to create the |
| * nodes are flushed, this will introduce a forward reference between the segments. |
| * The current cleanup mechanism cannot handle forward references and removes the referenced |
| * segment causing a SNFE. |
| * This is a regression introduced with OAK-1828. |
| */ |
| @Test |
| public void cleanupCyclicGraph() throws Exception { |
| FileStore fileStore = fileStoreBuilder(getFileStoreFolder()).build(); |
| final SegmentReader reader = fileStore.getReader(); |
| final SegmentWriter writer = fileStore.getWriter(); |
| final BlobStore blobStore = fileStore.getBlobStore(); |
| final SegmentNodeState oldHead = fileStore.getHead(); |
| |
| final SegmentNodeState child = run(new Callable<SegmentNodeState>() { |
| @Override |
| public SegmentNodeState call() throws Exception { |
| NodeBuilder builder = EMPTY_NODE.builder(); |
| return new SegmentNodeState(reader, writer, blobStore, writer.writeNode(EMPTY_NODE)); |
| } |
| }); |
| SegmentNodeState newHead = run(new Callable<SegmentNodeState>() { |
| @Override |
| public SegmentNodeState call() throws Exception { |
| NodeBuilder builder = oldHead.builder(); |
| builder.setChildNode("child", child); |
| return new SegmentNodeState(reader, writer, blobStore, writer.writeNode(builder.getNodeState())); |
| } |
| }); |
| |
| writer.flush(); |
| fileStore.getRevisions().setHead(oldHead.getRecordId(), newHead.getRecordId()); |
| fileStore.close(); |
| |
| fileStore = fileStoreBuilder(getFileStoreFolder()).build(); |
| |
| traverse(fileStore.getHead()); |
| fileStore.cleanup(); |
| |
| // Traversal after cleanup might result in an SNFE |
| traverse(fileStore.getHead()); |
| |
| fileStore.close(); |
| } |
| |
| private static void traverse(NodeState node) { |
| for (ChildNodeEntry childNodeEntry : node.getChildNodeEntries()) { |
| traverse(childNodeEntry.getNodeState()); |
| } |
| } |
| |
| private static <T> T run(Callable<T> callable) throws InterruptedException, ExecutionException { |
| FutureTask<T> task = new FutureTask<T>(callable); |
| new Thread(task).start(); |
| return task.get(); |
| } |
| |
| private static <T> FutureTask<T> runAsync(Callable<T> callable) { |
| FutureTask<T> task = new FutureTask<T>(callable); |
| new Thread(task).start(); |
| return task; |
| } |
| |
| /** |
| * Test asserting OAK-3348: Cross gc sessions might introduce references to pre-compacted segments |
| */ |
| @Test |
| public void preCompactionReferences() throws Exception { |
| for (String ref : new String[] {"merge-before-compact", "merge-after-compact"}) { |
| File repoDir = new File(getFileStoreFolder(), ref); |
| FileStore fileStore = fileStoreBuilder(repoDir) |
| .withMaxFileSize(2) |
| .withGCOptions(defaultGCOptions()) |
| .build(); |
| final SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| try { |
| // add some content |
| NodeBuilder preGCBuilder = nodeStore.getRoot().builder(); |
| preGCBuilder.setChildNode("test").setProperty("blob", createBlob(nodeStore, 1024 * 1024)); |
| nodeStore.merge(preGCBuilder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| |
| // remove it again so we have something to gc |
| preGCBuilder = nodeStore.getRoot().builder(); |
| preGCBuilder.getChildNode("test").remove(); |
| nodeStore.merge(preGCBuilder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| |
| // with a new builder simulate exceeding the update limit. |
| // This will cause changes to be pre-written to segments |
| preGCBuilder = nodeStore.getRoot().builder(); |
| preGCBuilder.setChildNode("test").setChildNode("a").setChildNode("b").setProperty("foo", "bar"); |
| for (int k = 0; k < getInteger("update.limit", 10000); k += 2) { |
| preGCBuilder.setChildNode("dummy").remove(); |
| } |
| |
| // case 1: merge above changes before compact |
| if ("merge-before-compact".equals(ref)) { |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setChildNode("n"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| nodeStore.merge(preGCBuilder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| } |
| |
| // Ensure cleanup is efficient by surpassing the number of |
| // retained generations |
| for (int k = 0; k < defaultGCOptions().getRetainedGenerations(); k++) { |
| fileStore.compactFull(); |
| } |
| |
| // case 2: merge above changes after compact |
| if ("merge-after-compact".equals(ref)) { |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setChildNode("n"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| nodeStore.merge(preGCBuilder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| } |
| } finally { |
| fileStore.close(); |
| } |
| |
| // Re-initialise the file store to simulate off-line gc |
| fileStore = fileStoreBuilder(repoDir).withMaxFileSize(2).build(); |
| try { |
| // The 1M blob should get gc-ed |
| fileStore.cleanup(); |
| assertTrue(ref + " repository size " + fileStore.getStats().getApproximateSize() + " < " + 1024 * 1024, |
| fileStore.getStats().getApproximateSize() < 1024 * 1024); |
| } finally { |
| fileStore.close(); |
| } |
| } |
| } |
| |
| private static void collectSegments(SegmentReader reader, RecordId headId, Consumer<SegmentId> onSegment) { |
| new SegmentParser(reader) { |
| @Override |
| protected void onNode(RecordId parentId, RecordId nodeId) { |
| super.onNode(parentId, nodeId); |
| onSegment.accept(nodeId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onTemplate(RecordId parentId, RecordId templateId) { |
| super.onTemplate(parentId, templateId); |
| onSegment.accept(templateId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onMap(RecordId parentId, RecordId mapId, MapRecord map) { |
| super.onMap(parentId, mapId, map); |
| onSegment.accept(mapId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onMapDiff(RecordId parentId, RecordId mapId, MapRecord map) { |
| super.onMapDiff(parentId, mapId, map); |
| onSegment.accept(mapId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onMapLeaf(RecordId parentId, RecordId mapId, MapRecord map) { |
| super.onMapLeaf(parentId, mapId, map); |
| onSegment.accept(mapId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onMapBranch(RecordId parentId, RecordId mapId, MapRecord map) { |
| super.onMapBranch(parentId, mapId, map); |
| onSegment.accept(mapId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onProperty(RecordId parentId, RecordId propertyId, PropertyTemplate template) { |
| super.onProperty(parentId, propertyId, template); |
| onSegment.accept(propertyId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onValue(RecordId parentId, RecordId valueId, Type<?> type) { |
| super.onValue(parentId, valueId, type); |
| onSegment.accept(valueId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onBlob(RecordId parentId, RecordId blobId) { |
| super.onBlob(parentId, blobId); |
| onSegment.accept(blobId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onString(RecordId parentId, RecordId stringId) { |
| super.onString(parentId, stringId); |
| onSegment.accept(stringId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onList(RecordId parentId, RecordId listId, int count) { |
| super.onList(parentId, listId, count); |
| onSegment.accept(listId.getSegmentId()); |
| } |
| |
| @Override |
| protected void onListBucket(RecordId parentId, RecordId listId, int index, int count, int capacity) { |
| super.onListBucket(parentId, listId, index, count, capacity); |
| onSegment.accept(listId.getSegmentId()); |
| } |
| }.parseNode(headId); |
| } |
| |
| private static void createNodes(NodeBuilder builder, int count, int depth) { |
| if (depth > 0) { |
| for (int k = 0; k < count; k++) { |
| NodeBuilder child = builder.setChildNode("node" + k); |
| createProperties(child, count); |
| createNodes(child, count, depth - 1); |
| } |
| } |
| } |
| |
| private static void createProperties(NodeBuilder builder, int count) { |
| for (int k = 0; k < count; k++) { |
| builder.setProperty("property-" + UUID.randomUUID().toString(), "value-" + UUID.randomUUID().toString()); |
| } |
| } |
| |
| @Test |
| public void propertyRetention() throws Exception { |
| SegmentGCOptions gcOptions = defaultGCOptions(); |
| FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withMaxFileSize(1) |
| .withGCOptions(gcOptions) |
| .build(); |
| try { |
| final SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| |
| // Add a property |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setChildNode("test").setProperty("property", "value"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| |
| // Segment id of the current segment |
| NodeState test = nodeStore.getRoot().getChildNode("test"); |
| SegmentId id = ((SegmentNodeState) test).getRecordId().getSegmentId(); |
| fileStore.flush(); |
| assertTrue(fileStore.containsSegment(id)); |
| |
| // Add enough content to fill up the current tar file |
| builder = nodeStore.getRoot().builder(); |
| addContent(builder.setChildNode("dump")); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| |
| // Segment and property still there |
| assertTrue(fileStore.containsSegment(id)); |
| PropertyState property = test.getProperty("property"); |
| assertEquals("value", property.getValue(STRING)); |
| |
| // GC should remove the segment |
| fileStore.flush(); |
| // Ensure cleanup is efficient by surpassing the number of |
| // retained generations |
| for (int k = 0; k < gcOptions.getRetainedGenerations(); k++) { |
| fileStore.compactFull(); |
| } |
| fileStore.cleanup(); |
| |
| try { |
| fileStore.readSegment(id); |
| fail("Segment " + id + " should be gc'ed"); |
| } catch (SegmentNotFoundException ignore) {} |
| } finally { |
| fileStore.close(); |
| } |
| } |
| |
| @Test |
| public void checkpointDeduplicationTest() throws Exception { |
| class CP { |
| String id; |
| NodeState uncompacted; |
| NodeState compacted; |
| } |
| CP[] cps = {new CP(), new CP(), new CP(), new CP()}; |
| |
| try (FileStore fileStore = fileStoreBuilder(getFileStoreFolder()).build()) { |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| |
| // Initial content and checkpoint |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setChildNode("a").setChildNode("aa"); |
| builder.setChildNode("b").setChildNode("bb"); |
| builder.setChildNode("c").setChildNode("cc"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| cps[0].id = nodeStore.checkpoint(Long.MAX_VALUE); |
| |
| // Add content and another checkpoint |
| builder = nodeStore.getRoot().builder(); |
| builder.setChildNode("1").setChildNode("11"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| cps[1].id = nodeStore.checkpoint(Long.MAX_VALUE); |
| |
| // Modify content and another checkpoint |
| builder = nodeStore.getRoot().builder(); |
| builder.getChildNode("a").getChildNode("aa").setChildNode("aaa"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| cps[2].id = nodeStore.checkpoint(Long.MAX_VALUE); |
| |
| // Remove content and another checkpoint |
| builder = nodeStore.getRoot().builder(); |
| builder.getChildNode("a").remove(); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| cps[3].id = nodeStore.checkpoint(Long.MAX_VALUE); |
| |
| // A final bit of content |
| builder = nodeStore.getRoot().builder(); |
| builder.setChildNode("d").setChildNode("dd"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| |
| NodeState uncompactedSuperRoot = fileStore.getHead(); |
| NodeState uncompactedRoot = nodeStore.getRoot(); |
| for (CP cp : cps) { |
| cp.uncompacted = nodeStore.retrieve(cp.id); |
| } |
| |
| fileStore.compactFull(); |
| |
| NodeState compactedSuperRoot = fileStore.getHead(); |
| NodeState compactedRoot = nodeStore.getRoot(); |
| for (CP cp : cps) { |
| cp.compacted = nodeStore.retrieve(cp.id); |
| } |
| |
| assertEquals(uncompactedSuperRoot, compactedSuperRoot); |
| |
| assertEquals(uncompactedRoot, compactedRoot); |
| assertStableIds(uncompactedRoot, compactedRoot, "/root"); |
| |
| for (CP cp : cps) { |
| assertEquals(cp.uncompacted, cp.compacted); |
| assertStableIds(cp.uncompacted, cp.compacted, concat("/root/checkpoints", cp.id)); |
| } |
| } |
| } |
| |
| @Test |
| public void keepStableIdOnFlush() throws Exception { |
| try (FileStore fileStore = fileStoreBuilder(getFileStoreFolder()).build()) { |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| |
| // Initial content and checkpoint |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setChildNode("a"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| nodeStore.checkpoint(Long.MAX_VALUE); |
| |
| // A final bit of content |
| builder = nodeStore.getRoot().builder(); |
| for (int k = 0; k < UPDATE_LIMIT; k++) { |
| builder.setChildNode("b-" + k); |
| } |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| |
| NodeState uncompactedSuperRoot = fileStore.getHead(); |
| NodeState uncompactedRoot = nodeStore.getRoot(); |
| |
| fileStore.compactFull(); |
| |
| NodeState compactedSuperRoot = fileStore.getHead(); |
| NodeState compactedRoot = nodeStore.getRoot(); |
| |
| assertEquals(uncompactedSuperRoot, compactedSuperRoot); |
| |
| assertEquals(uncompactedRoot, compactedRoot); |
| assertStableIds(uncompactedRoot, compactedRoot, "/root"); |
| } |
| } |
| |
| @Test |
| public void crossGCDeduplicationTest() throws Exception { |
| try (FileStore fileStore = fileStoreBuilder(getFileStoreFolder()).build()) { |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setChildNode("a").setChildNode("aa"); |
| builder.setChildNode("b").setChildNode("bb"); |
| builder.setChildNode("c").setChildNode("cc"); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| |
| NodeState a = nodeStore.getRoot().getChildNode("a"); |
| |
| builder = nodeStore.getRoot().builder(); |
| builder.setChildNode("x").setChildNode("xx"); |
| |
| SegmentNodeState uncompacted = (SegmentNodeState) nodeStore.getRoot(); |
| fileStore.compactFull(); |
| NodeState compacted = nodeStore.getRoot(); |
| |
| assertEquals(uncompacted, compacted); |
| assertStableIds(uncompacted, compacted, "/root"); |
| |
| builder.setChildNode("y").setChildNode("yy"); |
| builder.getChildNode("a").remove(); |
| NodeState deferCompacted = nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| assertEquals( |
| uncompacted.getSegment().getGcGeneration().nextFull().nonGC(), |
| ((SegmentNodeState)deferCompacted).getSegment().getGcGeneration()); |
| } |
| } |
| |
| private static void assertStableIds(NodeState node1, NodeState node2, String path) { |
| assertFalse("Nodes should be equal: " + path, node1 == node2); |
| assertTrue("Node should be a SegmentNodeState " + path, node1 instanceof SegmentNodeState); |
| assertTrue("Node should be a SegmentNodeState " + path, node2 instanceof SegmentNodeState); |
| SegmentNodeState sns1 = (SegmentNodeState) node1; |
| SegmentNodeState sns2 = (SegmentNodeState) node2; |
| assertEquals("GC generation should be bumped by one " + path, |
| sns1.getSegment().getGcGeneration().nextFull(), sns2.getSegment().getGcGeneration()); |
| assertEquals("Nodes should have same stable id: " + path, |
| sns1.getStableId(), sns2.getStableId()); |
| |
| for (ChildNodeEntry cne : node1.getChildNodeEntries()) { |
| assertStableIds( |
| cne.getNodeState(), node2.getChildNode(cne.getName()), |
| concat(path, cne.getName())); |
| } |
| } |
| |
| /** |
| * Test asserting OAK-4669: No new generation of tar should be created when the segments are the same |
| * and when various indices are created. |
| */ |
| @Test |
| public void concurrentWritesCleanupNoNewGen() throws Exception { |
| ScheduledExecutorService scheduler = newSingleThreadScheduledExecutor(); |
| StatisticsProvider statsProvider = new DefaultStatisticsProvider(scheduler); |
| final FileStoreGCMonitor fileStoreGCMonitor = new FileStoreGCMonitor(Clock.SIMPLE); |
| |
| File fileStoreFolder = getFileStoreFolder(); |
| |
| final FileStore fileStore = fileStoreBuilder(fileStoreFolder) |
| .withGCOptions(defaultGCOptions().setRetainedGenerations(2)) |
| .withGCMonitor(fileStoreGCMonitor) |
| .withStatisticsProvider(statsProvider) |
| .withMaxFileSize(1) |
| .withMemoryMapping(false) |
| .build(); |
| |
| final SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| ExecutorService executorService = newFixedThreadPool(5); |
| final AtomicInteger counter = new AtomicInteger(); |
| |
| try { |
| Callable<Void> concurrentWriteTask = new Callable<Void>() { |
| @Override |
| public Void call() throws Exception { |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setProperty("blob-" + counter.getAndIncrement(), createBlob(nodeStore, 512 * 512)); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| return null; |
| } |
| }; |
| |
| List<Future<?>> results = newArrayList(); |
| for (int i = 0; i < 5; i++) { |
| results.add(executorService.submit(concurrentWriteTask)); |
| } |
| |
| for (Future<?> result : results) { |
| assertNull(result.get()); |
| } |
| |
| fileStore.cleanup(); |
| |
| for (String fileName : fileStoreFolder.list()) { |
| if (fileName.endsWith(".tar")) { |
| int pos = fileName.length() - "a.tar".length(); |
| char generation = fileName.charAt(pos); |
| assertTrue("Expected generation is 'a', but instead was: '" + generation + "' for file " + fileName, |
| generation == 'a'); |
| } |
| } |
| } finally { |
| new ExecutorCloser(executorService).close(); |
| fileStore.close(); |
| new ExecutorCloser(scheduler).close(); |
| } |
| } |
| |
| @Test |
| public void concurrentWritesCleanupZeroReclaimedSize() throws Exception { |
| ScheduledExecutorService scheduler = newSingleThreadScheduledExecutor(); |
| StatisticsProvider statsProvider = new DefaultStatisticsProvider(scheduler); |
| final FileStoreGCMonitor fileStoreGCMonitor = new FileStoreGCMonitor(Clock.SIMPLE); |
| |
| final FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withGCOptions(defaultGCOptions().setRetainedGenerations(2)) |
| .withGCMonitor(fileStoreGCMonitor) |
| .withStatisticsProvider(statsProvider) |
| .withMaxFileSize(1) |
| .withMemoryMapping(false) |
| .build(); |
| |
| final SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| ExecutorService executorService = newFixedThreadPool(100); |
| final AtomicInteger counter = new AtomicInteger(); |
| |
| try { |
| Callable<Void> concurrentWriteTask = new Callable<Void>() { |
| @Override |
| public Void call() throws Exception { |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setProperty("blob-" + counter.getAndIncrement(), createBlob(nodeStore, 25 * 25)); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| return null; |
| } |
| }; |
| |
| List<Future<?>> results = newArrayList(); |
| for (int i = 0; i < 100; i++) { |
| results.add(executorService.submit(concurrentWriteTask)); |
| } |
| |
| Thread.sleep(100); |
| fileStore.cleanup(); |
| |
| for (Future<?> result : results) { |
| assertNull(result.get()); |
| } |
| |
| long reclaimedSize = fileStoreGCMonitor.getLastReclaimedSize(); |
| assertEquals("Reclaimed size expected is 0, but instead was: " + reclaimedSize, |
| 0, reclaimedSize); |
| } finally { |
| new ExecutorCloser(executorService).close(); |
| fileStore.close(); |
| new ExecutorCloser(scheduler).close(); |
| } |
| } |
| |
| @Test |
| public void randomAccessFileConcurrentReadAndLength() throws Exception { |
| final FileStore fileStore = fileStoreBuilder(getFileStoreFolder()) |
| .withGCOptions(defaultGCOptions().setRetainedGenerations(2)) |
| .withMaxFileSize(1) |
| .withMemoryMapping(false) |
| .build(); |
| |
| final SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| ExecutorService executorService = newFixedThreadPool(300); |
| final AtomicInteger counter = new AtomicInteger(); |
| |
| try { |
| Callable<Void> concurrentWriteTask = new Callable<Void>() { |
| @Override |
| public Void call() throws Exception { |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setProperty("blob-" + counter.getAndIncrement(), createBlob(nodeStore, 25 * 25)); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| return null; |
| } |
| }; |
| |
| Callable<Void> concurrentCleanupTask = new Callable<Void>() { |
| @Override |
| public Void call() throws Exception { |
| fileStore.cleanup(); |
| return null; |
| } |
| }; |
| |
| Callable<Void> concurrentReferenceCollector = new Callable<Void>() { |
| @Override |
| public Void call() throws Exception { |
| fileStore.collectBlobReferences(s -> { |
| // Do nothing. |
| }); |
| return null; |
| } |
| }; |
| |
| List<Future<?>> results = newArrayList(); |
| for (int i = 0; i < 100; i++) { |
| results.add(executorService.submit(concurrentWriteTask)); |
| results.add(executorService.submit(concurrentCleanupTask)); |
| results.add(executorService.submit(concurrentReferenceCollector)); |
| } |
| |
| for (Future<?> result : results) { |
| assertNull(result.get()); |
| } |
| |
| } finally { |
| new ExecutorCloser(executorService).close(); |
| fileStore.close(); |
| } |
| } |
| |
| /** |
| * Test asserting OAK-4700: Concurrent cleanup must not remove segments that are still reachable |
| */ |
| @Test |
| public void concurrentCleanup() throws Exception { |
| File fileStoreFolder = getFileStoreFolder(); |
| |
| final FileStore fileStore = fileStoreBuilder(fileStoreFolder) |
| .withMaxFileSize(1) |
| .build(); |
| |
| final SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| ExecutorService executorService = newFixedThreadPool(50); |
| final AtomicInteger counter = new AtomicInteger(); |
| |
| try { |
| Callable<Void> concurrentWriteTask = new Callable<Void>() { |
| @Override |
| public Void call() throws Exception { |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setProperty("blob-" + counter.getAndIncrement(), createBlob(nodeStore, 512 * 512)); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| return null; |
| } |
| }; |
| |
| final Callable<Void> concurrentCleanTask = new Callable<Void>() { |
| @Override |
| public Void call() throws Exception { |
| fileStore.cleanup(); |
| return null; |
| } |
| }; |
| |
| List<Future<?>> results = newArrayList(); |
| for (int i = 0; i < 50; i++) { |
| if (i % 2 == 0) { |
| results.add(executorService.submit(concurrentWriteTask)); |
| } else { |
| results.add(executorService.submit(concurrentCleanTask)); |
| } |
| } |
| |
| for (Future<?> result : results) { |
| assertNull(result.get()); |
| } |
| |
| for (String fileName : fileStoreFolder.list()) { |
| if (fileName.endsWith(".tar")) { |
| int pos = fileName.length() - "a.tar".length(); |
| char generation = fileName.charAt(pos); |
| assertEquals("Expected nothing to be cleaned but generation '" + generation + |
| "' for file " + fileName + " indicates otherwise.", |
| "a", valueOf(generation)); |
| } |
| } |
| } finally { |
| new ExecutorCloser(executorService).close(); |
| fileStore.close(); |
| } |
| } |
| |
| private static void addContent(NodeBuilder builder) { |
| for (int k = 0; k < 10000; k++) { |
| builder.setProperty(UUID.randomUUID().toString(), UUID.randomUUID().toString()); |
| } |
| } |
| |
| private static BlobStore newBlobStore(File directory) { |
| OakFileDataStore delegate = new OakFileDataStore(); |
| delegate.setPath(directory.getAbsolutePath()); |
| delegate.init(null); |
| return new DataStoreBlobStore(delegate); |
| } |
| |
| |
| @Test |
| public void binaryRetentionWithDS() |
| throws IOException, InvalidFileStoreVersionException, CommitFailedException { |
| try (FileStore fileStore = fileStoreBuilder(new File(getFileStoreFolder(), "segmentstore")) |
| .withBlobStore(newBlobStore(new File(getFileStoreFolder(), "blobstore"))) |
| .withGCOptions(defaultGCOptions().setGcSizeDeltaEstimation(0)) |
| .build()) |
| { |
| SegmentNodeStore nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build(); |
| |
| NodeBuilder builder = nodeStore.getRoot().builder(); |
| builder.setProperty("bin", createBlob(nodeStore, 1000000)); |
| nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); |
| fileStore.flush(); |
| |
| Set<String> expectedReferences = newHashSet(); |
| fileStore.collectBlobReferences(expectedReferences::add); |
| |
| for(int k = 1; k <= 3; k++) { |
| fileStore.fullGC(); |
| Set<String> actualReferences = newHashSet(); |
| fileStore.collectBlobReferences(actualReferences::add); |
| assertEquals("Binary should be retained after " + k + "-th gc cycle", |
| expectedReferences, actualReferences); |
| } |
| } |
| } |
| |
| @Test |
| public void latestFullCompactedStateShouldNotBeDeleted() throws Exception { |
| SegmentGCOptions gcOptions = defaultGCOptions() |
| .setEstimationDisabled(true) |
| .setRetainedGenerations(2); |
| |
| try (FileStore fileStore = fileStoreBuilder(getFileStoreFolder()).withGCOptions(gcOptions).build()) { |
| SegmentNodeState previousHead; |
| SegmentNodeState head = fileStore.getHead(); |
| |
| // Create a full, self consistent head state. This state will be the |
| // base for the following tail compactions. This increments the full generation. |
| fileStore.fullGC(); |
| previousHead = head; |
| head = fileStore.getHead(); |
| |
| // retainedGeneration = 2 -> the full compacted head and the previous uncompacted head must |
| // still be available. |
| traverse(previousHead); |
| traverse(head); |
| |
| // Create a tail head state on top of the previous full state. This |
| // increments the generation, but leaves the full generation untouched. |
| fileStore.tailGC(); |
| previousHead = head; |
| head = fileStore.getHead(); |
| |
| // retainedGeneration = 2 -> the tail compacted head and the previous uncompacted head must |
| // still be available. |
| traverse(previousHead); |
| traverse(head); |
| |
| // Create a tail state on top of the previous tail state. This increments the generation, |
| // but leaves the full generation untouched. This brings this generations two generations |
| // away from the latest full head state. Still, the full head state will not be deleted |
| // because doing so would generate an invalid repository at risk of SegmentNotFoundException. |
| fileStore.tailGC(); |
| previousHead = head; |
| head = fileStore.getHead(); |
| |
| // retainedGeneration = 2 -> the tail compacted head and the previous uncompacted head must |
| // still be available. |
| traverse(previousHead); |
| traverse(head); |
| |
| // Create a full, self consistent head state replacing the current tail of tail |
| // compacted heads. |
| fileStore.fullGC(); |
| previousHead = head; |
| head = fileStore.getHead(); |
| |
| // retainedGeneration = 2 -> the full compacted head and the previous uncompacted head must |
| // still be available. |
| traverse(previousHead); |
| traverse(head); |
| } |
| } |
| |
| } |
