You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by th...@apache.org on 2024/09/11 08:03:05 UTC
(jackrabbit-oak) 01/01: OAK-11108 Tree store: support parallel indexing
This is an automated email from the ASF dual-hosted git repository.
thomasm pushed a commit to branch OAK-11108
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
commit a02a30c9b786843f46257cfe33d8a61d85aec0f3
Author: Thomas Mueller <th...@apache.org>
AuthorDate: Wed Sep 11 10:02:48 2024 +0200
OAK-11108 Tree store: support parallel indexing
---
.../index/lucene/writer/DefaultIndexWriter.java | 9 +-
.../index/lucene/writer/IndexWriterUtils.java | 8 +-
.../lucene/writer/LuceneIndexWriterConfig.java | 11 +++
.../indexer/document/DocumentStoreIndexerBase.java | 29 ++++--
.../indexer/document/IndexerStatisticsTracker.java | 61 ++++++------
.../flatfile/FlatFileNodeStoreBuilder.java | 1 +
.../indexer/document/tree/ParallelIndexStore.java | 33 +++++++
.../indexer/document/tree/ParallelTreeStore.java | 103 +++++++++++++++++++++
.../index/indexer/document/tree/Prefetcher.java | 21 ++++-
.../oak/index/indexer/document/tree/TreeStore.java | 92 +++++++++++++++++-
.../indexer/document/tree/store/TreeSession.java | 25 +++--
.../tree/store/ParallelIndexStoreTest.java | 102 ++++++++++++++++++++
.../document/tree/store/RandomizedTest.java | 5 +-
.../oak/index/indexer/document/ElasticIndexer.java | 6 +-
.../jackrabbit/oak/index/LuceneIndexHelper.java | 3 +-
.../oak/index/indexer/document/LuceneIndexer.java | 8 +-
16 files changed, 446 insertions(+), 71 deletions(-)
diff --git a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/DefaultIndexWriter.java b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/DefaultIndexWriter.java
index 47b67e0184..af0e38eb8f 100644
--- a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/DefaultIndexWriter.java
+++ b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/DefaultIndexWriter.java
@@ -169,7 +169,14 @@ class DefaultIndexWriter implements LuceneIndexWriter {
if (localRefWriter == null) {
final long start = PERF_LOGGER.start();
directory = directoryFactory.newInstance(definition, definitionBuilder, dirName, reindex);
- IndexWriterConfig config = getIndexWriterConfig(definition, directoryFactory.remoteDirectory(), writerConfig);
+ boolean serialScheduler = directoryFactory.remoteDirectory();
+ if (writerConfig.getThreadCount() > 1) {
+ // for multi-threaded indexing (using oak-run), use the serial scheduler,
+ // to avoid concurrency issues with the concurrent merge policy
+ log.info("Using the serial scheduler for parallel indexing");
+ serialScheduler = true;
+ }
+ IndexWriterConfig config = getIndexWriterConfig(definition, serialScheduler, writerConfig);
config.setMergePolicy(definition.getMergePolicy());
writer = localRefWriter = new IndexWriter(directory, config);
genAtStart = getLatestGeneration(directory);
diff --git a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterUtils.java b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterUtils.java
index 4ef0e07dfb..ba70e0043f 100644
--- a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterUtils.java
+++ b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterUtils.java
@@ -43,11 +43,11 @@ public class IndexWriterUtils {
private static final int INDEX_WRITER_MAX_MERGE = Integer.getInteger("oak.indexer.writerMaxMerges", 1);
private static final int INDEX_WRITER_MAX_THREAD = Integer.getInteger("oak.indexer.writerMaxThreads", 1);
- public static IndexWriterConfig getIndexWriterConfig(LuceneIndexDefinition definition, boolean remoteDir){
- return getIndexWriterConfig(definition, remoteDir, new LuceneIndexWriterConfig());
+ public static IndexWriterConfig getIndexWriterConfig(LuceneIndexDefinition definition, boolean serialScheduler) {
+ return getIndexWriterConfig(definition, serialScheduler, new LuceneIndexWriterConfig());
}
- public static IndexWriterConfig getIndexWriterConfig(LuceneIndexDefinition definition, boolean remoteDir,
+ public static IndexWriterConfig getIndexWriterConfig(LuceneIndexDefinition definition, boolean serialScheduler,
LuceneIndexWriterConfig writerConfig) {
// FIXME: Hack needed to make Lucene work in an OSGi environment
Thread thread = Thread.currentThread();
@@ -71,7 +71,7 @@ public class IndexWriterUtils {
}
Analyzer analyzer = new PerFieldAnalyzerWrapper(definitionAnalyzer, analyzers);
IndexWriterConfig config = new IndexWriterConfig(VERSION, analyzer);
- if (remoteDir) {
+ if (serialScheduler) {
config.setMergeScheduler(new SerialMergeScheduler());
} else {
ConcurrentMergeScheduler concurrentMergeScheduler = new ConcurrentMergeScheduler();
diff --git a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/LuceneIndexWriterConfig.java b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/LuceneIndexWriterConfig.java
index e474e6624c..a64a8d57d6 100644
--- a/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/LuceneIndexWriterConfig.java
+++ b/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/LuceneIndexWriterConfig.java
@@ -39,6 +39,7 @@ public class LuceneIndexWriterConfig {
IndexWriterConfig.DISABLE_AUTO_FLUSH);
private final int ramPerThreadHardLimitMB = Integer.getInteger(RAM_PER_THREAD_HARD_LIMIT_MB_KEY,
IndexWriterConfig.DEFAULT_RAM_PER_THREAD_HARD_LIMIT_MB);
+ private final int threadCount;
public LuceneIndexWriterConfig() {
this(IndexWriterConfig.DEFAULT_RAM_BUFFER_SIZE_MB);
@@ -46,6 +47,12 @@ public class LuceneIndexWriterConfig {
public LuceneIndexWriterConfig(double ramBufferSizeMB) {
this.ramBufferSizeMB = ramBufferSizeMB;
+ this.threadCount = 1;
+ }
+
+ public LuceneIndexWriterConfig(double ramBufferSizeMB, int threadCount) {
+ this.ramBufferSizeMB = ramBufferSizeMB;
+ this.threadCount = threadCount;
}
public double getRamBufferSizeMB() {
@@ -59,4 +66,8 @@ public class LuceneIndexWriterConfig {
public int getRamPerThreadHardLimitMB() {
return ramPerThreadHardLimitMB;
}
+
+ public int getThreadCount() {
+ return threadCount;
+ }
}
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java
index 4add3d822b..9b30077aa3 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java
@@ -33,6 +33,7 @@ import org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeSto
import org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.ConfigHelper;
import org.apache.jackrabbit.oak.index.indexer.document.incrementalstore.IncrementalStoreBuilder;
import org.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStore;
+import org.apache.jackrabbit.oak.index.indexer.document.tree.ParallelIndexStore;
import org.apache.jackrabbit.oak.plugins.document.DocumentNodeState;
import org.apache.jackrabbit.oak.plugins.document.DocumentNodeStore;
import org.apache.jackrabbit.oak.plugins.document.RevisionVector;
@@ -173,7 +174,7 @@ public abstract class DocumentStoreIndexerBase implements Closeable {
IndexingReporter reporter) throws IOException {
List<IndexStore> storeList = new ArrayList<>();
- Stopwatch flatFileStoreWatch = Stopwatch.createStarted();
+ Stopwatch indexStoreWatch = Stopwatch.createStarted();
int executionCount = 1;
CompositeException lastException = null;
List<File> previousDownloadDirs = new ArrayList<>();
@@ -209,15 +210,24 @@ public abstract class DocumentStoreIndexerBase implements Closeable {
if (splitFlatFile) {
storeList = builder.buildList(indexHelper, indexerSupport, indexDefinitions);
} else {
- storeList.add(builder.build(indexHelper, indexer));
+ log.info("Building index store");
+ IndexStore store = builder.build(indexHelper, indexer);
+ int threads = IndexerConfiguration.indexThreadPoolSize();
+ if (store instanceof ParallelIndexStore && threads > 1) {
+ log.info("Indexing with {} threads", threads);
+ ParallelIndexStore ps = (ParallelIndexStore) store;
+ storeList.addAll(ps.buildParallelStores(threads));
+ } else {
+ storeList.add(store);
+ }
}
for (IndexStore item : storeList) {
closer.register(item);
}
} catch (CompositeException e) {
e.logAllExceptions("Underlying throwable caught during download", log);
- log.info("Could not build flat file store. Execution count {}. Retries left {}. Time elapsed {}",
- executionCount, MAX_DOWNLOAD_ATTEMPTS - executionCount, flatFileStoreWatch);
+ log.info("Could not build index store. Execution count {}. Retries left {}. Time elapsed {}",
+ executionCount, MAX_DOWNLOAD_ATTEMPTS - executionCount, indexStoreWatch);
lastException = e;
previousDownloadDirs.add(builder.getFlatFileStoreDir());
if (executionCount < MAX_DOWNLOAD_ATTEMPTS) {
@@ -233,9 +243,9 @@ public abstract class DocumentStoreIndexerBase implements Closeable {
executionCount++;
}
if (storeList.isEmpty()) {
- throw new IOException("Could not build flat file store", lastException);
+ throw new IOException("Could not build index store", lastException);
}
- log.info("Completed the flat file store build in {}", flatFileStoreWatch);
+ log.info("Completed the index store build in {}", indexStoreWatch);
return storeList;
}
@@ -432,10 +442,11 @@ public abstract class DocumentStoreIndexerBase implements Closeable {
private void indexParallel(List<IndexStore> storeList, CompositeIndexer indexer, IndexingProgressReporter progressReporter)
throws IOException {
ExecutorService service = Executors.newFixedThreadPool(IndexerConfiguration.indexThreadPoolSize());
- List<Future> futureList = new ArrayList<>();
+ List<Future<Boolean>> futureList = new ArrayList<>();
+ indexer.onIndexingStarting();
for (IndexStore item : storeList) {
- Future future = service.submit(() -> {
+ Future<Boolean> future = service.submit(() -> {
for (NodeStateEntry entry : item) {
reportDocumentRead(entry.getPath(), progressReporter);
log.trace("Indexing : {}", entry.getPath());
@@ -447,7 +458,7 @@ public abstract class DocumentStoreIndexerBase implements Closeable {
}
try {
- for (Future future : futureList) {
+ for (Future<Boolean> future : futureList) {
future.get();
}
log.info("All {} indexing jobs are done", storeList.size());
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/IndexerStatisticsTracker.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/IndexerStatisticsTracker.java
index 4508e1d2d5..c4ede412fe 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/IndexerStatisticsTracker.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/IndexerStatisticsTracker.java
@@ -18,6 +18,8 @@
*/
package org.apache.jackrabbit.oak.index.indexer.document;
+import java.util.concurrent.atomic.AtomicLong;
+
import org.apache.jackrabbit.oak.plugins.index.FormattingUtils;
import org.slf4j.Logger;
@@ -29,17 +31,12 @@ public final class IndexerStatisticsTracker {
// Timestamp of when indexing started.
private long startIndexingNanos = 0;
// Time spent indexing entries. Should be almost the same as totalMakeDocumentTimeNanos+totalWriteTimeNanos
- private long totalIndexingTimeNanos = 0;
+ private AtomicLong totalIndexingTimeNanos = new AtomicLong();
// Time generating the Lucene document.
- private long totalMakeDocumentTimeNanos = 0;
+ private AtomicLong totalMakeDocumentTimeNanos = new AtomicLong();
// Time writing the Lucene document to disk.
- private long totalWriteTimeNanos = 0;
- private long nodesIndexed = 0;
-
- // Timestamp of when the current entry started being indexed
- private long startEntryNanos = 0;
- // Timestamp of when the current entry finished the makeDocument phase.
- private long endEntryMakeDocumentNanos = 0;
+ private AtomicLong totalWriteTimeNanos = new AtomicLong();
+ private AtomicLong nodesIndexed = new AtomicLong();
public IndexerStatisticsTracker(Logger logger) {
this.logger = logger;
@@ -49,42 +46,42 @@ public final class IndexerStatisticsTracker {
this.startIndexingNanos = System.nanoTime();
}
- public void onEntryStart() {
- startEntryNanos = System.nanoTime();
- }
-
- public void onEntryEndMakeDocument() {
- endEntryMakeDocumentNanos = System.nanoTime();
- }
-
- public void onEntryEnd(String entryPath) {
+ /**
+ * Mark that an entry completed indexing. If indexing the entry was slow, then a
+ * message is logged.
+ *
+ * @param path the path
+ * @param startEntryNanos timestamp of when the current entry started being
+ * indexed
+ * @oaran endEntryMakeDocumentNanos timestamp of when the current entry finished
+ * the makeDocument phase.
+ */
+ public void onEntryEnd(String entryPath, long startEntryNanos, long endEntryMakeDocumentNanos) {
long endEntryWriteNanos = System.nanoTime();
- nodesIndexed++;
+ nodesIndexed.incrementAndGet();
long entryIndexingTimeNanos = endEntryWriteNanos - startEntryNanos;
long entryMakeDocumentTimeNanos = endEntryMakeDocumentNanos - startEntryNanos;
long entryWriteTimeNanos = endEntryWriteNanos - endEntryMakeDocumentNanos;
- totalIndexingTimeNanos += entryIndexingTimeNanos;
- totalMakeDocumentTimeNanos += entryMakeDocumentTimeNanos;
- totalWriteTimeNanos += entryWriteTimeNanos;
+ totalIndexingTimeNanos.addAndGet(entryIndexingTimeNanos);
+ totalMakeDocumentTimeNanos.addAndGet(entryWriteTimeNanos);
+ totalWriteTimeNanos.addAndGet(entryWriteTimeNanos);
if (entryIndexingTimeNanos >= (long) SLOW_DOCUMENT_LOG_THRESHOLD * 1_000_000) {
logger.info("Slow document: {}. Times: total={}ms, makeDocument={}ms, writeToIndex={}ms",
entryPath, entryIndexingTimeNanos / 1_000_000, entryMakeDocumentTimeNanos / 1_000_000, entryWriteTimeNanos / 1_000_000);
}
- startEntryNanos = 0;
- endEntryMakeDocumentNanos = 0;
}
public String formatStats() {
long endTimeNanos = System.nanoTime();
long totalTimeNanos = endTimeNanos - startIndexingNanos;
- long avgTimePerDocumentMicros = Math.round(FormattingUtils.safeComputeAverage(totalIndexingTimeNanos / 1000, nodesIndexed));
- double percentageIndexing = FormattingUtils.safeComputePercentage(totalIndexingTimeNanos, totalTimeNanos);
- double percentageMakingDocument = FormattingUtils.safeComputePercentage(totalMakeDocumentTimeNanos, totalIndexingTimeNanos);
- double percentageWritingToIndex = FormattingUtils.safeComputePercentage(totalWriteTimeNanos, totalIndexingTimeNanos);
+ long avgTimePerDocumentMicros = Math.round(FormattingUtils.safeComputeAverage(totalIndexingTimeNanos.get() / 1000, nodesIndexed.get()));
+ double percentageIndexing = FormattingUtils.safeComputePercentage(totalIndexingTimeNanos.get(), totalTimeNanos);
+ double percentageMakingDocument = FormattingUtils.safeComputePercentage(totalMakeDocumentTimeNanos.get(), totalIndexingTimeNanos.get());
+ double percentageWritingToIndex = FormattingUtils.safeComputePercentage(totalWriteTimeNanos.get(), totalIndexingTimeNanos.get());
return String.format("Indexed %d nodes in %s. Avg per node: %d microseconds. indexingTime: %s (%2.1f%% of total time). Breakup of indexing time: makeDocument: %s (%2.1f%%), writeIndex: %s (%2.1f%%)",
- nodesIndexed, FormattingUtils.formatNanosToSeconds(totalTimeNanos), avgTimePerDocumentMicros,
- FormattingUtils.formatNanosToSeconds(totalIndexingTimeNanos), percentageIndexing,
- FormattingUtils.formatNanosToSeconds(totalMakeDocumentTimeNanos), percentageMakingDocument,
- FormattingUtils.formatNanosToSeconds(totalWriteTimeNanos), percentageWritingToIndex);
+ nodesIndexed.get(), FormattingUtils.formatNanosToSeconds(totalTimeNanos), avgTimePerDocumentMicros,
+ FormattingUtils.formatNanosToSeconds(totalIndexingTimeNanos.get()), percentageIndexing,
+ FormattingUtils.formatNanosToSeconds(totalMakeDocumentTimeNanos.get()), percentageMakingDocument,
+ FormattingUtils.formatNanosToSeconds(totalWriteTimeNanos.get()), percentageWritingToIndex);
}
}
\ No newline at end of file
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java
index f84aefd933..a64cd628a7 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java
@@ -278,6 +278,7 @@ public class FlatFileNodeStoreBuilder {
}
prefetcher.setBlobSuffix(blobSuffix);
prefetcher.startPrefetch();
+ indexingTreeStore.setPrefetcher(prefetcher);
return indexingTreeStore;
}
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/ParallelIndexStore.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/ParallelIndexStore.java
new file mode 100644
index 0000000000..27b4a4bbf8
--- /dev/null
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/ParallelIndexStore.java
@@ -0,0 +1,33 @@
+/*
+ * 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.index.indexer.document.tree;
+
+import java.util.Collection;
+
+import org.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStore;
+
+/**
+ * A parallel store has the ability to "split itself" into multiple stores that
+ * each cover a subset of the content. This allows to index in parallel.
+ */
+public interface ParallelIndexStore extends IndexStore {
+
+ Collection<IndexStore> buildParallelStores(int parallelIndexingThreads);
+
+}
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/ParallelTreeStore.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/ParallelTreeStore.java
new file mode 100644
index 0000000000..9e9bb7aa69
--- /dev/null
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/ParallelTreeStore.java
@@ -0,0 +1,103 @@
+/*
+ * 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.index.indexer.document.tree;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntry;
+import org.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStore;
+
+/**
+ * A wrapper around the tree store that only iterates over a subset of the
+ * nodes. Each parallel tree store for reads the next block (a block is a tiny
+ * subset of the nodes). Once it has finished iterating over the block, it asks
+ * the backend for the next block. The result is that the whole range is
+ * covered, but it doesn't matter much which thread is how fast.
+ */
+public class ParallelTreeStore implements IndexStore {
+
+ private final TreeStore backend;
+ private final AtomicInteger openCount;
+
+ ParallelTreeStore(TreeStore backend, AtomicInteger openCount) {
+ this.backend = backend;
+ this.openCount = openCount;
+ }
+
+ @Override
+ public Iterator<NodeStateEntry> iterator() {
+ return new Iterator<NodeStateEntry>() {
+
+ private Iterator<NodeStateEntry> currentIterator = backend.nextSubsetIterator();
+
+ @Override
+ public boolean hasNext() {
+ while (currentIterator != null && !currentIterator.hasNext()) {
+ currentIterator = backend.nextSubsetIterator();
+ }
+ if (currentIterator == null) {
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public NodeStateEntry next() {
+ hasNext();
+ return currentIterator.next();
+ }
+
+ };
+ }
+
+ @Override
+ public String getStorePath() {
+ return backend.getStorePath();
+ }
+
+ @Override
+ public long getEntryCount() {
+ return backend.getEntryCount();
+ }
+
+ @Override
+ public void setEntryCount(long entryCount) {
+ backend.setEntryCount(entryCount);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (openCount.decrementAndGet() == 0) {
+ backend.close();
+ }
+ }
+
+ @Override
+ public String getIndexStoreType() {
+ return backend.getIndexStoreType();
+ }
+
+ @Override
+ public boolean isIncremental() {
+ return backend.isIncremental();
+ }
+
+}
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/Prefetcher.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/Prefetcher.java
index 779227244b..53ff615c49 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/Prefetcher.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/Prefetcher.java
@@ -26,6 +26,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -59,6 +60,7 @@ public class Prefetcher {
private final AtomicLong downloadMax = new AtomicLong();
private final AtomicLong iterateCount = new AtomicLong();
private final Semaphore semaphore = new Semaphore(PRETCH_THREADS);
+ private final AtomicBoolean closed = new AtomicBoolean();
private String blobSuffix;
@@ -93,9 +95,15 @@ public class Prefetcher {
this.nodeReadAheadCount = nodeReadAheadCount;
}
- public boolean shutdown() throws InterruptedException {
+ public boolean shutdown() {
+ closed.set(true);
executorService.shutdown();
- return executorService.awaitTermination(1, TimeUnit.SECONDS);
+ try {
+ return executorService.awaitTermination(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.info("Prefetcher termination was interrupted: {}", e.toString());
+ return false;
+ }
}
public void startPrefetch() {
@@ -126,6 +134,9 @@ public class Prefetcher {
long inlinedBlobCount = 0;
long totalBlobSize = 0;
while (it.hasNext()) {
+ if (closed.get()) {
+ break;
+ }
String path = it.next();
if (++count % 1_000_000 == 0) {
int available = semaphore.availablePermits();
@@ -183,9 +194,9 @@ public class Prefetcher {
}
String indexingPos = indexStore.getHighestReadKey();
if (indexingPos.compareTo(path) >= 0) {
- // we are behind indexing!
- // do not download, in order to catch up
- LOG.info("Indexing is ahead; ignoring {}", path);
+ // we are behind indexing, so skip downloading,
+ // in order to catch up
+ // (this can happen eg. when indexing in parallel)
continue;
}
while (totalBlobSize - blobReadAheadSize > downloadMax.get()) {
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/TreeStore.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/TreeStore.java
index 50084c0f2e..6d379d6f59 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/TreeStore.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/TreeStore.java
@@ -20,11 +20,15 @@ package org.apache.jackrabbit.oak.index.indexer.document.tree;
import java.io.File;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
+import java.util.Random;
import java.util.Set;
import java.util.SortedSet;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.jackrabbit.oak.commons.PathUtils;
@@ -50,7 +54,7 @@ import org.slf4j.LoggerFactory;
* key-value pairs in a single file, it stores the entries in multiple files
* (except if there are very few nodes).
*/
-public class TreeStore implements IndexStore {
+public class TreeStore implements ParallelIndexStore {
private static final Logger LOG = LoggerFactory.getLogger(TreeStore.class);
@@ -65,6 +69,8 @@ public class TreeStore implements IndexStore {
private static final long MAX_FILE_SIZE_MB = 4;
private static final long MB = 1024 * 1024;
+ private static final int SPLIT_POINT_COUNT = 512;
+
private final String name;
private final Store store;
private final long cacheSizeTreeStoreMB;
@@ -80,6 +86,10 @@ public class TreeStore implements IndexStore {
private final AtomicLong nodeCacheFills = new AtomicLong();
private int iterationCount;
private PathIteratorFilter filter = new PathIteratorFilter();
+ private ArrayList<String> splitPoints;
+
+ // the prefetcher, if any. we keep a references so we can shut it down on close
+ private Prefetcher prefetcher;
public TreeStore(String name, File fileOrDirectory, NodeStateEntryReader entryReader, long cacheSizeFactor) {
this.name = name;
@@ -138,7 +148,11 @@ public class TreeStore implements IndexStore {
}
public Iterator<String> iteratorOverPaths() {
- final Iterator<Entry<String, String>> firstIterator = session.iterator();
+ return iteratorOverPaths(null, null);
+ }
+
+ private Iterator<String> iteratorOverPaths(String start, String end) {
+ final Iterator<Entry<String, String>> firstIterator = session.iterator(start);
return new Iterator<String>() {
Iterator<Entry<String, String>> it = firstIterator;
@@ -152,6 +166,9 @@ public class TreeStore implements IndexStore {
while (it.hasNext()) {
Entry<String, String> e = it.next();
String key = e.getKey();
+ if (end != null && key.compareTo(end) > 0) {
+ break;
+ }
String value = e.getValue();
// if the value is empty (not null!) this is a child node reference,
// without node data
@@ -204,7 +221,11 @@ public class TreeStore implements IndexStore {
@Override
public Iterator<NodeStateEntry> iterator() {
- Iterator<String> it = iteratorOverPaths();
+ return iterator(null, null);
+ }
+
+ private Iterator<NodeStateEntry> iterator(String start, String end) {
+ Iterator<String> it = iteratorOverPaths(start, end);
return new Iterator<NodeStateEntry>() {
NodeStateEntry current;
@@ -234,8 +255,12 @@ public class TreeStore implements IndexStore {
@Override
public void close() throws IOException {
+ LOG.info("Closing " + toString());
session.flush();
store.close();
+ if (prefetcher != null) {
+ prefetcher.shutdown();
+ }
}
public String getHighestReadKey() {
@@ -385,4 +410,65 @@ public class TreeStore implements IndexStore {
return false;
}
+ @Override
+ public Collection<IndexStore> buildParallelStores(int pieces) {
+ LOG.info("Building in {} pieces", pieces);
+ this.splitPoints = new ArrayList<>();
+ splitPoints.add(session.getMinKey());
+ splitPoints.add(session.getMaxKey());
+ Random r = new Random(1);
+ // split in a loop, doubling the number of entries each time
+ while (splitPoints.size() < SPLIT_POINT_COUNT) {
+ ArrayList<String> nextSplitPoints = new ArrayList<>();
+ for (int i = 1; i < splitPoints.size(); i++) {
+ String min = splitPoints.get(i - 1);
+ String max = splitPoints.get(i);
+ String mid = session.getApproximateMedianKey(min, max, r);
+ if (mid.compareTo(min) < 0) {
+ throw new IllegalStateException();
+ }
+ if (mid.compareTo(max) > 0) {
+ throw new IllegalStateException();
+ }
+ nextSplitPoints.add(min);
+ nextSplitPoints.add(mid);
+ }
+ nextSplitPoints.add(splitPoints.get(splitPoints.size() - 1));
+ splitPoints = nextSplitPoints;
+ }
+ // remove duplicates (in case there are few entries)
+ for (int i = 1; i < splitPoints.size(); i++) {
+ String prev = splitPoints.get(i - 1);
+ String current = splitPoints.get(i);
+ if (prev.equals(current)) {
+ splitPoints.remove(i);
+ i--;
+ } else if (prev.compareTo(current) > 0) {
+ throw new IllegalStateException();
+ }
+ }
+ splitPoints.add(0, null);
+ splitPoints.add(null);
+ ArrayList<IndexStore> result = new ArrayList<>();
+ AtomicInteger openCount = new AtomicInteger(pieces);
+ for (int i = 0; i < pieces; i++) {
+ result.add(new ParallelTreeStore(this, openCount));
+ }
+ return result;
+ }
+
+ public synchronized Iterator<NodeStateEntry> nextSubsetIterator() {
+ if (splitPoints.size() < 2) {
+ return null;
+ }
+ String start = splitPoints.get(0);
+ String end = splitPoints.get(1);
+ splitPoints.remove(0);
+ return iterator(start, end);
+ }
+
+ public void setPrefetcher(Prefetcher prefetcher) {
+ this.prefetcher = prefetcher;
+ }
+
}
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/TreeSession.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/TreeSession.java
index a1ecef2bcc..55d3dbfd5c 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/TreeSession.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/TreeSession.java
@@ -25,6 +25,7 @@ import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.PriorityQueue;
import java.util.Properties;
+import java.util.Random;
import org.apache.jackrabbit.oak.index.indexer.document.tree.store.utils.ConcurrentLRUCache;
import org.apache.jackrabbit.oak.index.indexer.document.tree.store.utils.Position;
@@ -881,16 +882,16 @@ public class TreeSession {
}
}
- public String getApproximateMedianKey(String low, String high) {
+ public String getApproximateMedianKey(String low, String high, Random r) {
if (getFile(ROOT_NAME).getNextRoot() != null) {
throw new UnsupportedOperationException("Not fully merged");
}
String fileName = ROOT_NAME;
+ if (low.compareTo(high) >= 0) {
+ return low;
+ }
while (true) {
PageFile file = getFile(fileName);
- if (!file.isInnerNode()) {
- return file.getKey(0);
- }
int i1 = file.getKeyIndex(low);
int i2 = file.getKeyIndex(high);
if (i1 < 0) {
@@ -899,11 +900,21 @@ public class TreeSession {
if (i2 < 0) {
i2 = -i2 - 1;
}
- if (i2 != i1) {
- int middle = (i1 + i2) / 2;
+ int middle = Math.min(file.getValueCount() - 1, (i1 + i2) / 2);
+ if (middle != i1 || !file.isInnerNode()) {
return file.getKey(middle);
}
- fileName = file.getChildValue(i1);
+ if (middle >= file.getValueCount() - 1) {
+ fileName = file.getChildValue(middle);
+ } else if (high.compareTo(file.getKey(middle)) <= 0) {
+ fileName = file.getChildValue(middle);
+ } else if (low.compareTo(file.getKey(middle)) >= 0) {
+ fileName = file.getChildValue(middle + 1);
+ } else if (r.nextInt() > 0) {
+ fileName = file.getChildValue(middle);
+ } else {
+ fileName = file.getChildValue(middle + 1);
+ }
}
}
diff --git a/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/ParallelIndexStoreTest.java b/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/ParallelIndexStoreTest.java
new file mode 100644
index 0000000000..93027f2fb6
--- /dev/null
+++ b/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/ParallelIndexStoreTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.index.indexer.document.tree.store;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+
+import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntry;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryReader;
+import org.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStore;
+import org.apache.jackrabbit.oak.index.indexer.document.tree.TreeStore;
+import org.apache.jackrabbit.oak.spi.blob.MemoryBlobStore;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class ParallelIndexStoreTest {
+
+ @ClassRule
+ public static TemporaryFolder temporaryFolder = new TemporaryFolder(new File("target"));
+
+ @Test
+ public void test() throws IOException {
+ File dir = temporaryFolder.newFolder("sourceDir");
+ Store store = StoreBuilder.build(
+ "type=file\n" +
+ "dir="+dir.getAbsolutePath()+"\n" +
+ "cacheSizeMB=1\n" +
+ "maxFileSizeBytes=100");
+ store.setWriteCompression(Compression.LZ4);
+ store.removeAll();
+ TreeSession session = new TreeSession(store);
+ session.init();
+ int count = 1000;
+ session.checkpoint();
+ HashSet<String> paths = new HashSet<>();
+ for (int i = 0; i < count; i++) {
+ String path = "/c" + (i / 100) + "/d" + i;
+ session.put(path, "{\"x\":" + i + "}");
+ paths.add(path);
+ }
+ session.mergeRoots(Integer.MAX_VALUE);
+ session.flush();
+ store.close();
+
+ TreeStore treeStore = new TreeStore("test", dir, new NodeStateEntryReader(new MemoryBlobStore()), 1);
+ Collection<IndexStore> list = treeStore.buildParallelStores(10);
+ ArrayList<Iterator<NodeStateEntry>> iterators = new ArrayList<>();
+ for(IndexStore ps : list) {
+ iterators.add(ps.iterator());
+ }
+ int found = 0;
+ int[] countPerIterator = new int[10];
+ while (true) {
+ boolean oneIteratorHasNext = false;
+ for (int i = 0; i < 10; i++) {
+ Iterator<NodeStateEntry> it = iterators.get(i);
+ if (it.hasNext()) {
+ String path = it.next().getPath();
+ countPerIterator[i]++;
+ assertTrue(paths.remove(path));
+ oneIteratorHasNext = true;
+ found++;
+ }
+ }
+ if (!oneIteratorHasNext) {
+ break;
+ }
+ }
+ for (int i = 0; i < 10; i++) {
+ assertTrue(Arrays.toString(countPerIterator),
+ countPerIterator[i] > 50);
+ }
+ assertEquals(count, found);
+ treeStore.close();
+ }
+
+}
diff --git a/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/RandomizedTest.java b/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/RandomizedTest.java
index 67e92056aa..fc9d2a8e36 100644
--- a/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/RandomizedTest.java
+++ b/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/tree/store/RandomizedTest.java
@@ -71,8 +71,9 @@ public class RandomizedTest {
assertEquals("0", min);
String max = session.getMaxKey();
assertEquals("99", max);
- String median = session.getApproximateMedianKey(min, max);
- assertEquals("53", median);
+ Random r1 = new Random(42);
+ String median = session.getApproximateMedianKey(min, max, r1);
+ assertEquals("85", median);
}
private void verify(HashMap<String, String> verify, TreeSession session, String key) {
diff --git a/oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/ElasticIndexer.java b/oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/ElasticIndexer.java
index cb881a84e0..eee27d6e6b 100644
--- a/oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/ElasticIndexer.java
+++ b/oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/ElasticIndexer.java
@@ -102,14 +102,14 @@ public class ElasticIndexer implements NodeStateIndexer {
if (indexingRule == null) {
return false;
}
- indexerStatisticsTracker.onEntryStart();
+ long startEntryNanos = System.nanoTime();
ElasticDocumentMaker maker = newDocumentMaker(indexingRule, entry.getPath());
ElasticDocument doc = maker.makeDocument(entry.getNodeState());
- indexerStatisticsTracker.onEntryEndMakeDocument();
+ long endEntryMakeDocumentNanos = System.nanoTime();
if (doc != null) {
writeToIndex(doc, entry.getPath());
- indexerStatisticsTracker.onEntryEnd(entry.getPath());
+ indexerStatisticsTracker.onEntryEnd(entry.getPath(), startEntryNanos, endEntryMakeDocumentNanos);
progressReporter.indexUpdate(definition.getIndexPath());
return true;
}
diff --git a/oak-run/src/main/java/org/apache/jackrabbit/oak/index/LuceneIndexHelper.java b/oak-run/src/main/java/org/apache/jackrabbit/oak/index/LuceneIndexHelper.java
index 92e313f7ae..98dd05e194 100644
--- a/oak-run/src/main/java/org/apache/jackrabbit/oak/index/LuceneIndexHelper.java
+++ b/oak-run/src/main/java/org/apache/jackrabbit/oak/index/LuceneIndexHelper.java
@@ -23,6 +23,7 @@ import java.io.Closeable;
import java.io.File;
import java.io.IOException;
+import org.apache.jackrabbit.oak.index.indexer.document.IndexerConfiguration;
import org.apache.jackrabbit.oak.plugins.index.lucene.IndexCopier;
import org.apache.jackrabbit.oak.plugins.index.lucene.IndexCopier.COWDirectoryTracker;
import org.apache.jackrabbit.oak.plugins.index.lucene.LuceneIndexEditorProvider;
@@ -81,7 +82,7 @@ public class LuceneIndexHelper implements Closeable {
log.info("Setting RAMBufferSize for LuceneIndexWriter (configurable via " +
"system property '{}') to {} MB", PROP_BUFFER_SIZE, buffSize);
- return new LuceneIndexWriterConfig(buffSize);
+ return new LuceneIndexWriterConfig(buffSize, IndexerConfiguration.indexThreadPoolSize());
}
public void setDirectoryFactory(DirectoryFactory directoryFactory) {
diff --git a/oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/LuceneIndexer.java b/oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/LuceneIndexer.java
index 22ea5f97fd..ba8cfb9eb0 100644
--- a/oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/LuceneIndexer.java
+++ b/oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/LuceneIndexer.java
@@ -89,14 +89,14 @@ public class LuceneIndexer implements NodeStateIndexer, FacetsConfigProvider {
return false;
}
- indexerStatisticsTracker.onEntryStart();
-
+ long startEntryNanos = System.nanoTime();
LuceneDocumentMaker maker = newDocumentMaker(indexingRule, entry.getPath());
Document doc = maker.makeDocument(entry.getNodeState());
- indexerStatisticsTracker.onEntryEndMakeDocument();
+ long endEntryMakeDocumentNanos = System.nanoTime();
+
if (doc != null) {
writeToIndex(doc, entry.getPath());
- indexerStatisticsTracker.onEntryEnd(entry.getPath());
+ indexerStatisticsTracker.onEntryEnd(entry.getPath(), startEntryNanos, endEntryMakeDocumentNanos);
progressReporter.indexUpdate(definition.getIndexPath());
return true;
}