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

Use better overflow strategy

Browse Source
This commit is contained in:
Andrea Cavalli 2021-03-03 21:32:45 +01:00
parent b71f3dceed
commit e3fcf7f74f
1 changed files with 63 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -59,6 +59,7 @@ import org.jetbrains.annotations.Nullable;
import org.warp.commonutils.log.Logger; import org.warp.commonutils.log.Logger;
import org.warp.commonutils.log.LoggerFactory; import org.warp.commonutils.log.LoggerFactory;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink.OverflowStrategy;
import reactor.core.publisher.GroupedFlux; import reactor.core.publisher.GroupedFlux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Scheduler;
@ -563,62 +564,74 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
requestsAvailable.release(); requestsAvailable.release();
}); });
luceneQueryScheduler.schedule(() -> { try {
try { luceneQueryScheduler.schedule(() -> {
requestsAvailable.acquire(); try {
if (!cancelled.get()) { if (!cancelled.get()) {
if (doDistributedPre) { 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);
streamSearcher.search(indexSearcher, streamSearcher.search(indexSearcher,
luceneQuery, luceneQuery,
boundedLimit, boundedLimit,
luceneSort, luceneSort,
luceneScoreMode, luceneScoreMode,
minCompetitiveScore, minCompetitiveScore,
keyFieldName, keyFieldName,
keyScore -> { keyScore -> {
try { try {
if (requests.get() <= 0 && !cancelled.get()) { if (cancelled.get()) {
requestsAvailable.acquire(); return HandleResult.HALT;
} }
if (cancelled.get()) { while (requests.get() <= 0) {
requestsAvailable.acquire();
if (cancelled.get()) {
return HandleResult.HALT;
}
}
requests.decrementAndGet();
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;
} }
requests.updateAndGet(n -> n > 0 ? n - 1 : 0); },
sink.next(fixKeyScore(keyScore, scoreDivisor)); totalHitsCount -> {
return HandleResult.CONTINUE; try {
} catch (Exception ex) { if (cancelled.get()) {
sink.error(ex); return;
cancelled.set(true); }
requestsAvailable.release(); while (requests.get() <= 0) {
return HandleResult.HALT; requestsAvailable.acquire();
} if (cancelled.get()) {
}, return;
totalHitsCount -> { }
try { }
if (requests.get() <= 0 && !cancelled.get()) { requests.decrementAndGet();
requestsAvailable.acquire(); sink.next(new LLTotalHitsCount(totalHitsCount));
} catch (Exception ex) {
sink.error(ex);
cancelled.set(true);
requestsAvailable.release();
} }
requests.updateAndGet(n -> n > 0 ? n - 1 : 0);
sink.next(new LLTotalHitsCount(totalHitsCount));
} catch (Exception ex) {
sink.error(ex);
cancelled.set(true);
requestsAvailable.release();
} }
} );
); }
sink.complete();
} }
sink.complete(); } catch (Exception ex) {
sink.error(ex);
} }
} catch (Exception ex) { });
sink.error(ex); } catch (Exception ex) {
} sink.error(ex);
}); }
}).subscribeOn(Schedulers.boundedElastic())))); }, OverflowStrategy.ERROR).subscribeOn(Schedulers.boundedElastic()))));
} }
@Override @Override