diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreService.java oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreService.java index 8e95ee1..b30173b 100644 --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreService.java +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreService.java @@ -574,7 +574,11 @@ public class DocumentNodeStoreService { RevisionGC revisionGC = new RevisionGC(new Runnable() { @Override public void run() { - store.getVersionGarbageCollector().gc(versionGcMaxAgeInSecs, TimeUnit.SECONDS); + try { + store.getVersionGarbageCollector().gc(versionGcMaxAgeInSecs, TimeUnit.SECONDS); + } catch (IOException e) { + log.warn("Error occurred while executing the Version Garbage Collector", e); + } } }, executor); registrations.add(registerMBean(whiteboard, RevisionGCMBean.class, revisionGC, diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/VersionGarbageCollector.java oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/VersionGarbageCollector.java index 2b1b7f0..70d71af 100644 --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/VersionGarbageCollector.java +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/VersionGarbageCollector.java @@ -19,30 +19,49 @@ package org.apache.jackrabbit.oak.plugins.document; -import java.util.ArrayList; +import java.io.Closeable; +import java.io.File; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.PrintWriter; +import java.nio.charset.Charset; import java.util.Collections; +import java.util.Comparator; import java.util.EnumSet; +import java.util.Iterator; import java.util.List; import java.util.Set; import java.util.concurrent.TimeUnit; + +import com.google.common.base.Charsets; import com.google.common.base.Joiner; import com.google.common.base.StandardSystemProperty; import com.google.common.base.Stopwatch; import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import com.google.common.io.Closer; +import com.google.common.io.Files; +import org.apache.commons.io.FileUtils; +import org.apache.commons.io.LineIterator; +import org.apache.jackrabbit.oak.commons.sort.ExternalSort; import org.apache.jackrabbit.oak.plugins.document.util.Utils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static com.google.common.collect.Iterators.partition; import static org.apache.jackrabbit.oak.plugins.document.NodeDocument.SplitDocType.COMMIT_ROOT_ONLY; import static org.apache.jackrabbit.oak.plugins.document.NodeDocument.SplitDocType.DEFAULT_LEAF; public class VersionGarbageCollector { + //Kept less than MongoDocumentStore.IN_CLAUSE_BATCH_SIZE to avoid re-partitioning + private static final int DELETE_BATCH_SIZE = 450; private final DocumentNodeStore nodeStore; private final VersionGCSupport versionStore; + private int overflowToDiskThreshold = 100000; - private final Logger log = LoggerFactory.getLogger(getClass()); + private static final Logger log = LoggerFactory.getLogger(VersionGarbageCollector.class); /** * Split document types which can be safely garbage collected @@ -56,7 +75,7 @@ public class VersionGarbageCollector { this.versionStore = gcSupport; } - public VersionGCStats gc(long maxRevisionAge, TimeUnit unit) { + public VersionGCStats gc(long maxRevisionAge, TimeUnit unit) throws IOException { long maxRevisionAgeInMillis = unit.toMillis(maxRevisionAge); Stopwatch sw = Stopwatch.createStarted(); VersionGCStats stats = new VersionGCStats(); @@ -85,41 +104,60 @@ public class VersionGarbageCollector { return stats; } + public void setOverflowToDiskThreshold(int overflowToDiskThreshold) { + this.overflowToDiskThreshold = overflowToDiskThreshold; + } + private void collectSplitDocuments(VersionGCStats stats, long oldestRevTimeStamp) { versionStore.deleteSplitDocuments(GC_TYPES, oldestRevTimeStamp, stats); } - private void collectDeletedDocuments(VersionGCStats stats, Revision headRevision, long oldestRevTimeStamp) { - List docIdsToDelete = new ArrayList(); - Iterable itr = versionStore.getPossiblyDeletedDocs(oldestRevTimeStamp); + private void collectDeletedDocuments(VersionGCStats stats, Revision headRevision, long oldestRevTimeStamp) + throws IOException { + NodeDocIdCollector docIdsToDelete = new NodeDocIdCollector(overflowToDiskThreshold, NodeDocumentIdComparator.INSTANCE); try { - for (NodeDocument doc : itr) { - //Check if node is actually deleted at current revision - //As node is not modified since oldestRevTimeStamp then - //this node has not be revived again in past maxRevisionAge - //So deleting it is safe - if (doc.getNodeAtRevision(nodeStore, headRevision, null) == null) { - docIdsToDelete.add(doc.getId()); - //Collect id of all previous docs also - for (NodeDocument prevDoc : ImmutableList.copyOf(doc.getAllPreviousDocs())) { - docIdsToDelete.add(prevDoc.getId()); + Iterable itr = versionStore.getPossiblyDeletedDocs(oldestRevTimeStamp); + try { + for (NodeDocument doc : itr) { + //Check if node is actually deleted at current revision + //As node is not modified since oldestRevTimeStamp then + //this node has not be revived again in past maxRevisionAge + //So deleting it is safe + if (doc.getNodeAtRevision(nodeStore, headRevision, null) == null) { + docIdsToDelete.add(doc.getId()); + //Collect id of all previous docs also + for (NodeDocument prevDoc : ImmutableList.copyOf(doc.getAllPreviousDocs())) { + docIdsToDelete.add(prevDoc.getId()); + } } } + } finally { + Utils.closeIfCloseable(itr); + } + + if (docIdsToDelete.isEmpty()){ + return; + } + + docIdsToDelete.sort(); + log.info("Proceeding to delete [{}] documents", docIdsToDelete.getSize()); + + if (log.isDebugEnabled() && docIdsToDelete.size < 1000) { + StringBuilder sb = new StringBuilder("Deleted document with following ids were deleted as part of GC \n"); + Joiner.on(StandardSystemProperty.LINE_SEPARATOR.value()).appendTo(sb, docIdsToDelete.getIds()); + log.debug(sb.toString()); } - } finally { - Utils.closeIfCloseable(itr); - } - Collections.sort(docIdsToDelete, NodeDocumentIdComparator.INSTANCE); + Iterator> idListItr = partition(docIdsToDelete.getIds(), DELETE_BATCH_SIZE); + while (idListItr.hasNext()) { + nodeStore.getDocumentStore().remove(Collection.NODES, idListItr.next()); + } - if(log.isDebugEnabled()) { - StringBuilder sb = new StringBuilder("Deleted document with following ids were deleted as part of GC \n"); - Joiner.on(StandardSystemProperty.LINE_SEPARATOR.value()).appendTo(sb, docIdsToDelete); - log.debug(sb.toString()); + nodeStore.invalidateDocChildrenCache(); + stats.deletedDocGCCount += docIdsToDelete.getSize(); + } finally { + docIdsToDelete.close(); } - nodeStore.getDocumentStore().remove(Collection.NODES, docIdsToDelete); - nodeStore.invalidateDocChildrenCache(); - stats.deletedDocGCCount += docIdsToDelete.size(); } public static class VersionGCStats { @@ -139,4 +177,198 @@ public class VersionGarbageCollector { '}'; } } + + static class NodeDocIdCollector implements Closeable{ + public static final int BATCH_SIZE = 2048; + private final int overflowToDiskThreshold; + private final Comparator comparator; + private final List ids = Lists.newArrayList(); + private boolean useFile; + private final List inMemBatch = Lists.newArrayList(); + private VersionGCState gcState; + private int size; + + NodeDocIdCollector(int overflowToDiskThreshold, Comparator comparator) { + this.overflowToDiskThreshold = overflowToDiskThreshold; + this.comparator = comparator; + } + + public void add(String id) throws IOException{ + if (useFile){ + addToBatch(id); + } else { + ids.add(id); + if (ids.size() >= overflowToDiskThreshold){ + flushToFile(ids); + useFile = true; + log.debug("Number of documents to be deleted have exceeded the in memory threshold of {}. " + + "Switching to filesystem [{}] to manage the state", overflowToDiskThreshold, gcState); + } + } + size++; + } + + public void sort() throws IOException { + if (useFile){ + //Flush the last batch + flushToFile(inMemBatch); + gcState.sort(); + } else { + Collections.sort(ids, PathComparator.INSTANCE); + } + } + + public Iterator getIds() throws IOException { + if (useFile){ + return gcState.getLineIterator(); + } else { + return ids.iterator(); + } + } + + public int getSize() { + return size; + } + + public boolean isEmpty(){ + return size == 0; + } + + public boolean usingFile(){ + return useFile; + } + + @Override + public void close() throws IOException { + if (gcState != null){ + gcState.close(); + } + } + + private void addToBatch(String id) throws IOException { + inMemBatch.add(id); + if(inMemBatch.size() >= BATCH_SIZE){ + flushToFile(inMemBatch); + } + } + + private void flushToFile(List ids) throws IOException { + PrintWriter w = getPersistentState().getWriter(); + for (String id : ids){ + w.println(id); + } + ids.clear(); + } + + private VersionGCState getPersistentState(){ + if (gcState == null){ + gcState = new VersionGCState(comparator); + } + return gcState; + } + } + + static class VersionGCState implements Closeable{ + //Defaults taken from ExternalSort + static final int DEFAULTMAXTEMPFILES = 1024; + /** + * Defines the default maximum memory to be used while sorting (8 MB) + */ + static final long DEFAULT_MAX_MEM_BYTES = 8388608L; + /** Maximum loop count when creating temp directories. */ + private static final int TEMP_DIR_ATTEMPTS = 10000; + + private final Charset charset = Charsets.UTF_8; + private final File workDir; + private final Comparator comparator; + private File idFile; + private File sortedFile; + private PrintWriter writer; + private LineIterator lineIterator; + + public VersionGCState(Comparator comparator) { + this.workDir = createTempDir("oak-versiongc-"); + this.comparator = comparator; + } + + public PrintWriter getWriter() throws FileNotFoundException { + if (idFile == null){ + idFile = new File(workDir, "deletedDocIds.txt"); + sortedFile = new File(workDir, "deletedDocIds-sorted.txt"); + writer = new PrintWriter(Files.newWriter(idFile, charset)); + } + return writer; + } + + public void sort() throws IOException { + closeWriter(); + + List sortedFiles = ExternalSort.sortInBatch(idFile, + comparator, //Comparator to use + DEFAULTMAXTEMPFILES, + DEFAULT_MAX_MEM_BYTES, + charset, //charset + workDir, //temp directory where intermediate files are created + true //distinct + ); + + ExternalSort.mergeSortedFiles(sortedFiles, + sortedFile, + comparator, + charset, + true + ); + } + + public Iterator getLineIterator() throws IOException { + lineIterator = new LineIterator(Files.newReader(sortedFile, charset)); + return lineIterator; + } + + @Override + public String toString() { + return "VersionGCState : workDir=" + workDir.getAbsolutePath(); + } + + @Override + public void close() throws IOException { + Closer closer = Closer.create(); + try{ + closer.register(writer); + closer.register(new Closeable() { + @Override + public void close() throws IOException { + lineIterator.close(); + } + }); + FileUtils.deleteDirectory(workDir); + } finally { + closer.close(); + } + } + + private void closeWriter() { + writer.close(); + } + + /** + * Taken from com.google.common.io.Files#createTempDir() + * Modified to provide a prefix + * + */ + private static File createTempDir(String prefix) { + File baseDir = new File(System.getProperty("java.io.tmpdir")); + String baseName = System.currentTimeMillis() + "-"; + + for (int counter = 0; counter < TEMP_DIR_ATTEMPTS; counter++) { + File tempDir = new File(baseDir, prefix + baseName + counter); + if (tempDir.mkdir()) { + return tempDir; + } + } + throw new IllegalStateException("Failed to create directory within " + + TEMP_DIR_ATTEMPTS + " attempts (tried " + + baseName + "0 to " + baseName + (TEMP_DIR_ATTEMPTS - 1) + ')'); + } + } } diff --git oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/NodeDocIdCollectorTest.java oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/NodeDocIdCollectorTest.java new file mode 100644 index 0000000..f0e441a --- /dev/null +++ oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/NodeDocIdCollectorTest.java @@ -0,0 +1,119 @@ +/* + * 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.plugins.document; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import com.google.common.base.Joiner; +import com.google.common.collect.Collections2; +import com.google.common.collect.ImmutableList; +import org.apache.jackrabbit.oak.plugins.document.VersionGarbageCollector.NodeDocIdCollector; +import org.apache.jackrabbit.oak.plugins.document.util.Utils; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class NodeDocIdCollectorTest { + private NodeDocIdCollector collector; + + @Test + public void inMemory() throws Exception{ + List paths = createTestPaths(5, false, false); + collector = new NodeDocIdCollector(paths.size() + 1, NodeDocumentIdComparator.INSTANCE); + addPathsToCollector(paths); + + assertConstraints(paths); + assertFalse(collector.usingFile()); + collector.close(); + } + + @Test + public void overflowToDisk() throws Exception{ + //Create ~50k paths + List paths = createTestPaths(10, true, false); + collector = new NodeDocIdCollector(1000, NodeDocumentIdComparator.INSTANCE); + addPathsToCollector(paths); + + assertTrue(collector.usingFile()); + assertConstraints(paths); + + collector.close(); + } + + + private void assertConstraints(List paths) throws IOException { + assertEquals(paths.size(), collector.getSize()); + + Collections.sort(paths, PathComparator.INSTANCE); + collector.sort(); + + List sortedPaths = ImmutableList.copyOf(collector.getIds()); + assertEquals(paths.size(), sortedPaths.size()); + assertEquals(paths, sortedPaths); + } + + private void addPathsToCollector(Iterable paths) throws IOException { + for (String path : paths){ + collector.add(path); + } + } + + private static List createTestPaths(int depth, boolean permutation, boolean pathMode){ + List rootPaths = Arrays.asList("a", "b", "c", "d", "e", "f", "g"); + List paths = new ArrayList(); + + + if (permutation){ + List newRoots = new ArrayList(); + for (List permuts : Collections2.orderedPermutations(rootPaths)){ + newRoots.add(Joiner.on("").join(permuts)); + } + rootPaths = newRoots; + } + + for (String root : rootPaths){ + List pathElements = new ArrayList(); + pathElements.add(root); + paths.add(createId(pathElements)); + for (int i = 0; i < depth; i++){ + pathElements.add(root + i); + paths.add(createId(pathElements)); + } + } + + Set idSet = new HashSet(paths); + assertEquals(paths.size(), idSet.size()); + + Collections.shuffle(paths); + return paths; + } + + private static String createId(Iterable pathElements){ + return Utils.getIdFromPath("/" + Joiner.on('/').join(pathElements)); + } +} diff --git oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/VersionGCWithSplitTest.java oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/VersionGCWithSplitTest.java index e586961..8642f60 100644 --- oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/VersionGCWithSplitTest.java +++ oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/VersionGCWithSplitTest.java @@ -137,7 +137,11 @@ public class VersionGCWithSplitTest { Thread t = new Thread(new Runnable() { @Override public void run() { - stats.set(gc.gc(1, HOURS)); + try { + stats.set(gc.gc(1, HOURS)); + } catch (IOException e) { + throw new RuntimeException(e); + } } });