andreacavalli/CavalliumDBEngine
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

use a reentrantlock to avoid multiple merges at the same time

Browse Source
This commit is contained in:
Andrea Cavalli 2021-12-17 02:18:30 +01:00
parent 1dffb55572
commit 1a35930909
1 changed files with 67 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
src/main/java/it/cavallium/dbengine/database/disk/LLLocalLuceneIndex.java
View File

@ -2,6 +2,8 @@ package it.cavallium.dbengine.database.disk;
import static it.cavallium.dbengine.client.UninterruptibleScheduler.uninterruptibleScheduler; import static it.cavallium.dbengine.client.UninterruptibleScheduler.uninterruptibleScheduler;
import static it.cavallium.dbengine.database.LLUtils.MARKER_LUCENE; import static it.cavallium.dbengine.database.LLUtils.MARKER_LUCENE;
import static it.cavallium.dbengine.database.LLUtils.toDocument;
import static it.cavallium.dbengine.database.LLUtils.toFields;
import static it.cavallium.dbengine.lucene.searcher.LLSearchTransformer.NO_TRANSFORMATION; import static it.cavallium.dbengine.lucene.searcher.LLSearchTransformer.NO_TRANSFORMATION;
import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.MeterRegistry;
@ -50,6 +52,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Phaser; import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.Logger;
import org.apache.lucene.analysis.miscellaneous.PerFieldAnalyzerWrapper; import org.apache.lucene.analysis.miscellaneous.PerFieldAnalyzerWrapper;
@ -91,8 +94,9 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
* There is only a single thread globally to not overwhelm the disk with * There is only a single thread globally to not overwhelm the disk with
* concurrent commits or concurrent refreshes. * concurrent commits or concurrent refreshes.
*/ */
private static final ReentrantLock shutdownLock = new ReentrantLock();
private static final Scheduler luceneHeavyTasksScheduler = uninterruptibleScheduler(Schedulers.single(Schedulers.boundedElastic())); private static final Scheduler luceneHeavyTasksScheduler = uninterruptibleScheduler(Schedulers.single(Schedulers.boundedElastic()));
private static final ExecutorService SAFE_EXECUTOR = Executors.newCachedThreadPool(new ShortNamedThreadFactory("lucene-index-impl"));
private final MeterRegistry meterRegistry; private final MeterRegistry meterRegistry;
private final String luceneIndexName; private final String luceneIndexName;
@ -287,13 +291,6 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
.publishOn(Schedulers.parallel()); .publishOn(Schedulers.parallel());
} }
private <V> Mono<V> runSafe(IORunnable runnable) {
return Mono.<V>fromCallable(() -> {
runnable.run();
return null;
}).subscribeOn(uninterruptibleScheduler(Schedulers.boundedElastic())).publishOn(Schedulers.parallel());
}
@Override @Override
public Mono<Void> releaseSnapshot(LLSnapshot snapshot) { public Mono<Void> releaseSnapshot(LLSnapshot snapshot) {
return snapshotsManager.releaseSnapshot(snapshot); return snapshotsManager.releaseSnapshot(snapshot);
@ -301,40 +298,43 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
@Override @Override
public Mono<Void> addDocument(LLTerm key, LLUpdateDocument doc) { public Mono<Void> addDocument(LLTerm key, LLUpdateDocument doc) {
return this.<Void>runSafe(() -> indexWriter.addDocument(LLUtils.toDocument(doc))).transform(this::ensureOpen); return this.<Void>runSafe(() -> {
indexWriter.addDocument(toDocument(doc));
return null;
}).transform(this::ensureOpen);
} }
@Override @Override
public Mono<Void> addDocuments(Flux<Entry<LLTerm, LLUpdateDocument>> documents) { public Mono<Void> addDocuments(Flux<Entry<LLTerm, LLUpdateDocument>> documents) {
return documents return documents.collectList().flatMap(documentsList -> this.<Void>runSafe(() -> {
.collectList() indexWriter.addDocuments(LLUtils.toDocumentsFromEntries(documentsList));
.flatMap(documentsList -> this.<Void>runSafe(() -> indexWriter.addDocuments(LLUtils return null;
.toDocumentsFromEntries(documentsList)))) })).transform(this::ensureOpen);
.transform(this::ensureOpen);
} }
@Override @Override
public Mono<Void> deleteDocument(LLTerm id) { public Mono<Void> deleteDocument(LLTerm id) {
return this.<Void>runSafe(() -> indexWriter.deleteDocuments(LLUtils.toTerm(id))).transform(this::ensureOpen); return this.<Void>runSafe(() -> {
indexWriter.deleteDocuments(LLUtils.toTerm(id));
return null;
}).transform(this::ensureOpen);
} }
@Override @Override
public Mono<Void> update(LLTerm id, LLIndexRequest request) { public Mono<Void> update(LLTerm id, LLIndexRequest request) {
return this return this
.<Void>runSafe(() -> { .<Void>runSafe(() -> {
if (request instanceof LLUpdateDocument updateDocument) { switch (request) {
indexWriter.updateDocument(LLUtils.toTerm(id), LLUtils.toDocument(updateDocument)); case LLUpdateDocument updateDocument ->
} else if (request instanceof LLSoftUpdateDocument softUpdateDocument) { indexWriter.updateDocument(LLUtils.toTerm(id), toDocument(updateDocument));
indexWriter.softUpdateDocument(LLUtils.toTerm(id), case LLSoftUpdateDocument softUpdateDocument -> indexWriter.softUpdateDocument(LLUtils.toTerm(id),
LLUtils.toDocument(softUpdateDocument.items()), toDocument(softUpdateDocument.items()), toFields(softUpdateDocument.softDeleteItems()));
LLUtils.toFields(softUpdateDocument.softDeleteItems()) case LLUpdateFields updateFields ->
); indexWriter.updateDocValues(LLUtils.toTerm(id), toFields(updateFields.items()));
} else if (request instanceof LLUpdateFields updateFields) { case null, default -> throw new UnsupportedOperationException("Unexpected request type: " + request);
indexWriter.updateDocValues(LLUtils.toTerm(id), LLUtils.toFields(updateFields.items()));
} else {
throw new UnsupportedOperationException("Unexpected request type: " + request);
} }
return null;
}) })
.transform(this::ensureOpen); .transform(this::ensureOpen);
} }
@ -349,17 +349,24 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
for (Entry<LLTerm, LLUpdateDocument> entry : documentsMap.entrySet()) { for (Entry<LLTerm, LLUpdateDocument> entry : documentsMap.entrySet()) {
LLTerm key = entry.getKey(); LLTerm key = entry.getKey();
LLUpdateDocument value = entry.getValue(); LLUpdateDocument value = entry.getValue();
indexWriter.updateDocument(LLUtils.toTerm(key), LLUtils.toDocument(value)); indexWriter.updateDocument(LLUtils.toTerm(key), toDocument(value));
} }
return null;
}).transform(this::ensureOpen); }).transform(this::ensureOpen);
} }
@Override @Override
public Mono<Void> deleteAll() { public Mono<Void> deleteAll() {
return this.<Void>runSafe(() -> { return this.<Void>runSafe(() -> {
indexWriter.deleteAll(); shutdownLock.lock();
indexWriter.forceMergeDeletes(true); try {
indexWriter.commit(); indexWriter.deleteAll();
indexWriter.forceMergeDeletes(true);
indexWriter.commit();
} finally {
shutdownLock.unlock();
}
return null;
}).subscribeOn(luceneHeavyTasksScheduler).transform(this::ensureOpen); }).subscribeOn(luceneHeavyTasksScheduler).transform(this::ensureOpen);
} }
@ -431,12 +438,15 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
.subscribeOn(luceneHeavyTasksScheduler) .subscribeOn(luceneHeavyTasksScheduler)
.then(searcherManager.close()) .then(searcherManager.close())
.then(Mono.<Void>fromCallable(() -> { .then(Mono.<Void>fromCallable(() -> {
logger.info("Closing IndexWriter..."); shutdownLock.lock();
//noinspection BlockingMethodInNonBlockingContext try {
indexWriter.close(); logger.info("Closing IndexWriter...");
//noinspection BlockingMethodInNonBlockingContext indexWriter.close();
directory.close(); directory.close();
logger.info("IndexWriter closed"); logger.info("IndexWriter closed");
} finally {
shutdownLock.unlock();
}
return null; return null;
}).subscribeOn(luceneHeavyTasksScheduler)) }).subscribeOn(luceneHeavyTasksScheduler))
@ -457,8 +467,12 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
return Mono return Mono
.<Void>fromCallable(() -> { .<Void>fromCallable(() -> {
if (activeTasks.isTerminated()) return null; if (activeTasks.isTerminated()) return null;
//noinspection BlockingMethodInNonBlockingContext shutdownLock.lock();
indexWriter.flush(); try {
indexWriter.flush();
} finally {
shutdownLock.unlock();
}
return null; return null;
}) })
.subscribeOn(luceneHeavyTasksScheduler) .subscribeOn(luceneHeavyTasksScheduler)
@ -472,12 +486,15 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
activeTasks.register(); activeTasks.register();
try { try {
if (activeTasks.isTerminated()) return null; if (activeTasks.isTerminated()) return null;
if (force) { shutdownLock.lock();
//noinspection BlockingMethodInNonBlockingContext try {
searcherManager.maybeRefreshBlocking(); if (force) {
} else { searcherManager.maybeRefreshBlocking();
//noinspection BlockingMethodInNonBlockingContext } else {
searcherManager.maybeRefresh(); searcherManager.maybeRefresh();
}
} finally {
shutdownLock.unlock();
} }
} finally { } finally {
activeTasks.arriveAndDeregister(); activeTasks.arriveAndDeregister();
@ -488,18 +505,24 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
} }
private void scheduledCommit() { private void scheduledCommit() {
shutdownLock.lock();
try { try {
indexWriter.commit(); indexWriter.commit();
} catch (IOException ex) { } catch (IOException ex) {
logger.error(MARKER_LUCENE, "Failed to execute a scheduled commit", ex); logger.error(MARKER_LUCENE, "Failed to execute a scheduled commit", ex);
} finally {
shutdownLock.unlock();
} }
} }
private void scheduledMerge() { private void scheduledMerge() {
shutdownLock.lock();
try { try {
indexWriter.maybeMerge(); indexWriter.maybeMerge();
} catch (IOException ex) { } catch (IOException ex) {
logger.error(MARKER_LUCENE, "Failed to execute a scheduled merge", ex); logger.error(MARKER_LUCENE, "Failed to execute a scheduled merge", ex);
} finally {
shutdownLock.unlock();
} }
} }