Author: mduerig
Date: Mon Dec 11 13:56:50 2017
New Revision: 1817776
URL: http://svn.apache.org/viewvc?rev=1817776&view=rev
Log:
OAK-6984: High read IO in compaction retry cycles
Rebase diffs onto the previously compacted state in the compaction retry cycles
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CheckpointCompactor.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorTest.java
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CheckpointCompactor.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CheckpointCompactor.java?rev=1817776&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CheckpointCompactor.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CheckpointCompactor.java
Mon Dec 11 13:56:50 2017
@@ -0,0 +1,264 @@
+/*
+ * 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 com.google.common.collect.Lists.newArrayList;
+import static com.google.common.collect.Maps.newHashMap;
+import static com.google.common.collect.Maps.newLinkedHashMap;
+import static org.apache.jackrabbit.oak.commons.PathUtils.elements;
+import static org.apache.jackrabbit.oak.commons.PathUtils.getName;
+import static org.apache.jackrabbit.oak.commons.PathUtils.getParentPath;
+import static
org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.CheckForNull;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import com.google.common.base.Supplier;
+import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+import org.apache.jackrabbit.oak.spi.state.ChildNodeEntry;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+
+/**
+ * This compactor implementation is aware of the checkpoints in the repository.
+ * It uses this information to further optimise the compaction result by
+ * <ul>
+ * <li>Rebasing the checkpoints and subsequently the root on top of each
other
+ * in chronological order. This results minimises the deltas that need to
be
+ * processed and stored.</li>
+ * <li>Caching the compacted checkpoints and root states for deduplication
should
+ * the same checkpoint or root state occur again in a later compaction
retry cycle.</li>
+ * </ul>
+ */
+public class CheckpointCompactor {
+ @Nonnull
+ private final GCMonitor gcListener;
+
+ @Nonnull
+ private final AtomicLong gcCount;
+
+ @Nonnull
+ private final Map<NodeState, NodeState> cpCache = newHashMap();
+
+ @Nonnull
+ private final Compactor compactor;
+
+ @Nonnull
+ private final NodeWriter nodeWriter;
+
+ private interface NodeWriter {
+ @Nonnull
+ SegmentNodeState writeNode(@Nonnull NodeState node, @Nullable
ByteBuffer stableId) throws IOException;
+ }
+
+ /**
+ * Create a new instance based on the passed arguments.
+ * @param reader segment reader used to read from the segments
+ * @param writer segment writer used to serialise to segments
+ * @param blobStore the blob store or {@code null} if none
+ * @param cancel a flag that can be used to cancel the compaction
process
+ * @param compactionMonitor notification call back for each compacted
nodes,
+ * properties, and binaries
+ */
+ public CheckpointCompactor(
+ @Nonnull GCMonitor gcListener,
+ @Nonnull AtomicLong gcCount,
+ @Nonnull SegmentReader reader,
+ @Nonnull SegmentWriter writer,
+ @Nullable BlobStore blobStore,
+ @Nonnull Supplier<Boolean> cancel,
+ @Nonnull GCNodeWriteMonitor compactionMonitor) {
+ this.gcListener = gcListener;
+ this.gcCount = gcCount;
+ this.compactor = new Compactor(reader, writer, blobStore, cancel,
compactionMonitor);
+ this.nodeWriter = (node, stableId) -> {
+ RecordId nodeId = writer.writeNode(node, stableId);
+ return new SegmentNodeState(reader, writer, blobStore, nodeId);
+ };
+ }
+
+ /**
+ * Compact {@code uncompacted} on top of an optional {@code base}.
+ * @param base the base state to compact against
+ * @param uncompacted the uncompacted state to compact
+ * @param onto the state onto which to compact the change between
{@code base} and
+ * {@code uncompacted}
+ * @return compacted clone of {@code uncompacted} or {@code null} if
cancelled.
+ * @throws IOException
+ */
+ @CheckForNull
+ public SegmentNodeState compact(
+ @Nonnull NodeState base,
+ @Nonnull NodeState uncompacted,
+ @Nonnull NodeState onto)
+ throws IOException {
+ // Collect a chronologically ordered list of roots for the uncompacted
+ // state. This list consists of all checkpoints followed by the root.
+ LinkedHashMap<String, NodeState> uncompactedRoots =
collectRoots(uncompacted);
+
+ // Compact the list of uncompacted roots to a list of compacted roots.
+ LinkedHashMap<String, NodeState> compactedRoots = compact(
+ getRoot(base), uncompactedRoots, getRoot(onto));
+ if (compactedRoots == null) {
+ return null;
+ }
+
+ // Build a compacted super root by replacing the uncompacted roots with
+ // the compacted ones in the original node.
+ NodeBuilder builder = uncompacted.builder();
+ for (Entry<String, NodeState> compactedRoot :
compactedRoots.entrySet()) {
+ String path = compactedRoot.getKey();
+ NodeState state = compactedRoot.getValue();
+ NodeBuilder childBuilder = getChild(builder, getParentPath(path));
+ childBuilder.setChildNode(getName(path), state);
+ }
+
+ return nodeWriter.writeNode(builder.getNodeState(),
getStableIdBytes(uncompacted));
+ }
+
+ @CheckForNull
+ private static ByteBuffer getStableIdBytes(@Nonnull NodeState node) {
+ return node instanceof SegmentNodeState
+ ? ((SegmentNodeState) node).getStableIdBytes()
+ : null;
+ }
+
+ @Nonnull
+ private static NodeState getRoot(@Nonnull NodeState node) {
+ return node.hasChildNode("root")
+ ? node.getChildNode("root")
+ : EMPTY_NODE;
+ }
+
+ /**
+ * Compact a list of uncompacted roots on top of base roots of the same
key or
+ * an empty node if none.
+ */
+ @CheckForNull
+ private LinkedHashMap<String, NodeState> compact(
+ @Nonnull NodeState base,
+ @Nonnull LinkedHashMap<String, NodeState> uncompactedRoots,
+ @Nonnull NodeState onto)
+ throws IOException {
+ LinkedHashMap<String, NodeState> compactedRoots = newLinkedHashMap();
+ for (Entry<String, NodeState> uncompactedRoot :
uncompactedRoots.entrySet()) {
+ String path = uncompactedRoot.getKey();
+ NodeState uncompacted = uncompactedRoot.getValue();
+ Result result = compactWithCache(base, uncompacted, onto, path);
+ if (result == null) {
+ return null;
+ }
+ base = result.nextBefore;
+ onto = result.nextOnto;
+ compactedRoots.put(path, result.compacted);
+ }
+ return compactedRoots;
+ }
+
+ /**
+ * Collect a chronologically ordered list of roots for the base and the
uncompacted
+ * state from a {@code superRoot}. This list consists of all checkpoints
followed by
+ * the root.
+ */
+ @Nonnull
+ private LinkedHashMap<String, NodeState> collectRoots(@Nullable NodeState
superRoot) {
+ LinkedHashMap<String, NodeState> roots = newLinkedHashMap();
+ if (superRoot != null) {
+ List<ChildNodeEntry> checkpoints = newArrayList(
+
superRoot.getChildNode("checkpoints").getChildNodeEntries());
+
+ checkpoints.sort((cne1, cne2) -> {
+ long c1 = cne1.getNodeState().getLong("created");
+ long c2 = cne2.getNodeState().getLong("created");
+ return Long.compare(c1, c2);
+ });
+
+ for (ChildNodeEntry checkpoint : checkpoints) {
+ String name = checkpoint.getName();
+ NodeState node = checkpoint.getNodeState();
+ gcListener.info("TarMK GC #{}: Found checkpoint {} created at
{}.",
+ gcCount, name, new Date(node.getLong("created")));
+ roots.put("checkpoints/" + name + "/root",
node.getChildNode("root"));
+ }
+ roots.put("root", superRoot.getChildNode("root"));
+ }
+ return roots;
+ }
+
+ @Nonnull
+ private static NodeBuilder getChild(NodeBuilder builder, String path) {
+ for (String name : elements(path)) {
+ builder = builder.getChildNode(name);
+ }
+ return builder;
+ }
+
+ private static class Result {
+ final NodeState compacted;
+ final NodeState nextBefore;
+ final NodeState nextOnto;
+
+ Result(@Nonnull NodeState compacted, @Nonnull NodeState
nextBefore, @Nonnull NodeState nextOnto) {
+ this.compacted = compacted;
+ this.nextBefore = nextBefore;
+ this.nextOnto = nextOnto;
+ }
+ }
+
+ /**
+ * Compact {@code after} against {@code before} on top of {@code onto}
unless
+ * {@code after} has been compacted before and is found in the cache. In
this
+ * case the cached version of the previously compacted {@code before} is
returned.
+ */
+ @CheckForNull
+ private Result compactWithCache(
+ @Nonnull NodeState before,
+ @Nonnull NodeState after,
+ @Nonnull NodeState onto,
+ @Nonnull String path)
+ throws IOException {
+ gcListener.info("TarMK GC #{}: compacting {}.", gcCount, path);
+ NodeState compacted = cpCache.get(after);
+ if (compacted == null) {
+ compacted = compactor.compact(before, after, onto);
+ if (compacted == null) {
+ return null;
+ } else {
+ cpCache.put(after, compacted);
+ return new Result(compacted, after, compacted);
+ }
+ } else {
+ gcListener.info("TarMK GC #{}: Found {} in cache.", gcCount, path);
+ return new Result(compacted, before, onto);
+ }
+ }
+
+}
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java?rev=1817776&r1=1817775&r2=1817776&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
Mon Dec 11 13:56:50 2017
@@ -18,8 +18,6 @@
*/
package org.apache.jackrabbit.oak.segment.file;
-import static com.google.common.collect.Lists.newArrayList;
-import static com.google.common.collect.Maps.newLinkedHashMap;
import static com.google.common.collect.Sets.newHashSet;
import static java.lang.Integer.getInteger;
import static java.lang.String.format;
@@ -29,9 +27,6 @@ import static java.util.concurrent.TimeU
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCount;
-import static org.apache.jackrabbit.oak.commons.PathUtils.elements;
-import static org.apache.jackrabbit.oak.commons.PathUtils.getName;
-import static org.apache.jackrabbit.oak.commons.PathUtils.getParentPath;
import static
org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
import static
org.apache.jackrabbit.oak.segment.DefaultSegmentWriterBuilder.defaultSegmentWriterBuilder;
import static org.apache.jackrabbit.oak.segment.SegmentId.isDataSegmentId;
@@ -51,9 +46,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.FileLock;
import java.nio.channels.OverlappingFileLockException;
import java.util.Collection;
-import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
@@ -72,12 +65,11 @@ import com.google.common.base.Predicate;
import com.google.common.base.Stopwatch;
import com.google.common.base.Supplier;
import com.google.common.io.Closer;
+import org.apache.jackrabbit.oak.segment.CheckpointCompactor;
import com.google.common.util.concurrent.UncheckedExecutionException;
-import org.apache.jackrabbit.oak.segment.Compactor;
import org.apache.jackrabbit.oak.segment.RecordId;
import org.apache.jackrabbit.oak.segment.Segment;
import org.apache.jackrabbit.oak.segment.SegmentId;
-import org.apache.jackrabbit.oak.segment.SegmentNodeBuilder;
import org.apache.jackrabbit.oak.segment.SegmentNodeState;
import org.apache.jackrabbit.oak.segment.SegmentNotFoundException;
import org.apache.jackrabbit.oak.segment.SegmentNotFoundExceptionListener;
@@ -90,7 +82,6 @@ import org.apache.jackrabbit.oak.segment
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.segment.file.tar.TarFiles;
import org.apache.jackrabbit.oak.segment.file.tar.TarFiles.CleanupResult;
-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.slf4j.Logger;
@@ -721,7 +712,7 @@ public class FileStore extends AbstractF
synchronized CompactionResult compactFull() {
gcListener.info("TarMK GC #{}: running full compaction", GC_COUNT);
- return compact(null, getGcGeneration().nextFull());
+ return compact(EMPTY_NODE, getGcGeneration().nextFull());
}
synchronized CompactionResult compactTail() {
@@ -731,10 +722,12 @@ public class FileStore extends AbstractF
return compact(base, getGcGeneration().nextTail());
}
gcListener.info("TarMK GC #{}: no base state available, running
full compaction instead", GC_COUNT);
- return compact(null, getGcGeneration().nextFull());
+ return compact(EMPTY_NODE, getGcGeneration().nextFull());
}
- private CompactionResult compact(SegmentNodeState base, GCGeneration
newGeneration) {
+ private CompactionResult compact(
+ @Nonnull NodeState base,
+ @Nonnull GCGeneration newGeneration) {
try {
Stopwatch watch = Stopwatch.createStarted();
gcListener.info("TarMK GC #{}: compaction started, gc
options={}", GC_COUNT, gcOptions);
@@ -742,32 +735,36 @@ public class FileStore extends AbstractF
GCJournalEntry gcEntry = gcJournal.read();
long initialSize = size();
- compactionMonitor = new
GCNodeWriteMonitor(gcOptions.getGcLogInterval(), gcListener);
- compactionMonitor.init(GC_COUNT.get(), gcEntry.getRepoSize(),
gcEntry.getNodes(), initialSize);
- SegmentNodeState before = getHead();
- CancelCompactionSupplier cancel = new
CancelCompactionSupplier(FileStore.this);
SegmentWriter writer = defaultSegmentWriterBuilder("c")
.with(cacheManager)
.withGeneration(newGeneration)
.withoutWriterPool()
.build(FileStore.this);
- Compactor compactor = new Compactor(
+
+ CancelCompactionSupplier cancel = new
CancelCompactionSupplier(FileStore.this);
+
+ compactionMonitor = new
GCNodeWriteMonitor(gcOptions.getGcLogInterval(), gcListener);
+ compactionMonitor.init(GC_COUNT.get(), gcEntry.getRepoSize(),
gcEntry.getNodes(), initialSize);
+
+ CheckpointCompactor compactor = new
CheckpointCompactor(gcListener, GC_COUNT,
segmentReader, writer, getBlobStore(), cancel,
compactionMonitor);
- SegmentNodeState after = compact(base, before, compactor,
writer);
- if (after == null) {
+ SegmentNodeState head = getHead();
+ SegmentNodeState compacted = compactor.compact(base, head,
base);
+ if (compacted == null) {
gcListener.warn("TarMK GC #{}: compaction cancelled: {}.",
GC_COUNT, cancel);
return compactionAborted(newGeneration);
}
gcListener.info("TarMK GC #{}: compaction cycle 0 completed in
{} ({} ms). Compacted {} to {}",
- GC_COUNT, watch, watch.elapsed(MILLISECONDS),
before.getRecordId(), after.getRecordId());
+ GC_COUNT, watch, watch.elapsed(MILLISECONDS),
head.getRecordId(), compacted.getRecordId());
int cycles = 0;
boolean success = false;
+ SegmentNodeState previousHead = head;
while (cycles < gcOptions.getRetryCount() &&
- !(success = revisions.setHead(before.getRecordId(),
after.getRecordId(), EXPEDITE_OPTION))) {
+ !(success =
revisions.setHead(previousHead.getRecordId(), compacted.getRecordId(),
EXPEDITE_OPTION))) {
// Some other concurrent changes have been made.
// Rebase (and compact) those changes on top of the
// compacted state before retrying to set the head.
@@ -777,18 +774,18 @@ public class FileStore extends AbstractF
GC_COUNT, cycles, gcOptions.getRetryCount());
gcListener.updateStatus(COMPACTION_RETRY.message() +
cycles);
Stopwatch cycleWatch = Stopwatch.createStarted();
-
- SegmentNodeState head = getHead();
- after = compact(after, head, compactor, writer);
- if (after == null) {
+
+ head = getHead();
+ compacted = compactor.compact(previousHead, head,
compacted);
+ if (compacted == null) {
gcListener.warn("TarMK GC #{}: compaction cancelled:
{}.", GC_COUNT, cancel);
return compactionAborted(newGeneration);
}
gcListener.info("TarMK GC #{}: compaction cycle {}
completed in {} ({} ms). Compacted {} against {} to {}",
GC_COUNT, cycles, cycleWatch,
cycleWatch.elapsed(MILLISECONDS),
- head.getRecordId(), before.getRecordId(),
after.getRecordId());
- before = head;
+ head.getRecordId(), previousHead.getRecordId(),
compacted.getRecordId());
+ previousHead = head;
}
if (!success) {
@@ -804,8 +801,8 @@ public class FileStore extends AbstractF
cycles++;
cancel.timeOutAfter(forceTimeout, SECONDS);
- after = forceCompact(after, compactor, writer);
- success = after != null;
+ compacted = forceCompact(previousHead, compacted,
compactor);
+ success = compacted != null;
if (success) {
gcListener.info("TarMK GC #{}: compaction
succeeded to force compact remaining commits " +
"after {} ({} ms).",
@@ -829,7 +826,7 @@ public class FileStore extends AbstractF
flush();
gcListener.info("TarMK GC #{}: compaction succeeded in {}
({} ms), after {} cycles",
GC_COUNT, watch, watch.elapsed(MILLISECONDS),
cycles);
- return compactionSucceeded(newGeneration,
after.getRecordId());
+ return compactionSucceeded(newGeneration,
compacted.getRecordId());
} else {
gcListener.info("TarMK GC #{}: compaction failed after {}
({} ms), and {} cycles",
GC_COUNT, watch, watch.elapsed(MILLISECONDS),
cycles);
@@ -845,121 +842,10 @@ public class FileStore extends AbstractF
}
}
- /**
- * Compact {@code uncompacted} on top of an optional {@code base}.
- * @param base the base state to compact onto or {@code null}
for an empty state.
- * @param uncompacted the uncompacted state to compact
- * @param compactor the compactor for creating the new generation
of the
- * uncompacted state.
- * @param writer the segment writer used by {@code compactor}
for writing to the
- * new generation.
- * @return compacted clone of {@code uncompacted} or null if
cancelled.
- * @throws IOException
- */
- @CheckForNull
- private SegmentNodeState compact(
- @Nullable SegmentNodeState base,
- @Nonnull SegmentNodeState uncompacted,
- @Nonnull Compactor compactor,
- @Nonnull SegmentWriter writer)
- throws IOException {
- // Collect a chronologically ordered list of roots for the base
and the uncompacted
- // state. This list consists of all checkpoints followed by the
root.
- LinkedHashMap<String, NodeState> baseRoots = collectRoots(base);
- LinkedHashMap<String, NodeState> uncompactedRoots =
collectRoots(uncompacted);
-
- // Compact the list of uncompacted roots to a list of compacted
roots.
- LinkedHashMap<String, NodeState> compactedRoots =
compact(baseRoots, uncompactedRoots, compactor);
- if (compactedRoots == null) {
- return null;
- }
-
- // Build a compacted super root by replacing the uncompacted roots
with
- // the compacted ones in the original node.
- SegmentNodeBuilder builder = uncompacted.builder();
- for (Entry<String, NodeState> compactedRoot :
compactedRoots.entrySet()) {
- String path = compactedRoot.getKey();
- NodeState state = compactedRoot.getValue();
- NodeBuilder childBuilder = getChild(builder,
getParentPath(path));
- childBuilder.setChildNode(getName(path), state);
- }
-
- // Use the segment writer of the *new generation* to persist the
compacted super root.
- RecordId nodeId = writer.writeNode(builder.getNodeState(),
uncompacted.getStableIdBytes());
- return new SegmentNodeState(segmentReader, segmentWriter,
getBlobStore(), nodeId);
- }
-
- /**
- * Compact a list of uncompacted roots on top of base roots of the
same key or
- * an empty node if none.
- */
- @CheckForNull
- private LinkedHashMap<String, NodeState> compact(
- @Nonnull LinkedHashMap<String, NodeState> baseRoots,
- @Nonnull LinkedHashMap<String, NodeState> uncompactedRoots,
- @Nonnull Compactor compactor)
- throws IOException {
- NodeState onto = baseRoots.get("root");
- NodeState previous = onto;
- LinkedHashMap<String, NodeState> compactedRoots =
newLinkedHashMap();
- for (Entry<String, NodeState> uncompactedRoot :
uncompactedRoots.entrySet()) {
- String path = uncompactedRoot.getKey();
- NodeState state = uncompactedRoot.getValue();
- NodeState compacted;
- if (onto == null) {
- compacted = compactor.compact(state);
- } else {
- compacted = compactor.compact(previous, state, onto);
- }
- if (compacted == null) {
- return null;
- }
- previous = state;
- onto = compacted;
- compactedRoots.put(path, compacted);
- }
- return compactedRoots;
- }
-
- /**
- * Collect a chronologically ordered list of roots for the base and
the uncompacted
- * state from a {@code superRoot} . This list consists of all
checkpoints followed by
- * the root.
- */
- @Nonnull
- private LinkedHashMap<String, NodeState> collectRoots(@Nullable
SegmentNodeState superRoot) {
- LinkedHashMap<String, NodeState> roots = newLinkedHashMap();
- if (superRoot != null) {
- List<ChildNodeEntry> checkpoints = newArrayList(
-
superRoot.getChildNode("checkpoints").getChildNodeEntries());
-
- checkpoints.sort((cne1, cne2) -> {
- long c1 = cne1.getNodeState().getLong("created");
- long c2 = cne2.getNodeState().getLong("created");
- return Long.compare(c1, c2);
- });
-
- for (ChildNodeEntry checkpoint : checkpoints) {
- roots.put("checkpoints/" + checkpoint.getName() + "/root",
- checkpoint.getNodeState().getChildNode("root"));
- }
- roots.put("root", superRoot.getChildNode("root"));
- }
- return roots;
- }
-
- @Nonnull
- private NodeBuilder getChild(NodeBuilder builder, String path) {
- for (String name : elements(path)) {
- builder = builder.getChildNode(name);
- }
- return builder;
- }
-
private SegmentNodeState forceCompact(
- @Nonnull final SegmentNodeState base,
- @Nonnull final Compactor compactor,
- @Nonnull SegmentWriter writer)
+ @Nullable final NodeState base,
+ @Nullable final NodeState onto,
+ @Nonnull final CheckpointCompactor compactor)
throws InterruptedException {
RecordId compactedId = revisions.setHead(new Function<RecordId,
RecordId>() {
@Nullable
@@ -967,8 +853,8 @@ public class FileStore extends AbstractF
public RecordId apply(RecordId headId) {
try {
long t0 = currentTimeMillis();
- SegmentNodeState after = compact(
- base, segmentReader.readNode(headId),
compactor, writer);
+ SegmentNodeState after = compactor.compact(
+ base, segmentReader.readNode(headId), onto);
if (after == null) {
gcListener.info("TarMK GC #{}: compaction
cancelled after {} seconds",
GC_COUNT, (currentTimeMillis() - t0) /
1000);
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorTest.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorTest.java?rev=1817776&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorTest.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/CheckpointCompactorTest.java
Mon Dec 11 13:56:50 2017
@@ -0,0 +1,207 @@
+/*
+ * 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 com.google.common.collect.Lists.newArrayList;
+import static java.util.concurrent.TimeUnit.DAYS;
+import static
org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static
org.apache.jackrabbit.oak.plugins.memory.MultiBinaryPropertyState.binaryPropertyFromBlob;
+import static
org.apache.jackrabbit.oak.segment.DefaultSegmentWriterBuilder.defaultSegmentWriterBuilder;
+import static
org.apache.jackrabbit.oak.segment.file.FileStoreBuilder.fileStoreBuilder;
+import static
org.apache.jackrabbit.oak.segment.file.tar.GCGeneration.newGCGeneration;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nonnull;
+
+import com.google.common.base.Suppliers;
+import org.apache.jackrabbit.oak.api.Blob;
+import org.apache.jackrabbit.oak.api.CommitFailedException;
+import org.apache.jackrabbit.oak.segment.file.FileStore;
+import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.InvalidFileStoreVersionException;
+import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
+import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+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.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class CheckpointCompactorTest {
+ @Rule
+ public TemporaryFolder folder = new TemporaryFolder(new File("target"));
+
+ private FileStore fileStore;
+
+ private SegmentNodeStore nodeStore;
+
+ private CheckpointCompactor compactor;
+
+ private GCGeneration compactedGeneration;
+
+ @Before
+ public void setup() throws IOException, InvalidFileStoreVersionException {
+ fileStore = fileStoreBuilder(folder.getRoot()).build();
+ nodeStore = SegmentNodeStoreBuilders.builder(fileStore).build();
+ compactedGeneration = newGCGeneration(1,1, true);
+ compactor = createCompactor(fileStore, compactedGeneration);
+ }
+
+ @After
+ public void tearDown() {
+ fileStore.close();
+ }
+
+ @Test
+ public void testCompact() throws Exception {
+ addTestContent("cp1", nodeStore);
+ String cp1 = nodeStore.checkpoint(DAYS.toMillis(1));
+ addTestContent("cp2", nodeStore);
+ String cp2 = nodeStore.checkpoint(DAYS.toMillis(1));
+
+ SegmentNodeState uncompacted1 = fileStore.getHead();
+ SegmentNodeState compacted1 = compactor.compact(EMPTY_NODE,
uncompacted1, EMPTY_NODE);
+ assertNotNull(compacted1);
+ assertFalse(uncompacted1 == compacted1);
+ checkGeneration(compacted1, compactedGeneration);
+
+ assertSameStableId(uncompacted1, compacted1);
+ assertSameStableId(getCheckpoint(uncompacted1, cp1),
getCheckpoint(compacted1, cp1));
+ assertSameStableId(getCheckpoint(uncompacted1, cp2),
getCheckpoint(compacted1, cp2));
+ assertSameRecord(getCheckpoint(compacted1, cp2),
compacted1.getChildNode("root"));
+
+ // Simulate a 2nd compaction cycle
+ addTestContent("cp3", nodeStore);
+ String cp3 = nodeStore.checkpoint(DAYS.toMillis(1));
+ addTestContent("cp4", nodeStore);
+ String cp4 = nodeStore.checkpoint(DAYS.toMillis(1));
+
+ SegmentNodeState uncompacted2 = fileStore.getHead();
+ SegmentNodeState compacted2 = compactor.compact(uncompacted1,
uncompacted2, compacted1);
+ assertNotNull(compacted2);
+ assertFalse(uncompacted2 == compacted2);
+ checkGeneration(compacted2, compactedGeneration);
+
+
assertTrue(fileStore.getRevisions().setHead(uncompacted2.getRecordId(),
compacted2.getRecordId()));
+
+ assertEquals(uncompacted2, compacted2);
+ assertSameStableId(uncompacted2, compacted2);
+ assertSameStableId(getCheckpoint(uncompacted2, cp1),
getCheckpoint(compacted2, cp1));
+ assertSameStableId(getCheckpoint(uncompacted2, cp2),
getCheckpoint(compacted2, cp2));
+ assertSameStableId(getCheckpoint(uncompacted2, cp3),
getCheckpoint(compacted2, cp3));
+ assertSameStableId(getCheckpoint(uncompacted2, cp4),
getCheckpoint(compacted2, cp4));
+ assertSameRecord(getCheckpoint(compacted1, cp1),
getCheckpoint(compacted2, cp1));
+ assertSameRecord(getCheckpoint(compacted1, cp2),
getCheckpoint(compacted2, cp2));
+ assertSameRecord(getCheckpoint(compacted2, cp4),
compacted2.getChildNode("root"));
+ }
+
+ private static void checkGeneration(NodeState node, GCGeneration
gcGeneration) {
+ assertTrue(node instanceof SegmentNodeState);
+ assertEquals(gcGeneration, ((SegmentNodeState)
node).getRecordId().getSegmentId().getGcGeneration());
+
+ for (ChildNodeEntry cne : node.getChildNodeEntries()) {
+ checkGeneration(cne.getNodeState(), gcGeneration);
+ }
+ }
+
+ private static NodeState getCheckpoint(NodeState superRoot, String name) {
+ NodeState checkpoint = superRoot
+ .getChildNode("checkpoints")
+ .getChildNode(name)
+ .getChildNode("root");
+ assertTrue(checkpoint.exists());
+ return checkpoint;
+ }
+
+ private static void assertSameStableId(NodeState node1, NodeState node2) {
+ assertTrue(node1 instanceof SegmentNodeState);
+ assertTrue(node2 instanceof SegmentNodeState);
+
+ assertEquals("Nodes should have the same stable ids",
+ ((SegmentNodeState) node1).getStableId(),
+ ((SegmentNodeState) node2).getStableId());
+ }
+
+ private static void assertSameRecord(NodeState node1, NodeState node2) {
+ assertTrue(node1 instanceof SegmentNodeState);
+ assertTrue(node2 instanceof SegmentNodeState);
+
+ assertEquals("Nodes should have been deduplicated",
+ ((SegmentNodeState) node1).getRecordId(),
+ ((SegmentNodeState) node2).getRecordId());
+ }
+
+ @Nonnull
+ private static CheckpointCompactor createCompactor(@Nonnull FileStore
fileStore, @Nonnull GCGeneration generation) {
+ SegmentWriter writer = defaultSegmentWriterBuilder("c")
+ .withGeneration(generation)
+ .build(fileStore);
+
+ return new CheckpointCompactor(
+ GCMonitor.EMPTY,
+ new AtomicLong(),
+ fileStore.getReader(),
+ writer,
+ fileStore.getBlobStore(),
+ Suppliers.ofInstance(false),
+ GCNodeWriteMonitor.EMPTY);
+ }
+
+ private static void addTestContent(@Nonnull String parent, @Nonnull
NodeStore nodeStore)
+ throws CommitFailedException, IOException {
+ NodeBuilder rootBuilder = nodeStore.getRoot().builder();
+ NodeBuilder parentBuilder = rootBuilder.child(parent);
+ parentBuilder.setChildNode("a").setChildNode("aa").setProperty("p",
42);
+ parentBuilder.getChildNode("a").setChildNode("bb").setChildNode("bbb");
+ parentBuilder.setChildNode("b").setProperty("bin",
createBlob(nodeStore, 42));
+
parentBuilder.setChildNode("c").setProperty(binaryPropertyFromBlob("bins",
createBlobs(nodeStore, 42, 43, 44)));
+ nodeStore.merge(rootBuilder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+ }
+
+ 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));
+ }
+
+ private static List<Blob> createBlobs(NodeStore nodeStore, int... sizes)
throws IOException {
+ List<Blob> blobs = newArrayList();
+ for (int size : sizes) {
+ blobs.add(createBlob(nodeStore, size));
+ }
+ return blobs;
+ }
+
+}
