You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@jackrabbit.apache.org by GitBox <gi...@apache.org> on 2022/09/07 07:59:28 UTC
[GitHub] [jackrabbit-oak] lweitzendorf opened a new pull request, #692: GRANITE-41007
lweitzendorf opened a new pull request, #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692
Parallelize compaction over subtrees
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r995415860
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ParallelCompactor.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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 org.apache.jackrabbit.oak.api.PropertyState;
+import org.apache.jackrabbit.oak.plugins.index.counter.ApproximateCounter;
+import org.apache.jackrabbit.oak.plugins.memory.MemoryNodeBuilder;
+import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.spi.state.NodeStateDiff;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.*;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static org.apache.jackrabbit.oak.segment.ClassicCompactor.getStableIdBytes;
+
+/**
+ * This compactor implementation leverages the tree structure of the repository for concurrent compaction.
+ * It explores the tree breadth-first until the target node count is reached. Every node at this depth will be
+ * an entry point for asynchronous compaction. After the exploration phase, the main thread will collect
+ * these compaction results and write their parents' node state to disk.
+ */
+public class ParallelCompactor extends CheckpointCompactor {
+ /**
+ * Expand repository tree until there are this many nodes for each worker to compact. Tradeoff
+ * between low efficiency of many small tasks and high risk of at least one of the subtrees being
+ * significantly larger than totalSize / numWorkers (unequal work distribution).
+ */
+ private static final int MIN_NODES_PER_WORKER = 1000;
+
+ /**
+ * Stop expansion if tree size grows beyond this many nodes per worker at the latest.
+ */
+ private static final int MAX_NODES_PER_WORKER = 10_000;
+
+ private final int numWorkers;
+
+ private final long totalSizeEstimate;
+
+ /**
+ * Manages workers for asynchronous compaction.
+ */
+ @Nullable
+ private ExecutorService executorService;
+
+ /**
+ * Create a new instance based on the passed arguments.
+ * @param gcListener listener receiving notifications about the garbage collection process
+ * @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 compactionMonitor notification call back for each compacted nodes, properties, and binaries
+ * @param nThreads number of threads to use for parallel compaction,
+ * negative numbers are interpreted relative to the number of available processors
+ */
+ public ParallelCompactor(
+ @NotNull GCMonitor gcListener,
+ @NotNull SegmentReader reader,
+ @NotNull SegmentWriter writer,
+ @Nullable BlobStore blobStore,
+ @NotNull GCNodeWriteMonitor compactionMonitor,
+ int nThreads) {
+ super(gcListener, reader, writer, blobStore, compactionMonitor);
+
+ int availableProcessors = Runtime.getRuntime().availableProcessors();
+ if (nThreads < 0) {
+ nThreads += availableProcessors + 1;
+ }
+ numWorkers = Math.max(0, nThreads - 1);
Review Comment:
This is actually not a mistake but intended behavior. Negative numbers are interpreted relative to the number of available processors. So specifying -1 means using all available cores, -2 all but one and then for your example -4 means all but 3 of the available processors (8-3 = 5). I've seen these semantics use in libraries such as [sklearn](https://scikit-learn.org/stable/glossary.html#term-n_jobs). Granted this is Python, but I think the semantics still make sense.
Yes you are correct, it degrades to the behavior of `CheckpointCompactor` (see first comment).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r995415860
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ParallelCompactor.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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 org.apache.jackrabbit.oak.api.PropertyState;
+import org.apache.jackrabbit.oak.plugins.index.counter.ApproximateCounter;
+import org.apache.jackrabbit.oak.plugins.memory.MemoryNodeBuilder;
+import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.spi.state.NodeStateDiff;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.*;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static org.apache.jackrabbit.oak.segment.ClassicCompactor.getStableIdBytes;
+
+/**
+ * This compactor implementation leverages the tree structure of the repository for concurrent compaction.
+ * It explores the tree breadth-first until the target node count is reached. Every node at this depth will be
+ * an entry point for asynchronous compaction. After the exploration phase, the main thread will collect
+ * these compaction results and write their parents' node state to disk.
+ */
+public class ParallelCompactor extends CheckpointCompactor {
+ /**
+ * Expand repository tree until there are this many nodes for each worker to compact. Tradeoff
+ * between low efficiency of many small tasks and high risk of at least one of the subtrees being
+ * significantly larger than totalSize / numWorkers (unequal work distribution).
+ */
+ private static final int MIN_NODES_PER_WORKER = 1000;
+
+ /**
+ * Stop expansion if tree size grows beyond this many nodes per worker at the latest.
+ */
+ private static final int MAX_NODES_PER_WORKER = 10_000;
+
+ private final int numWorkers;
+
+ private final long totalSizeEstimate;
+
+ /**
+ * Manages workers for asynchronous compaction.
+ */
+ @Nullable
+ private ExecutorService executorService;
+
+ /**
+ * Create a new instance based on the passed arguments.
+ * @param gcListener listener receiving notifications about the garbage collection process
+ * @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 compactionMonitor notification call back for each compacted nodes, properties, and binaries
+ * @param nThreads number of threads to use for parallel compaction,
+ * negative numbers are interpreted relative to the number of available processors
+ */
+ public ParallelCompactor(
+ @NotNull GCMonitor gcListener,
+ @NotNull SegmentReader reader,
+ @NotNull SegmentWriter writer,
+ @Nullable BlobStore blobStore,
+ @NotNull GCNodeWriteMonitor compactionMonitor,
+ int nThreads) {
+ super(gcListener, reader, writer, blobStore, compactionMonitor);
+
+ int availableProcessors = Runtime.getRuntime().availableProcessors();
+ if (nThreads < 0) {
+ nThreads += availableProcessors + 1;
+ }
+ numWorkers = Math.max(0, nThreads - 1);
Review Comment:
This is actually not a mistake but intended behavior. Negative numbers are interpreted relative to the number of available processors. So specifying -1 means using all available cores, -2 all but one and then for your example -4 means all but 3 of the available processors (8-3 = 5). I've seen these semantics used in libraries such as [sklearn](https://scikit-learn.org/stable/glossary.html#term-n_jobs). Granted this is Python, but I think the semantics still make sense.
Yes you are correct, it degrades to the behavior of `CheckpointCompactor` (see first comment).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r995418228
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java:
##########
@@ -198,7 +192,7 @@ private static GarbageCollectionStrategy newGarbageCollectionStrategy() {
defaultSegmentWriterBuilder("c")
.with(builder.getCacheManager().withAccessTracking("COMPACT", statsProvider))
.withGeneration(generation)
- .withoutWriterPool()
+ .withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
Review Comment:
A single `SegmentBufferWriter`, which is used if there is no writer pool, is not thread-safe. A writer pool for a single thread behaves the same as a `SegmentBufferWriter` and should only have minimal overhead with the thread-specific implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] jelmini commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
jelmini commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r1050652047
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPool.java:
##########
@@ -82,151 +62,292 @@ public class SegmentBufferWriterPool implements WriteOperationHandler {
private short writerId = -1;
- public SegmentBufferWriterPool(
+ private SegmentBufferWriterPool(
@NotNull SegmentIdProvider idProvider,
@NotNull SegmentReader reader,
@NotNull String wid,
@NotNull Supplier<GCGeneration> gcGeneration) {
- this.idProvider = checkNotNull(idProvider);
- this.reader = checkNotNull(reader);
- this.wid = checkNotNull(wid);
- this.gcGeneration = checkNotNull(gcGeneration);
+ this.idProvider = idProvider;
+ this.reader = reader;
+ this.wid = wid;
+ this.gcGeneration = gcGeneration;
}
- @Override
- @NotNull
- public GCGeneration getGCGeneration() {
- return gcGeneration.get();
+ public enum PoolType {
+ GLOBAL,
+ THREAD_SPECIFIC;
}
- @NotNull
- @Override
- public RecordId execute(@NotNull GCGeneration gcGeneration,
- @NotNull WriteOperation writeOperation)
- throws IOException {
- SimpleImmutableEntry<?,?> key = new SimpleImmutableEntry<>(currentThread(), gcGeneration);
- SegmentBufferWriter writer = borrowWriter(key, gcGeneration);
- try {
- return writeOperation.execute(writer);
- } finally {
- returnWriter(key, writer);
+ public static class SegmentBufferWriterPoolFactory {
+ @NotNull
+ private final SegmentIdProvider idProvider;
+ @NotNull
+ private final SegmentReader reader;
+ @NotNull
+ private final String wid;
+ @NotNull
+ private final Supplier<GCGeneration> gcGeneration;
+
+ private SegmentBufferWriterPoolFactory(
+ @NotNull SegmentIdProvider idProvider,
+ @NotNull SegmentReader reader,
+ @NotNull String wid,
+ @NotNull Supplier<GCGeneration> gcGeneration) {
+ this.idProvider = checkNotNull(idProvider);
+ this.reader = checkNotNull(reader);
+ this.wid = checkNotNull(wid);
+ this.gcGeneration = checkNotNull(gcGeneration);
+ }
+
+ @NotNull
+ public SegmentBufferWriterPool newPool(@NotNull SegmentBufferWriterPool.PoolType poolType) {
+ switch (poolType) {
+ case GLOBAL:
+ return new GlobalSegmentBufferWriterPool(idProvider, reader, wid, gcGeneration);
+ case THREAD_SPECIFIC:
+ return new ThreadSpecificSegmentBufferWriterPool(idProvider, reader, wid, gcGeneration);
+ default:
+ throw new IllegalArgumentException("Unknown writer pool type.");
+ }
}
}
- @Override
- public void flush(@NotNull SegmentStore store) throws IOException {
- List<SegmentBufferWriter> toFlush = newArrayList();
- List<SegmentBufferWriter> toReturn = newArrayList();
-
- poolMonitor.enter();
- try {
- // Collect all writers that are not currently in use and clear
- // the list so they won't get re-used anymore.
- toFlush.addAll(writers.values());
- writers.clear();
-
- // Collect all borrowed writers, which we need to wait for.
- // Clear the list so they will get disposed once returned.
- toReturn.addAll(borrowed);
- borrowed.clear();
- } finally {
- poolMonitor.leave();
+ public static SegmentBufferWriterPoolFactory factory(
+ @NotNull SegmentIdProvider idProvider,
+ @NotNull SegmentReader reader,
+ @NotNull String wid,
+ @NotNull Supplier<GCGeneration> gcGeneration) {
+ return new SegmentBufferWriterPoolFactory(idProvider, reader, wid, gcGeneration);
+ }
+
+ private static class ThreadSpecificSegmentBufferWriterPool extends SegmentBufferWriterPool {
+ /**
+ * Read write lock protecting the state of this pool. Multiple threads can access their writers in parallel,
+ * acquiring the read lock. The writer lock is needed for the flush operation since it requires none
+ * of the writers to be in use.
+ */
+ private final ReadWriteLock lock = new ReentrantReadWriteLock(true);
+
+ /**
+ * Pool of writers. Every thread is assigned a unique writer per GC generation, therefore only requiring
+ * a concurrent map to synchronize access to them.
+ */
+ private final ConcurrentMap<Object, SegmentBufferWriter> writers = newConcurrentMap();
+
+ public ThreadSpecificSegmentBufferWriterPool(
+ @NotNull SegmentIdProvider idProvider,
+ @NotNull SegmentReader reader,
+ @NotNull String wid,
+ @NotNull Supplier<GCGeneration> gcGeneration) {
+ super(idProvider, reader, wid, gcGeneration);
}
- // Wait for the return of the borrowed writers. This is the
- // case once all of them appear in the disposed set.
- if (safeEnterWhen(poolMonitor, allReturned(toReturn))) {
+ @NotNull
+ @Override
+ public RecordId execute(@NotNull GCGeneration gcGeneration,
+ @NotNull WriteOperation writeOperation)
+ throws IOException {
+ lock.readLock().lock();
+ SegmentBufferWriter writer = getWriter(currentThread(), gcGeneration);
try {
- // Collect all disposed writers and clear the list to mark them
- // as flushed.
- toFlush.addAll(toReturn);
- disposed.removeAll(toReturn);
+ return writeOperation.execute(writer);
} finally {
- poolMonitor.leave();
+ lock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public void flush(@NotNull SegmentStore store) throws IOException {
+ lock.writeLock().lock();
+ try {
+ for (SegmentBufferWriter writer : writers.values()) {
+ writer.flush(store);
+ }
+ writers.clear();
+ } finally {
+ lock.writeLock().unlock();
}
}
- // Call flush from outside the pool monitor to avoid potential
- // deadlocks of that method calling SegmentStore.writeSegment
- for (SegmentBufferWriter writer : toFlush) {
- writer.flush(store);
+ @NotNull
+ private SegmentBufferWriter getWriter(@NotNull Thread thread, @NotNull GCGeneration gcGeneration) {
+ SimpleImmutableEntry<?,?> key = new SimpleImmutableEntry<>(thread, gcGeneration);
+ SegmentBufferWriter writer = writers.get(key);
Review Comment:
This code can be replaced with:
```java
return writers.computeIfAbsent(key, k -> newWriter(gcGeneration));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] jelmini commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
jelmini commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r1050643260
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPool.java:
##########
@@ -82,151 +62,292 @@ public class SegmentBufferWriterPool implements WriteOperationHandler {
private short writerId = -1;
- public SegmentBufferWriterPool(
+ private SegmentBufferWriterPool(
@NotNull SegmentIdProvider idProvider,
@NotNull SegmentReader reader,
@NotNull String wid,
@NotNull Supplier<GCGeneration> gcGeneration) {
- this.idProvider = checkNotNull(idProvider);
- this.reader = checkNotNull(reader);
- this.wid = checkNotNull(wid);
- this.gcGeneration = checkNotNull(gcGeneration);
+ this.idProvider = idProvider;
+ this.reader = reader;
+ this.wid = wid;
+ this.gcGeneration = gcGeneration;
}
- @Override
- @NotNull
- public GCGeneration getGCGeneration() {
- return gcGeneration.get();
+ public enum PoolType {
+ GLOBAL,
+ THREAD_SPECIFIC;
}
- @NotNull
- @Override
- public RecordId execute(@NotNull GCGeneration gcGeneration,
- @NotNull WriteOperation writeOperation)
- throws IOException {
- SimpleImmutableEntry<?,?> key = new SimpleImmutableEntry<>(currentThread(), gcGeneration);
- SegmentBufferWriter writer = borrowWriter(key, gcGeneration);
- try {
- return writeOperation.execute(writer);
- } finally {
- returnWriter(key, writer);
+ public static class SegmentBufferWriterPoolFactory {
+ @NotNull
+ private final SegmentIdProvider idProvider;
+ @NotNull
+ private final SegmentReader reader;
+ @NotNull
+ private final String wid;
+ @NotNull
+ private final Supplier<GCGeneration> gcGeneration;
+
+ private SegmentBufferWriterPoolFactory(
+ @NotNull SegmentIdProvider idProvider,
+ @NotNull SegmentReader reader,
+ @NotNull String wid,
+ @NotNull Supplier<GCGeneration> gcGeneration) {
+ this.idProvider = checkNotNull(idProvider);
Review Comment:
Can be replaced by `Objects.requireNonNull`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r997143724
##########
oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/RecordCacheStatsTest.java:
##########
@@ -37,18 +37,9 @@ public class RecordCacheStatsTest {
private final Random rnd = new Random();
private final MemoryStore store = new MemoryStore();
- private final RecordCache<Integer> cache = newRecordCache(KEYS);
+ private final RecordCache<Integer> cache = newRecordCache(KEYS * 4 / 3);
Review Comment:
Turns out using `ConcurrentLinkedHashMap` directly is significantly slower. The option of keeping both implementations is still there if necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r995532315
##########
oak-doc/src/site/markdown/nodestore/segment/overview.md:
##########
@@ -807,24 +823,25 @@ This option is optional and is disabled by default.
### <a name="compact"/> Compact
```
-java -jar oak-run.jar compact [--force] [--mmap] [--compactor] SOURCE [--target-path DESTINATION] [--persistent-cache-path PERSISTENT_CACHE_PATH] [--persistent-cache-size-gb <PERSISTENT_CACHE_SIZE_GB>]
+java -jar oak-run.jar compact [--force] [--mmap] [--compactor] [--threads] SOURCE [--target-path DESTINATION] [--persistent-cache-path PERSISTENT_CACHE_PATH] [--persistent-cache-size-gb <PERSISTENT_CACHE_SIZE_GB>]
```
The `compact` command performs offline compaction of the local/remote Segment Store at `SOURCE`.
`SOURCE` must be a valid path/uri to an existing Segment Store. Currently, Azure Segment Store and AWS Segment Store the supported remote Segment Stores.
Please refer to the [Remote Segment Stores](#remote-segment-stores) section for details on how to correctly specify connection URIs.
-If the optional `--force [Boolean]` argument is set to `true` the tool ignores a non
-matching Segment Store version. *CAUTION*: this will upgrade the Segment Store to the
+If the optional `--force [Boolean]` argument is set to `true` the tool ignores a non-matching Segment Store version. *CAUTION*: this will upgrade the Segment Store to the
latest version, which is incompatible with older versions. *There is no way to downgrade
an accidentally upgraded Segment Store*.
The optional `--mmap [Boolean]` argument can be used to control the file access mode. Set
to `true` for memory mapped access and `false` for file access. If not specified, memory
-mapped access is used on 64 bit systems and file access is used on 32 bit systems. On
+mapped access is used on 64-bit systems and file access is used on 32-bit systems. On
Windows, regular file access is always enforced and this option is ignored.
-The optional `--compactor [String]` argument can be used to pick the compactor type to be used. Valid choices are *classic* and *diff*. While the former is slower, it might be more stable, due to lack of optimisations employed by the *diff* compactor which compacts the checkpoints on top of each other. If not specified, *diff* compactor is used.
+The optional `--compactor [String]` argument can be used to pick the compactor type to be used. Valid choices are *classic*, *diff* and *parallel*. While *classic* is slower, it might be more stable, due to lack of optimisations employed by the *diff* compactor which compacts the checkpoints on top of each other and the *parallel* compactor, which additionally divides the repository into multiple parts to process in parallel. If not specified, *parallel* compactor is used.
Review Comment:
I agree that it's risky to default to the multithreaded `ParallelCompactor`. However, I also think defaulting to a single thread should be safe (`DEFAULT_CONCURRENCY = 1` in `SegmentGCOptions`). If you have a look at the `compactWithDelegate` method, it should be easy to verify that the behavior is in fact identical to `CheckpointCompactor` in this case. I will change the call for sequential compaction to `super.compactWithDelegate(before, after, onto, canceller)` to make this even more explicit. If you still have concerns, I will change all of the defaults.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] dulceanu commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
dulceanu commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r997888179
##########
oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/RecordCacheStatsTest.java:
##########
@@ -37,18 +37,9 @@ public class RecordCacheStatsTest {
private final Random rnd = new Random();
private final MemoryStore store = new MemoryStore();
- private final RecordCache<Integer> cache = newRecordCache(KEYS);
+ private final RecordCache<Integer> cache = newRecordCache(KEYS * 4 / 3);
Review Comment:
I also think that keeping both implementations of the record cache is the safest option here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] dulceanu merged pull request #692: OAK-9922 Parallel Compaction
Posted by "dulceanu (via GitHub)" <gi...@apache.org>.
dulceanu merged PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r995532315
##########
oak-doc/src/site/markdown/nodestore/segment/overview.md:
##########
@@ -807,24 +823,25 @@ This option is optional and is disabled by default.
### <a name="compact"/> Compact
```
-java -jar oak-run.jar compact [--force] [--mmap] [--compactor] SOURCE [--target-path DESTINATION] [--persistent-cache-path PERSISTENT_CACHE_PATH] [--persistent-cache-size-gb <PERSISTENT_CACHE_SIZE_GB>]
+java -jar oak-run.jar compact [--force] [--mmap] [--compactor] [--threads] SOURCE [--target-path DESTINATION] [--persistent-cache-path PERSISTENT_CACHE_PATH] [--persistent-cache-size-gb <PERSISTENT_CACHE_SIZE_GB>]
```
The `compact` command performs offline compaction of the local/remote Segment Store at `SOURCE`.
`SOURCE` must be a valid path/uri to an existing Segment Store. Currently, Azure Segment Store and AWS Segment Store the supported remote Segment Stores.
Please refer to the [Remote Segment Stores](#remote-segment-stores) section for details on how to correctly specify connection URIs.
-If the optional `--force [Boolean]` argument is set to `true` the tool ignores a non
-matching Segment Store version. *CAUTION*: this will upgrade the Segment Store to the
+If the optional `--force [Boolean]` argument is set to `true` the tool ignores a non-matching Segment Store version. *CAUTION*: this will upgrade the Segment Store to the
latest version, which is incompatible with older versions. *There is no way to downgrade
an accidentally upgraded Segment Store*.
The optional `--mmap [Boolean]` argument can be used to control the file access mode. Set
to `true` for memory mapped access and `false` for file access. If not specified, memory
-mapped access is used on 64 bit systems and file access is used on 32 bit systems. On
+mapped access is used on 64-bit systems and file access is used on 32-bit systems. On
Windows, regular file access is always enforced and this option is ignored.
-The optional `--compactor [String]` argument can be used to pick the compactor type to be used. Valid choices are *classic* and *diff*. While the former is slower, it might be more stable, due to lack of optimisations employed by the *diff* compactor which compacts the checkpoints on top of each other. If not specified, *diff* compactor is used.
+The optional `--compactor [String]` argument can be used to pick the compactor type to be used. Valid choices are *classic*, *diff* and *parallel*. While *classic* is slower, it might be more stable, due to lack of optimisations employed by the *diff* compactor which compacts the checkpoints on top of each other and the *parallel* compactor, which additionally divides the repository into multiple parts to process in parallel. If not specified, *parallel* compactor is used.
Review Comment:
I agree that it's risky to default to the multithreaded `ParallelCompactor`. However, I also think defaulting to a single thread should be safe. If you have a look at the `compactWithDelegate` method, it should be easy to verify that the behavior is in fact identical to `CheckpointCompactor` in this case. I will change the call for sequential compaction to `super.compactWithDelegate(before, after, onto, canceller)` to make this even more explicit. If you still have concerns, I will change all of the defaults to diff but keep the default thread count for parallel at -1 (if the semantics of thread count discussed in the other comment are acceptable).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r995418228
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java:
##########
@@ -198,7 +192,7 @@ private static GarbageCollectionStrategy newGarbageCollectionStrategy() {
defaultSegmentWriterBuilder("c")
.with(builder.getCacheManager().withAccessTracking("COMPACT", statsProvider))
.withGeneration(generation)
- .withoutWriterPool()
+ .withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
Review Comment:
A single `SegmentBufferWriter`, which is used if there is no writer pool, is not thread-safe and a writer pool for a single thread behaves the same as a `SegmentBufferWriter` and should only have minimal overhead with the thread-specific implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] reschke commented on pull request #692: GRANITE-41007
Posted by GitBox <gi...@apache.org>.
reschke commented on PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#issuecomment-1239119767
Please only use OAK JIRA tickets in PR titles and commit messages,
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r998168408
##########
oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/RecordCacheStatsTest.java:
##########
@@ -37,18 +37,9 @@ public class RecordCacheStatsTest {
private final Random rnd = new Random();
private final MemoryStore store = new MemoryStore();
- private final RecordCache<Integer> cache = newRecordCache(KEYS);
+ private final RecordCache<Integer> cache = newRecordCache(KEYS * 4 / 3);
Review Comment:
Nevermind the above. I restored the previous behavior by initializing the internal cache to 4/3 of the specified size. The old implementation also did this with its `LinkedHashMap`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r995415860
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ParallelCompactor.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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 org.apache.jackrabbit.oak.api.PropertyState;
+import org.apache.jackrabbit.oak.plugins.index.counter.ApproximateCounter;
+import org.apache.jackrabbit.oak.plugins.memory.MemoryNodeBuilder;
+import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.spi.state.NodeStateDiff;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.*;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static org.apache.jackrabbit.oak.segment.ClassicCompactor.getStableIdBytes;
+
+/**
+ * This compactor implementation leverages the tree structure of the repository for concurrent compaction.
+ * It explores the tree breadth-first until the target node count is reached. Every node at this depth will be
+ * an entry point for asynchronous compaction. After the exploration phase, the main thread will collect
+ * these compaction results and write their parents' node state to disk.
+ */
+public class ParallelCompactor extends CheckpointCompactor {
+ /**
+ * Expand repository tree until there are this many nodes for each worker to compact. Tradeoff
+ * between low efficiency of many small tasks and high risk of at least one of the subtrees being
+ * significantly larger than totalSize / numWorkers (unequal work distribution).
+ */
+ private static final int MIN_NODES_PER_WORKER = 1000;
+
+ /**
+ * Stop expansion if tree size grows beyond this many nodes per worker at the latest.
+ */
+ private static final int MAX_NODES_PER_WORKER = 10_000;
+
+ private final int numWorkers;
+
+ private final long totalSizeEstimate;
+
+ /**
+ * Manages workers for asynchronous compaction.
+ */
+ @Nullable
+ private ExecutorService executorService;
+
+ /**
+ * Create a new instance based on the passed arguments.
+ * @param gcListener listener receiving notifications about the garbage collection process
+ * @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 compactionMonitor notification call back for each compacted nodes, properties, and binaries
+ * @param nThreads number of threads to use for parallel compaction,
+ * negative numbers are interpreted relative to the number of available processors
+ */
+ public ParallelCompactor(
+ @NotNull GCMonitor gcListener,
+ @NotNull SegmentReader reader,
+ @NotNull SegmentWriter writer,
+ @Nullable BlobStore blobStore,
+ @NotNull GCNodeWriteMonitor compactionMonitor,
+ int nThreads) {
+ super(gcListener, reader, writer, blobStore, compactionMonitor);
+
+ int availableProcessors = Runtime.getRuntime().availableProcessors();
+ if (nThreads < 0) {
+ nThreads += availableProcessors + 1;
+ }
+ numWorkers = Math.max(0, nThreads - 1);
Review Comment:
This is actually not a mistake but intended behavior. Negative numbers are interpreted relative to the number of available processors. So specifying -1 means using all available cores, -2 all but one and then for your example -4 means all but 3 of the available processors (8-3 = 5). I've seen these semantics use in libraries such as [sklearn](https://scikit-learn.org/stable/glossary.html#term-n_jobs). Granted this is Python, but I think the semantics still make sense.
Yes you are correct, it degrades to the behavior of `CheckpointCompactor`. With this in mind, what do you think about keeping the `ParallelCompactor` as standard, but default to a single thread?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] jelmini commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
jelmini commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r994290034
##########
oak-segment-aws/src/main/java/org/apache/jackrabbit/oak/segment/aws/tool/AwsCompact.java:
##########
@@ -128,6 +133,27 @@ public Builder withGCLogInterval(long gcLogInterval) {
return this;
}
+ /**
+ * The compactor type to be used by compaction. If not specified it defaults to
+ * "parallel" compactor
Review Comment:
Should default to `diff`.
##########
oak-segment-aws/src/main/java/org/apache/jackrabbit/oak/segment/aws/tool/AwsCompact.java:
##########
@@ -72,6 +73,10 @@ public static class Builder {
private int segmentCacheSize = DEFAULT_SEGMENT_CACHE_MB;
+ private CompactorType compactorType = CompactorType.PARALLEL_COMPACTOR;
Review Comment:
Should default to `CompactorType.CHECKPOINT_COMPACTOR`
##########
oak-segment-tar/pom.xml:
##########
@@ -411,8 +411,6 @@
<groupId>org.apache.jackrabbit</groupId>
<artifactId>oak-core</artifactId>
<version>${project.version}</version>
- <classifier>tests</classifier>
Review Comment:
I don't think we should bring `oak-core` as a compile-time dependency. There was an effort to depend only on `oak-core-spi` and `oak-api`, related to [this Oak issue](https://issues.apache.org/jira/browse/OAK-6073) about allowing for independent releases.
As the objective of this dependency is only to bring a single class, `ApproximateCounter`, I would consider 2 options:
* Copy `ApproximateCounter` to `oak-store-spi`, modify all code to use it instead and deprecate the one in `oak-core`. `oak-store-spi` is the module where this class has the least dependencies to other modules. Moreover, it's depended upon by both `oak-core` and `oak-segment-tar`.
* If the above is too complex, simply copy `ApproximateCounter` to `oak-segment-tar`.
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/RecordCache.java:
##########
@@ -19,29 +19,21 @@
package org.apache.jackrabbit.oak.segment;
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import java.util.LinkedHashMap;
-import java.util.Map;
-
import com.google.common.base.Supplier;
-import com.google.common.cache.CacheStats;
-import com.google.common.cache.Weigher;
-
+import com.google.common.cache.*;
Review Comment:
[Code Style] We should not use * imports
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/memory/MemoryStore.java:
##########
@@ -72,7 +62,9 @@ public SegmentId newSegmentId(long msb, long lsb) {
});
this.revisions = new MemoryStoreRevisions();
this.segmentReader = new CachingSegmentReader(this::getWriter, null, 16, 2, NoopStats.INSTANCE);
- this.segmentWriter = defaultSegmentWriterBuilder("sys").withWriterPool().build(this);
+ this.segmentWriter = defaultSegmentWriterBuilder("sys")
Review Comment:
Given my other comment about keeping `withWriterPool()`, I suggest reverting this changes.
##########
oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/tool/AzureCompact.java:
##########
@@ -79,7 +79,9 @@ public static class Builder {
private int segmentCacheSize = 2048;
- private CompactorType compactorType = CompactorType.CHECKPOINT_COMPACTOR;
+ private CompactorType compactorType = CompactorType.PARALLEL_COMPACTOR;
Review Comment:
Should default to CompactorType.CHECKPOINT_COMPACTOR
##########
oak-run/src/main/java/org/apache/jackrabbit/oak/run/CompactCommand.java:
##########
@@ -56,11 +54,16 @@ public void execute(String... args) throws Exception {
.withOptionalArg()
.ofType(Boolean.class);
OptionSpec<String> compactor = parser.accepts("compactor",
- "Allow the user to control compactor type to be used. Valid choices are \"classic\" and \"diff\". " +
- "While the former is slower, it might be more stable, due to lack of optimisations employed " +
- "by the \"diff\" compactor which compacts the checkpoints on top of each other. If not " +
- "specified, \"diff\" compactor is used.")
+ "Allow the user to control compactor type to be used. Valid choices are \"classic\", \"diff\", \"parallel\". " +
+ "While \"classic\" is slower, it might be more stable, due to lack of optimisations employed " +
+ "by the \"diff\" compactor which compacts the checkpoints on top of each other and \"parallel\" compactor, which splits " +
+ "the repository into smaller parts and compacts them concurrently. If not specified, \"parallel\" compactor is used.")
Review Comment:
See the other comment about keeping `diff` as default compactor.
##########
oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorExternalBlobTest.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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 java.util.concurrent.TimeUnit.DAYS;
+import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.addTestContent;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.assertSameRecord;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.assertSameStableId;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.checkGeneration;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.createBlob;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.getCheckpoint;
+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.*;
Review Comment:
[Code style] We should not use * imports.
##########
oak-segment-aws/src/main/java/org/apache/jackrabbit/oak/segment/aws/tool/AwsToolUtils.java:
##########
@@ -69,14 +70,26 @@ public String description(String pathOrUri) {
}
public static FileStore newFileStore(SegmentNodeStorePersistence persistence, File directory,
- boolean strictVersionCheck, int segmentCacheSize, long gcLogInterval)
- throws IOException, InvalidFileStoreVersionException, URISyntaxException {
+ boolean strictVersionCheck, int segmentCacheSize, long gcLogInterval)
+ throws IOException, InvalidFileStoreVersionException {
+ return newFileStore(persistence, directory, strictVersionCheck, segmentCacheSize,
+ gcLogInterval, CompactorType.PARALLEL_COMPACTOR, 1);
Review Comment:
Should default to CompactorType.CHECKPOINT_COMPACTOR
##########
oak-segment-azure/src/main/java/org/apache/jackrabbit/oak/segment/azure/tool/AzureCompact.java:
##########
@@ -159,7 +161,7 @@ public Builder withGCLogInterval(long gcLogInterval) {
/**
* The compactor type to be used by compaction. If not specified it defaults to
- * "diff" compactor
+ * "parallel" compactor
Review Comment:
Should default to `diff`
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/DefaultSegmentWriterBuilder.java:
##########
@@ -29,10 +30,11 @@
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.segment.memory.MemoryStore;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
* Builder for building {@link DefaultSegmentWriter} instances.
- * The returned instances are thread safe if {@link #withWriterPool()}
+ * The returned instances are thread safe if {@link #withWriterPool(PoolType)}
* was specified and <em>not</em> thread sage if {@link #withoutWriterPool()}
Review Comment:
Typo: thread sage -> thread safe
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CheckpointCompactor.java:
##########
@@ -18,13 +18,15 @@
package org.apache.jackrabbit.oak.segment;
+import static com.google.common.base.Preconditions.checkNotNull;
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 static org.apache.jackrabbit.oak.segment.ClassicCompactor.getStableIdBytes;
Review Comment:
I appreciate the idea of removing duplication by depending on a single implementation of `getStableIdBytes()`. However, this method does not really belong to `ClassicCompactor`, nor any other `Compactor` implementation, and now there is an arbitrary dependency to `ClassicCompactor` here. I suggest to move this method to a new class `CompactorUtils` (under `util` package).
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/DefaultSegmentWriterBuilder.java:
##########
@@ -106,8 +108,8 @@ public DefaultSegmentWriterBuilder withGeneration(@NotNull GCGeneration generati
* The returned instance is thread safe.
*/
@NotNull
- public DefaultSegmentWriterBuilder withWriterPool() {
- this.pooled = true;
+ public DefaultSegmentWriterBuilder withWriterPool(PoolType writerType) {
Review Comment:
This is changing the signature of a public method in a public class. If some third-party implemented a custom store using this class, then this change could break their code.
I suggest adding an override with the previous signature, setting a default value:
```java
public DefaultSegmentWriterBuilder withWriterPool() {
this.poolType = PoolType.GLOBAL;
return this;
}
```
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ParallelCompactor.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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 org.apache.jackrabbit.oak.api.PropertyState;
+import org.apache.jackrabbit.oak.plugins.index.counter.ApproximateCounter;
+import org.apache.jackrabbit.oak.plugins.memory.MemoryNodeBuilder;
+import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.spi.state.NodeStateDiff;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.*;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static org.apache.jackrabbit.oak.segment.ClassicCompactor.getStableIdBytes;
+
+/**
+ * This compactor implementation leverages the tree structure of the repository for concurrent compaction.
+ * It explores the tree breadth-first until the target node count is reached. Every node at this depth will be
+ * an entry point for asynchronous compaction. After the exploration phase, the main thread will collect
+ * these compaction results and write their parents' node state to disk.
+ */
+public class ParallelCompactor extends CheckpointCompactor {
+ /**
+ * Expand repository tree until there are this many nodes for each worker to compact. Tradeoff
+ * between low efficiency of many small tasks and high risk of at least one of the subtrees being
+ * significantly larger than totalSize / numWorkers (unequal work distribution).
+ */
+ private static final int MIN_NODES_PER_WORKER = 1000;
+
+ /**
+ * Stop expansion if tree size grows beyond this many nodes per worker at the latest.
+ */
+ private static final int MAX_NODES_PER_WORKER = 10_000;
+
+ private final int numWorkers;
+
+ private final long totalSizeEstimate;
+
+ /**
+ * Manages workers for asynchronous compaction.
+ */
+ @Nullable
+ private ExecutorService executorService;
+
+ /**
+ * Create a new instance based on the passed arguments.
+ * @param gcListener listener receiving notifications about the garbage collection process
+ * @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 compactionMonitor notification call back for each compacted nodes, properties, and binaries
+ * @param nThreads number of threads to use for parallel compaction,
+ * negative numbers are interpreted relative to the number of available processors
+ */
+ public ParallelCompactor(
+ @NotNull GCMonitor gcListener,
+ @NotNull SegmentReader reader,
+ @NotNull SegmentWriter writer,
+ @Nullable BlobStore blobStore,
+ @NotNull GCNodeWriteMonitor compactionMonitor,
+ int nThreads) {
+ super(gcListener, reader, writer, blobStore, compactionMonitor);
+
+ int availableProcessors = Runtime.getRuntime().availableProcessors();
+ if (nThreads < 0) {
+ nThreads += availableProcessors + 1;
+ }
+ numWorkers = Math.max(0, nThreads - 1);
Review Comment:
Not sure I understand the logic to determine the number of workers.
First, I think the `+=` is a mistake and should actually be a simple `=`. Otherwise, assuming `availableProcessors=8`, if someone passes for example `nThreads=-4` in the constructor, then `nThreads` would be set to `-4 + 8 + 1 = 5` which would be surprising.
Second, we seem to support `numWorkers = 0`. What's the behaviour in this case? Does it degrade to the same behaviour than a single-threaded `CheckpointCompactor`?
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ParallelCompactor.java:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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 org.apache.jackrabbit.oak.api.PropertyState;
+import org.apache.jackrabbit.oak.plugins.index.counter.ApproximateCounter;
+import org.apache.jackrabbit.oak.plugins.memory.MemoryNodeBuilder;
+import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.spi.state.NodeStateDiff;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.*;
Review Comment:
[Code Style] We should not use * imports
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java:
##########
@@ -40,13 +40,7 @@
import com.google.common.util.concurrent.UncheckedExecutionException;
import org.apache.jackrabbit.oak.commons.Buffer;
-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.SegmentNodeState;
-import org.apache.jackrabbit.oak.segment.SegmentNotFoundException;
-import org.apache.jackrabbit.oak.segment.SegmentNotFoundExceptionListener;
-import org.apache.jackrabbit.oak.segment.SegmentWriter;
+import org.apache.jackrabbit.oak.segment.*;
Review Comment:
[Code Style] We should not use * imports
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/AbstractCompactionStrategy.java:
##########
@@ -29,12 +29,7 @@
import com.google.common.base.Function;
-import org.apache.jackrabbit.oak.segment.CheckpointCompactor;
-import org.apache.jackrabbit.oak.segment.ClassicCompactor;
-import org.apache.jackrabbit.oak.segment.Compactor;
-import org.apache.jackrabbit.oak.segment.RecordId;
-import org.apache.jackrabbit.oak.segment.SegmentNodeState;
-import org.apache.jackrabbit.oak.segment.SegmentWriter;
+import org.apache.jackrabbit.oak.segment.*;
Review Comment:
[Code Style] We should not use * imports
##########
oak-doc/src/site/markdown/nodestore/segment/overview.md:
##########
@@ -807,24 +823,25 @@ This option is optional and is disabled by default.
### <a name="compact"/> Compact
```
-java -jar oak-run.jar compact [--force] [--mmap] [--compactor] SOURCE [--target-path DESTINATION] [--persistent-cache-path PERSISTENT_CACHE_PATH] [--persistent-cache-size-gb <PERSISTENT_CACHE_SIZE_GB>]
+java -jar oak-run.jar compact [--force] [--mmap] [--compactor] [--threads] SOURCE [--target-path DESTINATION] [--persistent-cache-path PERSISTENT_CACHE_PATH] [--persistent-cache-size-gb <PERSISTENT_CACHE_SIZE_GB>]
```
The `compact` command performs offline compaction of the local/remote Segment Store at `SOURCE`.
`SOURCE` must be a valid path/uri to an existing Segment Store. Currently, Azure Segment Store and AWS Segment Store the supported remote Segment Stores.
Please refer to the [Remote Segment Stores](#remote-segment-stores) section for details on how to correctly specify connection URIs.
-If the optional `--force [Boolean]` argument is set to `true` the tool ignores a non
-matching Segment Store version. *CAUTION*: this will upgrade the Segment Store to the
+If the optional `--force [Boolean]` argument is set to `true` the tool ignores a non-matching Segment Store version. *CAUTION*: this will upgrade the Segment Store to the
latest version, which is incompatible with older versions. *There is no way to downgrade
an accidentally upgraded Segment Store*.
The optional `--mmap [Boolean]` argument can be used to control the file access mode. Set
to `true` for memory mapped access and `false` for file access. If not specified, memory
-mapped access is used on 64 bit systems and file access is used on 32 bit systems. On
+mapped access is used on 64-bit systems and file access is used on 32-bit systems. On
Windows, regular file access is always enforced and this option is ignored.
-The optional `--compactor [String]` argument can be used to pick the compactor type to be used. Valid choices are *classic* and *diff*. While the former is slower, it might be more stable, due to lack of optimisations employed by the *diff* compactor which compacts the checkpoints on top of each other. If not specified, *diff* compactor is used.
+The optional `--compactor [String]` argument can be used to pick the compactor type to be used. Valid choices are *classic*, *diff* and *parallel*. While *classic* is slower, it might be more stable, due to lack of optimisations employed by the *diff* compactor which compacts the checkpoints on top of each other and the *parallel* compactor, which additionally divides the repository into multiple parts to process in parallel. If not specified, *parallel* compactor is used.
Review Comment:
It says here that `if not specified, *parallel* compactor is used`. I think it's safer to keep the previous behaviour, which is to use the `diff` compactor.
In my opinion the new parallel compactor should be considered in preview mode at first. Once we have more experience with it in production (validating it on a limited number of customers) and we become certain of its robustness, we may make it the default compactor.
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/PriorityCache.java:
##########
@@ -156,13 +154,40 @@ public static long nextPowerOfTwo(int size) {
* @param weigher Needed to provide an estimation of the cache weight in memory
*/
public PriorityCache(int size, int rehash, @NotNull Weigher<K, V> weigher) {
+ this(size, rehash, weigher, 1024);
+ }
+
+ /**
+ * Create a new instance of the given {@code size}. {@code rehash} specifies the number
+ * of rehashes to resolve a clash.
+ * @param size Size of the cache. Must be a power of {@code 2}.
+ * @param rehash Number of rehashes. Must be greater or equal to {@code 0} and
+ * smaller than {@code 32 - numberOfTrailingZeros(size)}.
+ * @param weigher Needed to provide an estimation of the cache weight in memory
+ * @param numSegments Number of separately locked segments
+ */
+ public PriorityCache(int size, int rehash, @NotNull Weigher<K, V> weigher, int numSegments) {
checkArgument(bitCount(size) == 1);
checkArgument(rehash >= 0);
checkArgument(rehash < 32 - numberOfTrailingZeros(size));
this.rehash = rehash;
entries = new Entry<?,?>[size];
fill(entries, Entry.NULL);
this.weigher = checkNotNull(weigher);
+
+ numSegments = Math.min(numSegments, size);
+ checkArgument((size % numSegments) == 0);
Review Comment:
I suggest to add an error message explaining why this check failed and how to fix it, ie what are acceptables values of `numSegments` in relation to `size`.
The Javadoc for `numSegments` should also explain this.
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java:
##########
@@ -198,7 +192,7 @@ private static GarbageCollectionStrategy newGarbageCollectionStrategy() {
defaultSegmentWriterBuilder("c")
.with(builder.getCacheManager().withAccessTracking("COMPACT", statsProvider))
.withGeneration(generation)
- .withoutWriterPool()
+ .withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
Review Comment:
Why not keep using `withoutWriterPool()` here?
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/PriorityCache.java:
##########
@@ -218,42 +259,45 @@ public synchronized boolean put(@NotNull K key, @NotNull V value, int generation
initialCost++;
}
eviction = false;
- break;
} else if (entry.generation < generation) {
// Old generation -> use this index
index = i;
eviction = false;
- break;
} else if (entry.cost < cheapest) {
// Candidate slot, keep on searching for even cheaper slots
cheapest = entry.cost;
index = i;
eviction = true;
+ if (k < rehash) {
+ continue;
+ }
+ } else {
Review Comment:
Is there a reason other than stylistic for the use of `continue` instead of `break` in this loop?
I find the previous version using `break` more readable than this one.
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/PriorityCache.java:
##########
@@ -289,35 +338,42 @@ public synchronized V get(@NotNull K key, int generation) {
* passed {@code purge} predicate.
* @param purge
*/
- public synchronized void purgeGenerations(@NotNull Predicate<Integer> purge) {
- for (int i = 0; i < entries.length; i++) {
- Entry<?, ?> entry = entries[i];
- if (entry != Entry.NULL && purge.apply(entry.generation)) {
- entries[i] = Entry.NULL;
- size--;
- weight -= weighEntry(entry);
+ public void purgeGenerations(@NotNull Predicate<Integer> purge) {
+ int numSegments = segments.length;
+ int entriesPerSegment = entries.length / numSegments;
+ for (int s = 0; s < numSegments; s++) {
+ segments[s].lock();
+ for (int i = 0; i < entriesPerSegment; i++) {
+ int j = i + s * entriesPerSegment;
+ Entry<?, ?> entry = entries[j];
+ if (entry != Entry.NULL && purge.apply(entry.generation)) {
+ entries[j] = Entry.NULL;
+ size.decrementAndGet();
+ weight.addAndGet(-weighEntry(entry));
+ }
}
+ segments[s].unlock();
Review Comment:
In case of exception, `unlock()` will not be called. It's safer to call it in a `finally` clause. The `try` block should start just after `lock()` is called.
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/PriorityCache.java:
##########
@@ -218,42 +259,45 @@ public synchronized boolean put(@NotNull K key, @NotNull V value, int generation
initialCost++;
}
eviction = false;
- break;
} else if (entry.generation < generation) {
// Old generation -> use this index
index = i;
eviction = false;
- break;
} else if (entry.cost < cheapest) {
// Candidate slot, keep on searching for even cheaper slots
cheapest = entry.cost;
index = i;
eviction = true;
+ if (k < rehash) {
+ continue;
+ }
+ } else {
+ continue;
}
- }
- if (index >= 0) {
Entry<?, ?> old = entries[index];
Entry<?, ?> newE = new Entry<>(key, value, generation, initialCost);
entries[index] = newE;
- loadCount++;
- costs[initialCost - Byte.MIN_VALUE]++;
+ loadCount.incrementAndGet();
+ costs[initialCost - Byte.MIN_VALUE].incrementAndGet();
if (old != Entry.NULL) {
- costs[old.cost - Byte.MIN_VALUE]--;
+ costs[old.cost - Byte.MIN_VALUE].decrementAndGet();
if (eviction) {
- evictions[old.cost - Byte.MIN_VALUE]++;
- evictionCount++;
+ evictions[old.cost - Byte.MIN_VALUE].incrementAndGet();
+ evictionCount.incrementAndGet();
}
- weight -= weighEntry(old);
+ weight.addAndGet(-weighEntry(old)) ;
} else {
- size++;
+ size.incrementAndGet();
}
- weight += weighEntry(newE);
+ weight.addAndGet(weighEntry(newE));
+ lockedSegment.unlock();
return true;
- } else {
- loadExceptionCount++;
- return false;
}
+
+ checkNotNull(lockedSegment).unlock();
Review Comment:
In case of exception, `unlock()` will not be called. It's safer to call it in a `finally` clause. The `try` block should start just after `lockedSegment` is defined. The `unlock()` call at line 294 can then be removed.
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/tool/Compact.java:
##########
@@ -164,7 +166,7 @@ public Builder withGCLogInterval(long gcLogInterval) {
/**
* The compactor type to be used by compaction. If not specified it defaults to
- * "diff" compactor
+ * "parallel" compactor
Review Comment:
Should default to `diff`.
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/memory/MemoryStore.java:
##########
@@ -27,17 +27,7 @@
import com.google.common.collect.Maps;
import org.apache.jackrabbit.oak.commons.Buffer;
-import org.apache.jackrabbit.oak.segment.CachingSegmentReader;
-import org.apache.jackrabbit.oak.segment.Revisions;
-import org.apache.jackrabbit.oak.segment.Segment;
-import org.apache.jackrabbit.oak.segment.SegmentId;
-import org.apache.jackrabbit.oak.segment.SegmentIdFactory;
-import org.apache.jackrabbit.oak.segment.SegmentIdProvider;
-import org.apache.jackrabbit.oak.segment.SegmentNotFoundException;
-import org.apache.jackrabbit.oak.segment.SegmentReader;
-import org.apache.jackrabbit.oak.segment.SegmentStore;
-import org.apache.jackrabbit.oak.segment.SegmentTracker;
-import org.apache.jackrabbit.oak.segment.SegmentWriter;
+import org.apache.jackrabbit.oak.segment.*;
Review Comment:
[Code Style] We should not use * imports
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/PriorityCache.java:
##########
@@ -265,22 +309,27 @@ public synchronized boolean put(@NotNull K key, @NotNull V value, int generation
*/
@SuppressWarnings("unchecked")
@Nullable
- public synchronized V get(@NotNull K key, int generation) {
+ public V get(@NotNull K key, int generation) {
int hashCode = key.hashCode();
for (int k = 0; k <= rehash; k++) {
int i = project(hashCode, k);
+ Segment segment = getSegment(i);
+ segment.lock();
Entry<?, ?> entry = entries[i];
if (generation == entry.generation && key.equals(entry.key)) {
if (entry.cost < Byte.MAX_VALUE) {
- costs[entry.cost - Byte.MIN_VALUE]--;
+ costs[entry.cost - Byte.MIN_VALUE].decrementAndGet();
entry.cost++;
- costs[entry.cost - Byte.MIN_VALUE]++;
+ costs[entry.cost - Byte.MIN_VALUE].incrementAndGet();
}
- hitCount++;
- return (V) entry.value;
+ hitCount.incrementAndGet();
+ V value = (V) entry.value;
+ segment.unlock();
+ return value;
}
+ segment.unlock();
Review Comment:
In case of exception, `unlock()` will not be called. It's safer to call it in a `finally` clause. The `try` block should start just after `lock()` is called. The call to `unlock()` at line 327 can be removed.
##########
oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/RecordCacheStatsTest.java:
##########
@@ -37,18 +37,9 @@ public class RecordCacheStatsTest {
private final Random rnd = new Random();
private final MemoryStore store = new MemoryStore();
- private final RecordCache<Integer> cache = newRecordCache(KEYS);
+ private final RecordCache<Integer> cache = newRecordCache(KEYS * 4 / 3);
Review Comment:
I've noticed that if I remove your change and initialise the `RecordCache` with only KEYS (=100) values like before, then some tests fail. This means the behaviour of `RecordCache` is not preserved. This could break other code relying on the previous behaviour.
Can you explain why the stats are not the same? Can you restore the previous behaviour?
##########
oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorTest.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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 java.util.concurrent.TimeUnit.DAYS;
+import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.addTestContent;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.assertSameRecord;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.assertSameStableId;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.checkGeneration;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.getCheckpoint;
+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.File;
+import java.io.IOException;
+
+import org.apache.jackrabbit.oak.segment.file.FileStore;
+import org.apache.jackrabbit.oak.segment.file.InvalidFileStoreVersionException;
+import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
+import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.jetbrains.annotations.NotNull;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public abstract class AbstractCompactorTest {
+ @Rule
+ public TemporaryFolder folder = new TemporaryFolder(new File("target"));
+
+ private FileStore fileStore;
+
+ private SegmentNodeStore nodeStore;
+
+ private Compactor 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);
+ }
+
+ protected abstract Compactor createCompactor(@NotNull FileStore fileStore, @NotNull GCGeneration generation);
+
+ @After
+ public void tearDown() {
+ fileStore.close();
+ }
+
+ @Test
+ public void testCompact() throws Exception {
+ addTestContent("cp1", nodeStore, 42);
+ String cp1 = nodeStore.checkpoint(DAYS.toMillis(1));
+ addTestContent("cp2", nodeStore, 42);
+ String cp2 = nodeStore.checkpoint(DAYS.toMillis(1));
+
+ SegmentNodeState uncompacted1 = fileStore.getHead();
+ SegmentNodeState compacted1 = compactor.compact(EMPTY_NODE, uncompacted1, EMPTY_NODE, Canceller.newCanceller());
+ assertNotNull(compacted1);
+ assertFalse(uncompacted1 == compacted1);
Review Comment:
Can be simplified to:
```java
assertNotSame(uncompacted1, compacted1);
```
##########
oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NodeRecordTest.java:
##########
@@ -105,7 +105,7 @@ public void baseNodeStateShouldBeReusedAcrossGenerations() throws Exception {
SegmentWriter writer = defaultSegmentWriterBuilder("test")
.withGeneration(generation)
- .withWriterPool()
+ .withWriterPool(SegmentBufferWriterPool.PoolType.GLOBAL)
Review Comment:
Given my other comment about keeping `withWriterPool()`, I suggest reverting this changes.
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java:
##########
@@ -146,7 +140,7 @@ private static GarbageCollectionStrategy newGarbageCollectionStrategy() {
this.segmentWriter = defaultSegmentWriterBuilder("sys")
.withGeneration(() -> getGcGeneration().nonGC())
- .withWriterPool()
+ .withWriterPool(SegmentBufferWriterPool.PoolType.GLOBAL)
Review Comment:
Given my other comment about keeping withWriterPool(), I suggest reverting this change.
##########
oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/AbstractCompactorTest.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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 java.util.concurrent.TimeUnit.DAYS;
+import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.addTestContent;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.assertSameRecord;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.assertSameStableId;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.checkGeneration;
+import static org.apache.jackrabbit.oak.segment.CompactorTestUtils.getCheckpoint;
+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.File;
+import java.io.IOException;
+
+import org.apache.jackrabbit.oak.segment.file.FileStore;
+import org.apache.jackrabbit.oak.segment.file.InvalidFileStoreVersionException;
+import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
+import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
+import org.jetbrains.annotations.NotNull;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public abstract class AbstractCompactorTest {
+ @Rule
+ public TemporaryFolder folder = new TemporaryFolder(new File("target"));
+
+ private FileStore fileStore;
+
+ private SegmentNodeStore nodeStore;
+
+ private Compactor 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);
+ }
+
+ protected abstract Compactor createCompactor(@NotNull FileStore fileStore, @NotNull GCGeneration generation);
+
+ @After
+ public void tearDown() {
+ fileStore.close();
+ }
+
+ @Test
+ public void testCompact() throws Exception {
+ addTestContent("cp1", nodeStore, 42);
+ String cp1 = nodeStore.checkpoint(DAYS.toMillis(1));
+ addTestContent("cp2", nodeStore, 42);
+ String cp2 = nodeStore.checkpoint(DAYS.toMillis(1));
+
+ SegmentNodeState uncompacted1 = fileStore.getHead();
+ SegmentNodeState compacted1 = compactor.compact(EMPTY_NODE, uncompacted1, EMPTY_NODE, Canceller.newCanceller());
+ 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, 42);
+ String cp3 = nodeStore.checkpoint(DAYS.toMillis(1));
+ addTestContent("cp4", nodeStore, 42);
+ String cp4 = nodeStore.checkpoint(DAYS.toMillis(1));
+
+ SegmentNodeState uncompacted2 = fileStore.getHead();
+ SegmentNodeState compacted2 = compactor.compact(uncompacted1, uncompacted2, compacted1, Canceller.newCanceller());
+ assertNotNull(compacted2);
+ assertFalse(uncompacted2 == compacted2);
Review Comment:
Can be simplified to:
```java
assertNotSame(uncompacted2, compacted2);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] jelmini commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
jelmini commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r1050639411
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPool.java:
##########
@@ -19,55 +19,35 @@
package org.apache.jackrabbit.oak.segment;
-import static com.google.common.base.Preconditions.checkNotNull;
-import static com.google.common.base.Preconditions.checkState;
-import static com.google.common.collect.Lists.newArrayList;
-import static com.google.common.collect.Maps.newHashMap;
-import static com.google.common.collect.Sets.newHashSet;
-import static java.lang.Thread.currentThread;
+import com.google.common.base.Supplier;
Review Comment:
Can be replaced by `java.util.function.Supplier`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r995757962
##########
oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/RecordCacheStatsTest.java:
##########
@@ -37,18 +37,9 @@ public class RecordCacheStatsTest {
private final Random rnd = new Random();
private final MemoryStore store = new MemoryStore();
- private final RecordCache<Integer> cache = newRecordCache(KEYS);
+ private final RecordCache<Integer> cache = newRecordCache(KEYS * 4 / 3);
Review Comment:
Actually, there is another option. The Guava cache uses a `ConcurrentLinkedHashMap` internally. It would be possible to just use it manually. Downside of this approach is that I would also need to keep track of the `CacheStats` manually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] reschke commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
reschke commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r994982906
##########
oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/RecordCache.java:
##########
@@ -19,29 +19,21 @@
package org.apache.jackrabbit.oak.segment;
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import java.util.LinkedHashMap;
-import java.util.Map;
-
import com.google.common.base.Supplier;
-import com.google.common.cache.CacheStats;
-import com.google.common.cache.Weigher;
-
+import com.google.common.cache.*;
import org.jetbrains.annotations.NotNull;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static com.google.common.base.Preconditions.checkNotNull;
Review Comment:
Actually, we should avoid Guava methods everywhere we can use the JDK.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [jackrabbit-oak] lweitzendorf commented on a diff in pull request #692: OAK-9922 Parallel Compaction
Posted by GitBox <gi...@apache.org>.
lweitzendorf commented on code in PR #692:
URL: https://github.com/apache/jackrabbit-oak/pull/692#discussion_r995408894
##########
oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/RecordCacheStatsTest.java:
##########
@@ -37,18 +37,9 @@ public class RecordCacheStatsTest {
private final Random rnd = new Random();
private final MemoryStore store = new MemoryStore();
- private final RecordCache<Integer> cache = newRecordCache(KEYS);
+ private final RecordCache<Integer> cache = newRecordCache(KEYS * 4 / 3);
Review Comment:
The reason for this change in behavior is that the Guava cache may preemptively evict items for performance reasons. This means that filling the cache exactly to capacity no longer results in an eviction count of 0. I cannot change this behavior. What I can do is keep both implementations like with the SegmentBufferWriterPool and only use the concurrent cache for compaction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org