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

Fix deadlock

Browse Source
This commit is contained in:
Andrea Cavalli 2021-03-06 17:28:33 +01:00
parent 08434d475c
commit 99f3686eab
1 changed files with 61 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -89,11 +89,11 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
"lucene-" + name, "lucene-" + name,
60 60
); );
private static final Supplier<Scheduler> lowMemorySchedulerSupplier = Suppliers.memoize(() -> private final Supplier<Scheduler> lowMemorySchedulerSupplier = Suppliers.memoize(() ->
Schedulers.newBoundedElastic(1, Schedulers.DEFAULT_BOUNDED_ELASTIC_QUEUESIZE, "lucene-low-memory", Integer.MAX_VALUE))::get; Schedulers.newBoundedElastic(1, Schedulers.DEFAULT_BOUNDED_ELASTIC_QUEUESIZE, "lucene-low-memory", Integer.MAX_VALUE))::get;
private static final Supplier<Scheduler> querySchedulerSupplier = Suppliers.memoize(() -> boundedSchedulerSupplier.apply("query"))::get; private final Supplier<Scheduler> querySchedulerSupplier = Suppliers.memoize(() -> boundedSchedulerSupplier.apply("query"))::get;
private static final Supplier<Scheduler> blockingSchedulerSupplier = Suppliers.memoize(() -> boundedSchedulerSupplier.apply("blocking"))::get; private final Supplier<Scheduler> blockingSchedulerSupplier = Suppliers.memoize(() -> boundedSchedulerSupplier.apply("blocking"))::get;
private static final Supplier<Scheduler> blockingLuceneSearchSchedulerSupplier = Suppliers.memoize(() -> boundedSchedulerSupplier.apply("search-blocking"))::get; private final Supplier<Scheduler> blockingLuceneSearchSchedulerSupplier = Suppliers.memoize(() -> boundedSchedulerSupplier.apply("search-blocking"))::get;
/** /**
* Lucene query scheduler. * Lucene query scheduler.
*/ */
@ -475,7 +475,7 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
} else { } else {
return Mono.just(signal); return Mono.just(signal);
} }
}).<LLSearchResult>dematerialize()); }).dematerialize());
}); });
} }
@ -537,7 +537,7 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
return Mono.just(signal); return Mono.just(signal);
} }
}) })
.<LLSearchResult>dematerialize() .dematerialize()
); );
} }
@ -568,73 +568,69 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
}); });
try { try {
blockingLuceneSearchScheduler.schedule(() -> { if (!cancelled.get()) {
try { if (doDistributedPre) {
if (!cancelled.get()) { //noinspection BlockingMethodInNonBlockingContext
if (doDistributedPre) { allowOnlyQueryParsingCollectorStreamSearcher.search(indexSearcher, luceneQuery);
allowOnlyQueryParsingCollectorStreamSearcher.search(indexSearcher, luceneQuery); sink.next(new LLTotalHitsCount(0L));
sink.next(new LLTotalHitsCount(0L)); } else {
} else { int boundedLimit = Math.max(0, limit > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) limit);
int boundedLimit = Math.max(0, limit > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) limit); //noinspection BlockingMethodInNonBlockingContext
streamSearcher.search(indexSearcher, streamSearcher.search(indexSearcher,
luceneQuery, luceneQuery,
boundedLimit, boundedLimit,
luceneSort, luceneSort,
luceneScoreMode, luceneScoreMode,
minCompetitiveScore, minCompetitiveScore,
keyFieldName, keyFieldName,
keyScore -> { keyScore -> {
try { try {
if (cancelled.get()) { if (cancelled.get()) {
return HandleResult.HALT; return HandleResult.HALT;
} }
while (requests.decrementAndGet() < 0) { while (requests.decrementAndGet() < 0) {
requests.incrementAndGet(); requests.incrementAndGet();
requestsAvailable.acquire(); requestsAvailable.acquire();
if (cancelled.get()) { if (cancelled.get()) {
return HandleResult.HALT;
}
}
sink.next(fixKeyScore(keyScore, scoreDivisor));
return HandleResult.CONTINUE;
} catch (Exception ex) {
sink.error(ex);
cancelled.set(true);
requestsAvailable.release();
return HandleResult.HALT; return HandleResult.HALT;
} }
}, }
totalHitsCount -> { sink.next(fixKeyScore(keyScore, scoreDivisor));
try { return HandleResult.CONTINUE;
if (cancelled.get()) { } catch (Exception ex) {
return; sink.error(ex);
} cancelled.set(true);
while (requests.decrementAndGet() < 0) { requestsAvailable.release();
requests.incrementAndGet(); return HandleResult.HALT;
requestsAvailable.acquire(); }
if (cancelled.get()) { },
return; totalHitsCount -> {
} try {
} if (cancelled.get()) {
sink.next(new LLTotalHitsCount(totalHitsCount)); return;
} catch (Exception ex) { }
sink.error(ex); while (requests.decrementAndGet() < 0) {
cancelled.set(true); requests.incrementAndGet();
requestsAvailable.release(); requestsAvailable.acquire();
if (cancelled.get()) {
return;
} }
} }
); sink.next(new LLTotalHitsCount(totalHitsCount));
} } catch (Exception ex) {
sink.complete(); sink.error(ex);
} cancelled.set(true);
} catch (Exception ex) { requestsAvailable.release();
sink.error(ex); }
}
);
} }
}); sink.complete();
}
} catch (Exception ex) { } catch (Exception ex) {
sink.error(ex); sink.error(ex);
} }
}, OverflowStrategy.BUFFER).subscribeOn(luceneQueryScheduler)))); }, OverflowStrategy.ERROR).subscribeOn(blockingLuceneSearchScheduler).publishOn(luceneQueryScheduler))));
} }
@Override @Override