You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-dev@jackrabbit.apache.org by "nfsantos (via GitHub)" <gi...@apache.org> on 2025/01/28 17:19:06 UTC
[PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
nfsantos opened a new pull request, #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031
(no 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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "fabriziofortino (via GitHub)" <gi...@apache.org>.
fabriziofortino commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1957861002
##########
oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/DefaultIndexWriterFactory.java:
##########
@@ -50,14 +64,30 @@ public LuceneIndexWriter newInstance(IndexDefinition def, NodeBuilder definition
"Expected %s but found %s for index definition",
LuceneIndexDefinition.class, def.getClass());
- LuceneIndexDefinition definition = (LuceneIndexDefinition)def;
+ LuceneIndexDefinition definition = (LuceneIndexDefinition) def;
- if (mountInfoProvider.hasNonDefaultMounts()){
- return new MultiplexingIndexWriter(directoryFactory, mountInfoProvider, definition,
- definitionBuilder, reindex, writerConfig);
+ if (mountInfoProvider.hasNonDefaultMounts()) {
+ return wrapWithPipelinedIndexWriter(
+ new MultiplexingIndexWriter(directoryFactory, mountInfoProvider, definition, definitionBuilder, reindex, writerConfig),
+ definition.getIndexName());
}
- return new DefaultIndexWriter(definition, definitionBuilder, directoryFactory,
+ DefaultIndexWriter writer = new DefaultIndexWriter(definition, definitionBuilder, directoryFactory,
FulltextIndexConstants.INDEX_DATA_CHILD_NAME,
LuceneIndexConstants.SUGGEST_DATA_CHILD_NAME, reindex, writerConfig);
+
+ return wrapWithPipelinedIndexWriter(writer, definition.getIndexName());
+ }
+
+ public void close() {
+ LOG.info("Closing LuceneIndexWriterFactory");
+ if (indexWriterPool == null) {
+ LOG.info("Not using an Index writer pool");
+ } else {
+ indexWriterPool.close();
+ }
Review Comment:
I am not sure if incremental indexing opens/closes a writer every time it kicks in.
In that case, we will see a lot of these logs. Should we lower the debug level for the log messages 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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "nfsantos (via GitHub)" <gi...@apache.org>.
nfsantos commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1957775714
##########
oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterPool.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.index.lucene.writer;
+
+import org.apache.jackrabbit.guava.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.plugins.index.ConfigHelper;
+import org.apache.jackrabbit.oak.plugins.index.FormattingUtils;
+import org.apache.lucene.index.IndexableField;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+
+public class IndexWriterPool {
+ private final static Logger LOG = LoggerFactory.getLogger(IndexWriterPool.class);
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = "oak.indexer.parallelWriter.maxBatchSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = 256;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = "oak.indexer.parallelWriter.queueSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = 64;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = "oak.indexer.parallelWriter.numberThreads";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = 4;
+
+ private final int maxBatchSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE);
+ private final int queueSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE);
+ private final int numberOfThreads = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS);
+
+ // The batch of operations that will be sent to the workers.
+ // Batching individual operations reduces the overhead of synchronization and context switching.
+ private final ArrayList<Operation> batch = new ArrayList<>(maxBatchSize);
+ // Shared queue between producer and workers
+ private final BlockingQueue<OperationBatch> queue = new ArrayBlockingQueue<>(queueSize);
+ private final List<Worker> workers;
+ private final List<Future<?>> workerFutures;
+ private final ExecutorService writersPool;
+ // Used to schedule a task that periodically prints statistics
+ private final ScheduledExecutorService monitorTaskExecutor;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ // Used to keep track of the sequence number of the batches that are currently being processed.
+ // This is used to wait until all operations for a writer are processed before closing it.
+ private final Object pendingBatchesLock = new Object();
+ private final HashSet<Long> pendingBatches = new HashSet<>();
+ private long batchSequenceNumber = 0;
+
+ // Statistics
+ private final long startTimeNanos = System.nanoTime();
+ private long updateCount = 0;
+ private long deleteCount = 0;
+ private long totalEnqueueTimeNanos = 0;
+
+ private static class OperationBatch {
+ final long sequenceNumber;
+ final Operation[] operations;
+
+ public OperationBatch(long sequenceNumber, Operation[] operations) {
+ Objects.requireNonNull(operations);
+ this.sequenceNumber = sequenceNumber;
+ this.operations = operations;
+ }
+ }
+
+ private static abstract class Operation {
+ final LuceneIndexWriter delegate;
+
+ public Operation(LuceneIndexWriter delegate) {
+ this.delegate = delegate;
+ }
+
+ abstract void execute() throws IOException;
+ }
+
+ private static class UpdateOperation extends Operation {
+ private final String path;
+ private final Iterable<? extends IndexableField> doc;
+
+ UpdateOperation(LuceneIndexWriter delegate, String path, Iterable<? extends IndexableField> doc) {
+ super(delegate);
+ this.path = path;
+ this.doc = doc;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.updateDocument(path, doc);
+ }
+ }
+
+ private static class DeleteOperation extends Operation {
+ private final String path;
+
+ DeleteOperation(LuceneIndexWriter delegate, String path) {
+ super(delegate);
+ this.path = path;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.deleteDocuments(path);
+ }
+ }
+
+ private static class CloseResult {
+ // Either result or error is non-null. The two constructors enforce this invariant.
+ final Boolean result;
+ final Throwable error;
+
+ CloseResult(boolean result) {
+ this.result = result;
+ this.error = null;
+ }
+
+ CloseResult(Throwable error) {
+ this.result = null;
+ this.error = error;
+ }
+
+ @Override
+ public String toString() {
+ return "CloseResult{" +
+ "result=" + result +
+ ", error=" + error +
+ '}';
+ }
+ }
+
+ private static class CloseWriterOperation extends Operation {
+ private final long timestamp;
+ private final SynchronousQueue<CloseResult> sync;
+
+ /**
+ * The close operation should be synchronous and applied only after all the write operations for this writer
+ * are processed.
+ *
+ * @param sync A synchronous queue used to wait for the result of the close operation.
+ */
+ CloseWriterOperation(LuceneIndexWriter delegate, long timestamp, SynchronousQueue<CloseResult> sync) {
+ super(delegate);
+ this.timestamp = timestamp;
+ this.sync = sync;
+ }
+
+ @Override
+ public void execute() {
+ try {
+ try {
+ boolean closeResult = delegate.close(timestamp);
+ sync.put(new CloseResult(closeResult));
+ } catch (IOException e) {
+ sync.put(new CloseResult(e));
+ }
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
Review Comment:
`sync.put()` throws an `InterrupedException`, so we cannot use it on the outermost exception handler. If `sync.put()` fails, then I don't think we can do anything else other than quitting with a `RuntimeException`.
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "nfsantos (via GitHub)" <gi...@apache.org>.
nfsantos merged PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "nfsantos (via GitHub)" <gi...@apache.org>.
nfsantos commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1957777212
##########
oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterPool.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.index.lucene.writer;
+
+import org.apache.jackrabbit.guava.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.plugins.index.ConfigHelper;
+import org.apache.jackrabbit.oak.plugins.index.FormattingUtils;
+import org.apache.lucene.index.IndexableField;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+
+public class IndexWriterPool {
+ private final static Logger LOG = LoggerFactory.getLogger(IndexWriterPool.class);
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = "oak.indexer.parallelWriter.maxBatchSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = 256;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = "oak.indexer.parallelWriter.queueSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = 64;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = "oak.indexer.parallelWriter.numberThreads";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = 4;
+
+ private final int maxBatchSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE);
+ private final int queueSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE);
+ private final int numberOfThreads = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS);
+
+ // The batch of operations that will be sent to the workers.
+ // Batching individual operations reduces the overhead of synchronization and context switching.
+ private final ArrayList<Operation> batch = new ArrayList<>(maxBatchSize);
+ // Shared queue between producer and workers
+ private final BlockingQueue<OperationBatch> queue = new ArrayBlockingQueue<>(queueSize);
+ private final List<Worker> workers;
+ private final List<Future<?>> workerFutures;
+ private final ExecutorService writersPool;
+ // Used to schedule a task that periodically prints statistics
+ private final ScheduledExecutorService monitorTaskExecutor;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ // Used to keep track of the sequence number of the batches that are currently being processed.
+ // This is used to wait until all operations for a writer are processed before closing it.
+ private final Object pendingBatchesLock = new Object();
+ private final HashSet<Long> pendingBatches = new HashSet<>();
+ private long batchSequenceNumber = 0;
+
+ // Statistics
+ private final long startTimeNanos = System.nanoTime();
+ private long updateCount = 0;
+ private long deleteCount = 0;
+ private long totalEnqueueTimeNanos = 0;
+
+ private static class OperationBatch {
+ final long sequenceNumber;
+ final Operation[] operations;
+
+ public OperationBatch(long sequenceNumber, Operation[] operations) {
+ Objects.requireNonNull(operations);
+ this.sequenceNumber = sequenceNumber;
+ this.operations = operations;
+ }
+ }
+
+ private static abstract class Operation {
+ final LuceneIndexWriter delegate;
+
+ public Operation(LuceneIndexWriter delegate) {
+ this.delegate = delegate;
+ }
+
+ abstract void execute() throws IOException;
+ }
+
+ private static class UpdateOperation extends Operation {
+ private final String path;
+ private final Iterable<? extends IndexableField> doc;
+
+ UpdateOperation(LuceneIndexWriter delegate, String path, Iterable<? extends IndexableField> doc) {
+ super(delegate);
+ this.path = path;
+ this.doc = doc;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.updateDocument(path, doc);
+ }
+ }
+
+ private static class DeleteOperation extends Operation {
+ private final String path;
+
+ DeleteOperation(LuceneIndexWriter delegate, String path) {
+ super(delegate);
+ this.path = path;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.deleteDocuments(path);
+ }
+ }
+
+ private static class CloseResult {
+ // Either result or error is non-null. The two constructors enforce this invariant.
+ final Boolean result;
+ final Throwable error;
+
+ CloseResult(boolean result) {
+ this.result = result;
+ this.error = null;
+ }
+
+ CloseResult(Throwable error) {
+ this.result = null;
+ this.error = error;
+ }
+
+ @Override
+ public String toString() {
+ return "CloseResult{" +
+ "result=" + result +
+ ", error=" + error +
+ '}';
+ }
+ }
+
+ private static class CloseWriterOperation extends Operation {
+ private final long timestamp;
+ private final SynchronousQueue<CloseResult> sync;
+
+ /**
+ * The close operation should be synchronous and applied only after all the write operations for this writer
+ * are processed.
+ *
+ * @param sync A synchronous queue used to wait for the result of the close operation.
+ */
+ CloseWriterOperation(LuceneIndexWriter delegate, long timestamp, SynchronousQueue<CloseResult> sync) {
+ super(delegate);
+ this.timestamp = timestamp;
+ this.sync = sync;
+ }
+
+ @Override
+ public void execute() {
+ try {
+ try {
+ boolean closeResult = delegate.close(timestamp);
+ sync.put(new CloseResult(closeResult));
+ } catch (IOException e) {
+ sync.put(new CloseResult(e));
+ }
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ // Sentinel used to terminate worker threads
+ final static OperationBatch SHUTDOWN = new OperationBatch(-1, new Operation[0]);
+
+ private class Worker implements Runnable {
+ private final long startTime = System.nanoTime();
+ private final int id;
+ private long opCount = 0;
+ private long batchesProcessed = 0;
+ private long totalBusyTime = 0;
+
+ public Worker(int id) {
+ this.id = id;
+ }
+
+ @Override
+ public void run() {
+ String oldName = Thread.currentThread().getName();
+ Thread.currentThread().setName("index-writer-" + id);
+ try {
+ LOG.info("[{}] Worker started", id);
+ while (true) {
+ OperationBatch op = queue.take();
+ if (op == SHUTDOWN) {
+ queue.add(SHUTDOWN);
+ LOG.info("[{}] Shutting down worker", id);
+ printStatistics();
+ return;
+ }
+ long start = System.nanoTime();
+ for (Operation operation : op.operations) {
+ operation.execute();
+ opCount++;
+ }
+ batchesProcessed++;
+ long durationNanos = System.nanoTime() - start;
+ totalBusyTime += durationNanos;
+ long durationMillis = durationNanos / 1_000_000;
+ if (durationMillis > 1000) {
+ LOG.info("[{}] Processing batch {} of size {} took {} millis.",
+ id, op.sequenceNumber, op.operations.length, durationMillis);
+ }
+ synchronized (pendingBatchesLock) {
+ pendingBatches.remove(op.sequenceNumber);
+ pendingBatchesLock.notifyAll();
+ }
+ }
+ } catch (InterruptedException | IOException e) {
+ LOG.warn("[{}] Interrupted while waiting to take an update operation from the queue", id, e);
+ throw new RuntimeException(e);
Review Comment:
Yes, here we are not trying to return a value to the thread that is waiting for these tasks to complete, the only information that needs to be sent back is success vs failure.
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "nfsantos (via GitHub)" <gi...@apache.org>.
nfsantos commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1953858315
##########
oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterPoolTest.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.index.lucene.writer;
+
+import org.apache.jackrabbit.oak.plugins.index.lucene.TestUtil;
+import org.apache.jackrabbit.oak.plugins.index.search.FieldNames;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.index.IndexableField;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.contrib.java.lang.system.ProvideSystemProperty;
+import org.junit.contrib.java.lang.system.RestoreSystemProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.junit.Assert.*;
Review Comment:
Removed the wildcard import
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "thomasmueller (via GitHub)" <gi...@apache.org>.
thomasmueller commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1955920579
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java:
##########
@@ -413,12 +413,13 @@ public void reindex() throws CommitFailedException, IOException {
}
}
log.info("Top slowest nodes to index (ms): {}", slowestTopKElements);
+ }
- indexerProviders.forEach(indexProvider -> {
- ExtractedTextCache extractedTextCache = indexProvider.getTextCache();
- CacheStats cacheStats = extractedTextCache == null ? null : extractedTextCache.getCacheStats();
- log.info("Text extraction cache statistics: {}", cacheStats == null ? "N/A" : cacheStats.cacheInfoAsString());
- });
+ for (NodeStateIndexerProvider indexerProvider : indexerProviders) {
+ indexerProvider.close();
Review Comment:
Hm, shouldn't we close it after logging?
(I guess currently "indexerProvider.getTextCache()" will work after closing, but it seems wrong to me to close it first).
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "thomasmueller (via GitHub)" <gi...@apache.org>.
thomasmueller commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1958287185
##########
oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/DefaultIndexWriter.java:
##########
@@ -146,6 +146,7 @@ public boolean close(long timestamp) throws IOException {
PERF_LOGGER.end(start, -1, "Completed suggester for directory {}", definition);
}
+ writer.forceMergeDeletes();
Review Comment:
What is the reason for this change?
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "thomasmueller (via GitHub)" <gi...@apache.org>.
thomasmueller commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1953059403
##########
oak-lucene/src/test/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterPoolTest.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.index.lucene.writer;
+
+import org.apache.jackrabbit.oak.plugins.index.lucene.TestUtil;
+import org.apache.jackrabbit.oak.plugins.index.search.FieldNames;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.index.IndexableField;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.contrib.java.lang.system.ProvideSystemProperty;
+import org.junit.contrib.java.lang.system.RestoreSystemProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.junit.Assert.*;
Review Comment:
I would avoid star imports.
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "thomasmueller (via GitHub)" <gi...@apache.org>.
thomasmueller commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1955929121
##########
oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterPool.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.index.lucene.writer;
+
+import org.apache.jackrabbit.guava.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.plugins.index.ConfigHelper;
+import org.apache.jackrabbit.oak.plugins.index.FormattingUtils;
+import org.apache.lucene.index.IndexableField;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+
+public class IndexWriterPool {
+ private final static Logger LOG = LoggerFactory.getLogger(IndexWriterPool.class);
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = "oak.indexer.parallelWriter.maxBatchSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = 256;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = "oak.indexer.parallelWriter.queueSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = 64;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = "oak.indexer.parallelWriter.numberThreads";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = 4;
+
+ private final int maxBatchSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE);
+ private final int queueSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE);
+ private final int numberOfThreads = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS);
+
+ // The batch of operations that will be sent to the workers.
+ // Batching individual operations reduces the overhead of synchronization and context switching.
+ private final ArrayList<Operation> batch = new ArrayList<>(maxBatchSize);
+ // Shared queue between producer and workers
+ private final BlockingQueue<OperationBatch> queue = new ArrayBlockingQueue<>(queueSize);
+ private final List<Worker> workers;
+ private final List<Future<?>> workerFutures;
+ private final ExecutorService writersPool;
+ // Used to schedule a task that periodically prints statistics
+ private final ScheduledExecutorService monitorTaskExecutor;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ // Used to keep track of the sequence number of the batches that are currently being processed.
+ // This is used to wait until all operations for a writer are processed before closing it.
+ private final Object pendingBatchesLock = new Object();
+ private final HashSet<Long> pendingBatches = new HashSet<>();
+ private long batchSequenceNumber = 0;
+
+ // Statistics
+ private final long startTimeNanos = System.nanoTime();
+ private long updateCount = 0;
+ private long deleteCount = 0;
+ private long totalEnqueueTimeNanos = 0;
+
+ private static class OperationBatch {
+ final long sequenceNumber;
+ final Operation[] operations;
+
+ public OperationBatch(long sequenceNumber, Operation[] operations) {
+ Objects.requireNonNull(operations);
+ this.sequenceNumber = sequenceNumber;
+ this.operations = operations;
+ }
+ }
+
+ private static abstract class Operation {
+ final LuceneIndexWriter delegate;
+
+ public Operation(LuceneIndexWriter delegate) {
+ this.delegate = delegate;
+ }
+
+ abstract void execute() throws IOException;
+ }
+
+ private static class UpdateOperation extends Operation {
+ private final String path;
+ private final Iterable<? extends IndexableField> doc;
+
+ UpdateOperation(LuceneIndexWriter delegate, String path, Iterable<? extends IndexableField> doc) {
+ super(delegate);
+ this.path = path;
+ this.doc = doc;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.updateDocument(path, doc);
+ }
+ }
+
+ private static class DeleteOperation extends Operation {
+ private final String path;
+
+ DeleteOperation(LuceneIndexWriter delegate, String path) {
+ super(delegate);
+ this.path = path;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.deleteDocuments(path);
+ }
+ }
+
+ private static class CloseResult {
+ // Either result or error is non-null. The two constructors enforce this invariant.
+ final Boolean result;
+ final Throwable error;
+
+ CloseResult(boolean result) {
+ this.result = result;
+ this.error = null;
+ }
+
+ CloseResult(Throwable error) {
+ this.result = null;
+ this.error = error;
+ }
+
+ @Override
+ public String toString() {
+ return "CloseResult{" +
+ "result=" + result +
+ ", error=" + error +
+ '}';
+ }
+ }
+
+ private static class CloseWriterOperation extends Operation {
+ private final long timestamp;
+ private final SynchronousQueue<CloseResult> sync;
+
+ /**
+ * The close operation should be synchronous and applied only after all the write operations for this writer
+ * are processed.
+ *
+ * @param sync A synchronous queue used to wait for the result of the close operation.
+ */
+ CloseWriterOperation(LuceneIndexWriter delegate, long timestamp, SynchronousQueue<CloseResult> sync) {
+ super(delegate);
+ this.timestamp = timestamp;
+ this.sync = sync;
+ }
+
+ @Override
+ public void execute() {
+ try {
+ try {
+ boolean closeResult = delegate.close(timestamp);
+ sync.put(new CloseResult(closeResult));
+ } catch (IOException e) {
+ sync.put(new CloseResult(e));
+ }
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
Review Comment:
I wonder if this is the right thing to do... Why not
```suggestion
LOG.warn("[{}] Interrupted while closing", timestamp, e);
sync.put(new CloseResult(new IOException(e)));
```
or something like that?
##########
oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterPool.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.index.lucene.writer;
+
+import org.apache.jackrabbit.guava.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.plugins.index.ConfigHelper;
+import org.apache.jackrabbit.oak.plugins.index.FormattingUtils;
+import org.apache.lucene.index.IndexableField;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+
+public class IndexWriterPool {
+ private final static Logger LOG = LoggerFactory.getLogger(IndexWriterPool.class);
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = "oak.indexer.parallelWriter.maxBatchSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = 256;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = "oak.indexer.parallelWriter.queueSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = 64;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = "oak.indexer.parallelWriter.numberThreads";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = 4;
+
+ private final int maxBatchSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE);
+ private final int queueSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE);
+ private final int numberOfThreads = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS);
+
+ // The batch of operations that will be sent to the workers.
+ // Batching individual operations reduces the overhead of synchronization and context switching.
+ private final ArrayList<Operation> batch = new ArrayList<>(maxBatchSize);
+ // Shared queue between producer and workers
+ private final BlockingQueue<OperationBatch> queue = new ArrayBlockingQueue<>(queueSize);
+ private final List<Worker> workers;
+ private final List<Future<?>> workerFutures;
+ private final ExecutorService writersPool;
+ // Used to schedule a task that periodically prints statistics
+ private final ScheduledExecutorService monitorTaskExecutor;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ // Used to keep track of the sequence number of the batches that are currently being processed.
+ // This is used to wait until all operations for a writer are processed before closing it.
+ private final Object pendingBatchesLock = new Object();
+ private final HashSet<Long> pendingBatches = new HashSet<>();
+ private long batchSequenceNumber = 0;
+
+ // Statistics
+ private final long startTimeNanos = System.nanoTime();
+ private long updateCount = 0;
+ private long deleteCount = 0;
+ private long totalEnqueueTimeNanos = 0;
+
+ private static class OperationBatch {
+ final long sequenceNumber;
+ final Operation[] operations;
+
+ public OperationBatch(long sequenceNumber, Operation[] operations) {
+ Objects.requireNonNull(operations);
+ this.sequenceNumber = sequenceNumber;
+ this.operations = operations;
+ }
+ }
+
+ private static abstract class Operation {
+ final LuceneIndexWriter delegate;
+
+ public Operation(LuceneIndexWriter delegate) {
+ this.delegate = delegate;
+ }
+
+ abstract void execute() throws IOException;
+ }
+
+ private static class UpdateOperation extends Operation {
+ private final String path;
+ private final Iterable<? extends IndexableField> doc;
+
+ UpdateOperation(LuceneIndexWriter delegate, String path, Iterable<? extends IndexableField> doc) {
+ super(delegate);
+ this.path = path;
+ this.doc = doc;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.updateDocument(path, doc);
+ }
+ }
+
+ private static class DeleteOperation extends Operation {
+ private final String path;
+
+ DeleteOperation(LuceneIndexWriter delegate, String path) {
+ super(delegate);
+ this.path = path;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.deleteDocuments(path);
+ }
+ }
+
+ private static class CloseResult {
+ // Either result or error is non-null. The two constructors enforce this invariant.
+ final Boolean result;
+ final Throwable error;
+
+ CloseResult(boolean result) {
+ this.result = result;
+ this.error = null;
+ }
+
+ CloseResult(Throwable error) {
+ this.result = null;
+ this.error = error;
+ }
+
+ @Override
+ public String toString() {
+ return "CloseResult{" +
+ "result=" + result +
+ ", error=" + error +
+ '}';
+ }
+ }
+
+ private static class CloseWriterOperation extends Operation {
+ private final long timestamp;
+ private final SynchronousQueue<CloseResult> sync;
+
+ /**
+ * The close operation should be synchronous and applied only after all the write operations for this writer
+ * are processed.
+ *
+ * @param sync A synchronous queue used to wait for the result of the close operation.
+ */
+ CloseWriterOperation(LuceneIndexWriter delegate, long timestamp, SynchronousQueue<CloseResult> sync) {
+ super(delegate);
+ this.timestamp = timestamp;
+ this.sync = sync;
+ }
+
+ @Override
+ public void execute() {
+ try {
+ try {
+ boolean closeResult = delegate.close(timestamp);
+ sync.put(new CloseResult(closeResult));
+ } catch (IOException e) {
+ sync.put(new CloseResult(e));
+ }
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ // Sentinel used to terminate worker threads
+ final static OperationBatch SHUTDOWN = new OperationBatch(-1, new Operation[0]);
+
+ private class Worker implements Runnable {
+ private final long startTime = System.nanoTime();
+ private final int id;
+ private long opCount = 0;
+ private long batchesProcessed = 0;
+ private long totalBusyTime = 0;
+
+ public Worker(int id) {
+ this.id = id;
+ }
+
+ @Override
+ public void run() {
+ String oldName = Thread.currentThread().getName();
+ Thread.currentThread().setName("index-writer-" + id);
+ try {
+ LOG.info("[{}] Worker started", id);
+ while (true) {
+ OperationBatch op = queue.take();
+ if (op == SHUTDOWN) {
+ queue.add(SHUTDOWN);
+ LOG.info("[{}] Shutting down worker", id);
+ printStatistics();
+ return;
+ }
+ long start = System.nanoTime();
+ for (Operation operation : op.operations) {
+ operation.execute();
+ opCount++;
+ }
+ batchesProcessed++;
+ long durationNanos = System.nanoTime() - start;
+ totalBusyTime += durationNanos;
+ long durationMillis = durationNanos / 1_000_000;
+ if (durationMillis > 1000) {
+ LOG.info("[{}] Processing batch {} of size {} took {} millis.",
+ id, op.sequenceNumber, op.operations.length, durationMillis);
+ }
+ synchronized (pendingBatchesLock) {
+ pendingBatches.remove(op.sequenceNumber);
+ pendingBatchesLock.notifyAll();
+ }
+ }
+ } catch (InterruptedException | IOException e) {
+ LOG.warn("[{}] Interrupted while waiting to take an update operation from the queue", id, e);
+ throw new RuntimeException(e);
Review Comment:
Here, I assume it makes sense to throw a RuntimeException
##########
oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterPool.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.index.lucene.writer;
+
+import org.apache.jackrabbit.guava.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.plugins.index.ConfigHelper;
+import org.apache.jackrabbit.oak.plugins.index.FormattingUtils;
+import org.apache.lucene.index.IndexableField;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+
+public class IndexWriterPool {
+ private final static Logger LOG = LoggerFactory.getLogger(IndexWriterPool.class);
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = "oak.indexer.parallelWriter.maxBatchSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = 256;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = "oak.indexer.parallelWriter.queueSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = 64;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = "oak.indexer.parallelWriter.numberThreads";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = 4;
+
+ private final int maxBatchSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE);
+ private final int queueSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE);
+ private final int numberOfThreads = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS);
+
+ // The batch of operations that will be sent to the workers.
+ // Batching individual operations reduces the overhead of synchronization and context switching.
+ private final ArrayList<Operation> batch = new ArrayList<>(maxBatchSize);
+ // Shared queue between producer and workers
+ private final BlockingQueue<OperationBatch> queue = new ArrayBlockingQueue<>(queueSize);
+ private final List<Worker> workers;
+ private final List<Future<?>> workerFutures;
+ private final ExecutorService writersPool;
+ // Used to schedule a task that periodically prints statistics
+ private final ScheduledExecutorService monitorTaskExecutor;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ // Used to keep track of the sequence number of the batches that are currently being processed.
+ // This is used to wait until all operations for a writer are processed before closing it.
+ private final Object pendingBatchesLock = new Object();
+ private final HashSet<Long> pendingBatches = new HashSet<>();
+ private long batchSequenceNumber = 0;
+
+ // Statistics
+ private final long startTimeNanos = System.nanoTime();
+ private long updateCount = 0;
+ private long deleteCount = 0;
+ private long totalEnqueueTimeNanos = 0;
+
+ private static class OperationBatch {
+ final long sequenceNumber;
+ final Operation[] operations;
+
+ public OperationBatch(long sequenceNumber, Operation[] operations) {
+ Objects.requireNonNull(operations);
+ this.sequenceNumber = sequenceNumber;
+ this.operations = operations;
+ }
+ }
+
+ private static abstract class Operation {
+ final LuceneIndexWriter delegate;
+
+ public Operation(LuceneIndexWriter delegate) {
+ this.delegate = delegate;
+ }
+
+ abstract void execute() throws IOException;
+ }
+
+ private static class UpdateOperation extends Operation {
+ private final String path;
+ private final Iterable<? extends IndexableField> doc;
+
+ UpdateOperation(LuceneIndexWriter delegate, String path, Iterable<? extends IndexableField> doc) {
+ super(delegate);
+ this.path = path;
+ this.doc = doc;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.updateDocument(path, doc);
+ }
+ }
+
+ private static class DeleteOperation extends Operation {
+ private final String path;
+
+ DeleteOperation(LuceneIndexWriter delegate, String path) {
+ super(delegate);
+ this.path = path;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.deleteDocuments(path);
+ }
+ }
+
+ private static class CloseResult {
+ // Either result or error is non-null. The two constructors enforce this invariant.
+ final Boolean result;
+ final Throwable error;
+
+ CloseResult(boolean result) {
+ this.result = result;
+ this.error = null;
+ }
+
+ CloseResult(Throwable error) {
+ this.result = null;
+ this.error = error;
+ }
+
+ @Override
+ public String toString() {
+ return "CloseResult{" +
+ "result=" + result +
+ ", error=" + error +
+ '}';
+ }
+ }
+
+ private static class CloseWriterOperation extends Operation {
+ private final long timestamp;
+ private final SynchronousQueue<CloseResult> sync;
+
+ /**
+ * The close operation should be synchronous and applied only after all the write operations for this writer
+ * are processed.
+ *
+ * @param sync A synchronous queue used to wait for the result of the close operation.
+ */
+ CloseWriterOperation(LuceneIndexWriter delegate, long timestamp, SynchronousQueue<CloseResult> sync) {
+ super(delegate);
+ this.timestamp = timestamp;
+ this.sync = sync;
+ }
+
+ @Override
+ public void execute() {
+ try {
+ try {
+ boolean closeResult = delegate.close(timestamp);
+ sync.put(new CloseResult(closeResult));
+ } catch (IOException e) {
+ sync.put(new CloseResult(e));
+ }
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ // Sentinel used to terminate worker threads
+ final static OperationBatch SHUTDOWN = new OperationBatch(-1, new Operation[0]);
+
+ private class Worker implements Runnable {
+ private final long startTime = System.nanoTime();
+ private final int id;
+ private long opCount = 0;
+ private long batchesProcessed = 0;
+ private long totalBusyTime = 0;
+
+ public Worker(int id) {
+ this.id = id;
+ }
+
+ @Override
+ public void run() {
+ String oldName = Thread.currentThread().getName();
+ Thread.currentThread().setName("index-writer-" + id);
+ try {
+ LOG.info("[{}] Worker started", id);
+ while (true) {
+ OperationBatch op = queue.take();
+ if (op == SHUTDOWN) {
+ queue.add(SHUTDOWN);
+ LOG.info("[{}] Shutting down worker", id);
+ printStatistics();
+ return;
+ }
+ long start = System.nanoTime();
+ for (Operation operation : op.operations) {
+ operation.execute();
+ opCount++;
+ }
+ batchesProcessed++;
+ long durationNanos = System.nanoTime() - start;
+ totalBusyTime += durationNanos;
+ long durationMillis = durationNanos / 1_000_000;
+ if (durationMillis > 1000) {
+ LOG.info("[{}] Processing batch {} of size {} took {} millis.",
+ id, op.sequenceNumber, op.operations.length, durationMillis);
+ }
+ synchronized (pendingBatchesLock) {
+ pendingBatches.remove(op.sequenceNumber);
+ pendingBatchesLock.notifyAll();
+ }
+ }
+ } catch (InterruptedException | IOException e) {
Review Comment:
Should we have a different log message for IOException?
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "nfsantos (via GitHub)" <gi...@apache.org>.
nfsantos commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1958271472
##########
oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/DefaultIndexWriterFactory.java:
##########
@@ -50,14 +64,30 @@ public LuceneIndexWriter newInstance(IndexDefinition def, NodeBuilder definition
"Expected %s but found %s for index definition",
LuceneIndexDefinition.class, def.getClass());
- LuceneIndexDefinition definition = (LuceneIndexDefinition)def;
+ LuceneIndexDefinition definition = (LuceneIndexDefinition) def;
- if (mountInfoProvider.hasNonDefaultMounts()){
- return new MultiplexingIndexWriter(directoryFactory, mountInfoProvider, definition,
- definitionBuilder, reindex, writerConfig);
+ if (mountInfoProvider.hasNonDefaultMounts()) {
+ return wrapWithPipelinedIndexWriter(
+ new MultiplexingIndexWriter(directoryFactory, mountInfoProvider, definition, definitionBuilder, reindex, writerConfig),
+ definition.getIndexName());
}
- return new DefaultIndexWriter(definition, definitionBuilder, directoryFactory,
+ DefaultIndexWriter writer = new DefaultIndexWriter(definition, definitionBuilder, directoryFactory,
FulltextIndexConstants.INDEX_DATA_CHILD_NAME,
LuceneIndexConstants.SUGGEST_DATA_CHILD_NAME, reindex, writerConfig);
+
+ return wrapWithPipelinedIndexWriter(writer, definition.getIndexName());
+ }
+
+ public void close() {
+ LOG.info("Closing LuceneIndexWriterFactory");
+ if (indexWriterPool == null) {
+ LOG.info("Not using an Index writer pool");
+ } else {
+ indexWriterPool.close();
+ }
Review Comment:
I have lowered the level to DEBUG.
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "nfsantos (via GitHub)" <gi...@apache.org>.
nfsantos commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1958543138
##########
oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/DefaultIndexWriter.java:
##########
@@ -146,6 +146,7 @@ public boolean close(long timestamp) throws IOException {
PERF_LOGGER.end(start, -1, "Completed suggester for directory {}", definition);
}
+ writer.forceMergeDeletes();
Review Comment:
It went in by mistake, I reverted it.
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "nfsantos (via GitHub)" <gi...@apache.org>.
nfsantos commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1957784231
##########
oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/writer/IndexWriterPool.java:
##########
@@ -0,0 +1,407 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.index.lucene.writer;
+
+import org.apache.jackrabbit.guava.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
+import org.apache.jackrabbit.oak.plugins.index.ConfigHelper;
+import org.apache.jackrabbit.oak.plugins.index.FormattingUtils;
+import org.apache.lucene.index.IndexableField;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+
+public class IndexWriterPool {
+ private final static Logger LOG = LoggerFactory.getLogger(IndexWriterPool.class);
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = "oak.indexer.parallelWriter.maxBatchSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE = 256;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = "oak.indexer.parallelWriter.queueSize";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE = 64;
+
+ public static final String OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = "oak.indexer.parallelWriter.numberThreads";
+ public static final int DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS = 4;
+
+ private final int maxBatchSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_MAX_BATCH_SIZE);
+ private final int queueSize = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_QUEUE_SIZE);
+ private final int numberOfThreads = ConfigHelper.getSystemPropertyAsInt(OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS, DEFAULT_OAK_INDEXER_PARALLEL_WRITER_NUMBER_THREADS);
+
+ // The batch of operations that will be sent to the workers.
+ // Batching individual operations reduces the overhead of synchronization and context switching.
+ private final ArrayList<Operation> batch = new ArrayList<>(maxBatchSize);
+ // Shared queue between producer and workers
+ private final BlockingQueue<OperationBatch> queue = new ArrayBlockingQueue<>(queueSize);
+ private final List<Worker> workers;
+ private final List<Future<?>> workerFutures;
+ private final ExecutorService writersPool;
+ // Used to schedule a task that periodically prints statistics
+ private final ScheduledExecutorService monitorTaskExecutor;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ // Used to keep track of the sequence number of the batches that are currently being processed.
+ // This is used to wait until all operations for a writer are processed before closing it.
+ private final Object pendingBatchesLock = new Object();
+ private final HashSet<Long> pendingBatches = new HashSet<>();
+ private long batchSequenceNumber = 0;
+
+ // Statistics
+ private final long startTimeNanos = System.nanoTime();
+ private long updateCount = 0;
+ private long deleteCount = 0;
+ private long totalEnqueueTimeNanos = 0;
+
+ private static class OperationBatch {
+ final long sequenceNumber;
+ final Operation[] operations;
+
+ public OperationBatch(long sequenceNumber, Operation[] operations) {
+ Objects.requireNonNull(operations);
+ this.sequenceNumber = sequenceNumber;
+ this.operations = operations;
+ }
+ }
+
+ private static abstract class Operation {
+ final LuceneIndexWriter delegate;
+
+ public Operation(LuceneIndexWriter delegate) {
+ this.delegate = delegate;
+ }
+
+ abstract void execute() throws IOException;
+ }
+
+ private static class UpdateOperation extends Operation {
+ private final String path;
+ private final Iterable<? extends IndexableField> doc;
+
+ UpdateOperation(LuceneIndexWriter delegate, String path, Iterable<? extends IndexableField> doc) {
+ super(delegate);
+ this.path = path;
+ this.doc = doc;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.updateDocument(path, doc);
+ }
+ }
+
+ private static class DeleteOperation extends Operation {
+ private final String path;
+
+ DeleteOperation(LuceneIndexWriter delegate, String path) {
+ super(delegate);
+ this.path = path;
+ }
+
+ @Override
+ public void execute() throws IOException {
+ delegate.deleteDocuments(path);
+ }
+ }
+
+ private static class CloseResult {
+ // Either result or error is non-null. The two constructors enforce this invariant.
+ final Boolean result;
+ final Throwable error;
+
+ CloseResult(boolean result) {
+ this.result = result;
+ this.error = null;
+ }
+
+ CloseResult(Throwable error) {
+ this.result = null;
+ this.error = error;
+ }
+
+ @Override
+ public String toString() {
+ return "CloseResult{" +
+ "result=" + result +
+ ", error=" + error +
+ '}';
+ }
+ }
+
+ private static class CloseWriterOperation extends Operation {
+ private final long timestamp;
+ private final SynchronousQueue<CloseResult> sync;
+
+ /**
+ * The close operation should be synchronous and applied only after all the write operations for this writer
+ * are processed.
+ *
+ * @param sync A synchronous queue used to wait for the result of the close operation.
+ */
+ CloseWriterOperation(LuceneIndexWriter delegate, long timestamp, SynchronousQueue<CloseResult> sync) {
+ super(delegate);
+ this.timestamp = timestamp;
+ this.sync = sync;
+ }
+
+ @Override
+ public void execute() {
+ try {
+ try {
+ boolean closeResult = delegate.close(timestamp);
+ sync.put(new CloseResult(closeResult));
+ } catch (IOException e) {
+ sync.put(new CloseResult(e));
+ }
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ // Sentinel used to terminate worker threads
+ final static OperationBatch SHUTDOWN = new OperationBatch(-1, new Operation[0]);
+
+ private class Worker implements Runnable {
+ private final long startTime = System.nanoTime();
+ private final int id;
+ private long opCount = 0;
+ private long batchesProcessed = 0;
+ private long totalBusyTime = 0;
+
+ public Worker(int id) {
+ this.id = id;
+ }
+
+ @Override
+ public void run() {
+ String oldName = Thread.currentThread().getName();
+ Thread.currentThread().setName("index-writer-" + id);
+ try {
+ LOG.info("[{}] Worker started", id);
+ while (true) {
+ OperationBatch op = queue.take();
+ if (op == SHUTDOWN) {
+ queue.add(SHUTDOWN);
+ LOG.info("[{}] Shutting down worker", id);
+ printStatistics();
+ return;
+ }
+ long start = System.nanoTime();
+ for (Operation operation : op.operations) {
+ operation.execute();
+ opCount++;
+ }
+ batchesProcessed++;
+ long durationNanos = System.nanoTime() - start;
+ totalBusyTime += durationNanos;
+ long durationMillis = durationNanos / 1_000_000;
+ if (durationMillis > 1000) {
+ LOG.info("[{}] Processing batch {} of size {} took {} millis.",
+ id, op.sequenceNumber, op.operations.length, durationMillis);
+ }
+ synchronized (pendingBatchesLock) {
+ pendingBatches.remove(op.sequenceNumber);
+ pendingBatchesLock.notifyAll();
+ }
+ }
+ } catch (InterruptedException | IOException e) {
Review Comment:
Yes, the message was not correct for the case of an IOException. I removed the IOException from the handler of the InterruptedException and left only the catch Throwable afterwards to deal with all other errors.
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] OAK-11438 - Add support for parallelization in the Lucene index writer backend. [jackrabbit-oak]
Posted by "nfsantos (via GitHub)" <gi...@apache.org>.
nfsantos commented on code in PR #2031:
URL: https://github.com/apache/jackrabbit-oak/pull/2031#discussion_r1957784902
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java:
##########
@@ -413,12 +413,13 @@ public void reindex() throws CommitFailedException, IOException {
}
}
log.info("Top slowest nodes to index (ms): {}", slowestTopKElements);
+ }
- indexerProviders.forEach(indexProvider -> {
- ExtractedTextCache extractedTextCache = indexProvider.getTextCache();
- CacheStats cacheStats = extractedTextCache == null ? null : extractedTextCache.getCacheStats();
- log.info("Text extraction cache statistics: {}", cacheStats == null ? "N/A" : cacheStats.cacheInfoAsString());
- });
+ for (NodeStateIndexerProvider indexerProvider : indexerProviders) {
+ indexerProvider.close();
Review Comment:
Good point, it's a bad practise to access an object after closing it. I moved the close to the end of the block.
--
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: oak-dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org