From 24cf7ea58d18683af109f16b0cf9e026b59a91d4 Mon Sep 17 00:00:00 2001 From: Andrea Cavalli Date: Tue, 5 Apr 2022 00:37:44 +0200 Subject: [PATCH] Add dedicated scheduler --- .../dbengine/database/disk/LLLocalLuceneIndex.java | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalLuceneIndex.java b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalLuceneIndex.java index 2e8c780..e57964a 100644 --- a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalLuceneIndex.java +++ b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalLuceneIndex.java @@ -93,7 +93,13 @@ public class LLLocalLuceneIndex implements LLLuceneIndex { new ShortNamedThreadFactory("heavy-tasks").setDaemon(true).withGroup(new ThreadGroup("lucene-heavy-tasks")), Math.toIntExact(Duration.ofHours(1).toSeconds()) )); - private static final Scheduler bulkScheduler = uninterruptibleScheduler(Schedulers.boundedElastic()); + private static final Scheduler luceneWriteScheduler = uninterruptibleScheduler(Schedulers.newBoundedElastic( + DEFAULT_BOUNDED_ELASTIC_SIZE, + DEFAULT_BOUNDED_ELASTIC_QUEUESIZE, + new ShortNamedThreadFactory("lucene-write").setDaemon(true).withGroup(new ThreadGroup("lucene-write")), + Math.toIntExact(Duration.ofHours(1).toSeconds()) + )); + private static final Scheduler bulkScheduler = luceneWriteScheduler; static { LLUtils.initHooks(); @@ -258,7 +264,7 @@ public class LLLocalLuceneIndex implements LLLuceneIndex { private Mono runSafe(Callable callable) { return Mono .fromCallable(callable) - .subscribeOn(uninterruptibleScheduler(Schedulers.boundedElastic())) + .subscribeOn(luceneWriteScheduler) .publishOn(Schedulers.parallel()); } @@ -422,7 +428,7 @@ public class LLLocalLuceneIndex implements LLLuceneIndex { shutdownLock.unlock(); } return null; - }).subscribeOn(luceneHeavyTasksScheduler).transform(this::ensureOpen); + }).subscribeOn(luceneHeavyTasksScheduler).publishOn(Schedulers.parallel()).transform(this::ensureOpen); } @Override