diff --git a/pom.xml b/pom.xml index 8aabc35..c4c242e 100644 --- a/pom.xml +++ b/pom.xml @@ -99,10 +99,6 @@ io.net5 netty-buffer - - io.net5.incubator - netty-incubator-buffer-memseg - javax.xml.bind jaxb-api @@ -275,11 +271,6 @@ netty-buffer 5.0.0.Final-SNAPSHOT - - io.net5.incubator - netty-incubator-buffer-memseg - 0.0.1.Final-SNAPSHOT - javax.xml.bind jaxb-api diff --git a/src/main/java/it/cavallium/dbengine/client/LuceneIndexImpl.java b/src/main/java/it/cavallium/dbengine/client/LuceneIndexImpl.java index e3d6ef7..f34168b 100644 --- a/src/main/java/it/cavallium/dbengine/client/LuceneIndexImpl.java +++ b/src/main/java/it/cavallium/dbengine/client/LuceneIndexImpl.java @@ -94,7 +94,7 @@ public class LuceneIndexImpl implements LuceneIndex { .fromCallable(signal::key) .map(indicizer::getKey), signal.score())), llSearchResult.totalHitsCount(), - d -> llSearchResult.close() + llSearchResult::close ).send(); }); } @@ -107,7 +107,7 @@ public class LuceneIndexImpl implements LuceneIndex { return new SearchResult<>(llSearchResult.results().map(signal -> { var key = Mono.fromCallable(signal::key).map(indicizer::getKey); return new SearchResultItem<>(key, key.flatMap(valueGetter::get), signal.score()); - }), llSearchResult.totalHitsCount(), d -> llSearchResult.close()).send(); + }), llSearchResult.totalHitsCount(), llSearchResult::close).send(); }); } @@ -131,10 +131,7 @@ public class LuceneIndexImpl implements LuceneIndex { Mono.just(tuple3.getT3().orElseThrow()), tuple3.getT1() )); - return new SearchResult<>(resultItemsFlux, - llSearchResult.totalHitsCount(), - d -> llSearchResult.close() - ).send(); + return new SearchResult<>(resultItemsFlux, llSearchResult.totalHitsCount(), llSearchResult::close).send(); }); } diff --git a/src/main/java/it/cavallium/dbengine/client/SearchResult.java b/src/main/java/it/cavallium/dbengine/client/SearchResult.java index 0a9bec2..6542f2e 100644 --- a/src/main/java/it/cavallium/dbengine/client/SearchResult.java +++ b/src/main/java/it/cavallium/dbengine/client/SearchResult.java @@ -14,18 +14,39 @@ import reactor.core.publisher.Mono; public final class SearchResult extends LiveResourceSupport, SearchResult> { + private static final Drop> DROP = new Drop<>() { + @Override + public void drop(SearchResult obj) { + if (obj.onClose != null) { + obj.onClose.run(); + } + } + + @Override + public Drop> fork() { + return this; + } + + @Override + public void attach(SearchResult obj) { + + } + }; + private Flux> results; private TotalHitsCount totalHitsCount; + private Runnable onClose; - public SearchResult(Flux> results, TotalHitsCount totalHitsCount, - Drop> drop) { - super(drop); + @SuppressWarnings({"unchecked", "rawtypes"}) + public SearchResult(Flux> results, TotalHitsCount totalHitsCount, Runnable onClose) { + super((Drop>) (Drop) DROP); this.results = results; this.totalHitsCount = totalHitsCount; + this.onClose = onClose; } public static SearchResult empty() { - return new SearchResult(Flux.empty(), TotalHitsCount.of(0, true), d -> {}); + return new SearchResult(Flux.empty(), TotalHitsCount.of(0, true), null); } public Flux> results() { @@ -50,11 +71,17 @@ public final class SearchResult extends LiveResourceSupport> prepareSend() { var results = this.results; var totalHitsCount = this.totalHitsCount; - return drop -> new SearchResult<>(results, totalHitsCount, drop); + var onClose = this.onClose; + return drop -> { + var instance = new SearchResult<>(results, totalHitsCount, onClose); + drop.attach(instance); + return instance; + }; } protected void makeInaccessible() { this.results = null; this.totalHitsCount = null; + this.onClose = null; } } diff --git a/src/main/java/it/cavallium/dbengine/client/SearchResultKeys.java b/src/main/java/it/cavallium/dbengine/client/SearchResultKeys.java index 7478031..13f97d3 100644 --- a/src/main/java/it/cavallium/dbengine/client/SearchResultKeys.java +++ b/src/main/java/it/cavallium/dbengine/client/SearchResultKeys.java @@ -2,42 +2,58 @@ package it.cavallium.dbengine.client; import io.net5.buffer.api.Drop; import io.net5.buffer.api.Owned; -import io.net5.buffer.api.internal.ResourceSupport; import it.cavallium.dbengine.client.query.current.data.TotalHitsCount; -import it.cavallium.dbengine.database.LLSearchResultShard; import it.cavallium.dbengine.database.LiveResourceSupport; import it.cavallium.dbengine.database.collections.ValueGetter; -import java.util.Objects; -import org.reactivestreams.Publisher; import org.warp.commonutils.log.Logger; import org.warp.commonutils.log.LoggerFactory; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; @SuppressWarnings("unused") public final class SearchResultKeys extends LiveResourceSupport, SearchResultKeys> { private static final Logger logger = LoggerFactory.getLogger(SearchResultKeys.class); + private static final Drop> DROP = new Drop<>() { + @Override + public void drop(SearchResultKeys obj) { + if (obj.onClose != null) { + obj.onClose.run(); + } + } + + @Override + public Drop> fork() { + return this; + } + + @Override + public void attach(SearchResultKeys obj) { + + } + }; + private Flux> results; private TotalHitsCount totalHitsCount; + private Runnable onClose; - public SearchResultKeys(Flux> results, TotalHitsCount totalHitsCount, - Drop> drop) { - super(drop); + @SuppressWarnings({"unchecked", "rawtypes"}) + public SearchResultKeys(Flux> results, TotalHitsCount totalHitsCount, Runnable onClose) { + super((Drop>) (Drop) DROP); this.results = results; this.totalHitsCount = totalHitsCount; + this.onClose = onClose; } public static SearchResultKeys empty() { - return new SearchResultKeys(Flux.empty(), TotalHitsCount.of(0, true), d -> {}); + return new SearchResultKeys<>(Flux.empty(), TotalHitsCount.of(0, true), null); } public SearchResult withValues(ValueGetter valuesGetter) { return new SearchResult<>(results.map(item -> new SearchResultItem<>(item.key(), item.key().flatMap(valuesGetter::get), item.score() - )), totalHitsCount, d -> this.close()); + )), totalHitsCount, this::close); } public Flux> results() { @@ -62,13 +78,19 @@ public final class SearchResultKeys extends LiveResourceSupport> prepareSend() { var results = this.results; var totalHitsCount = this.totalHitsCount; + var onClose = this.onClose; makeInaccessible(); - return drop -> new SearchResultKeys<>(results, totalHitsCount, drop); + return drop -> { + var instance = new SearchResultKeys<>(results, totalHitsCount, onClose); + drop.attach(instance); + return instance; + }; } protected void makeInaccessible() { this.results = null; this.totalHitsCount = null; + this.onClose = null; } } diff --git a/src/main/java/it/cavallium/dbengine/database/LLDelta.java b/src/main/java/it/cavallium/dbengine/database/LLDelta.java index 458616b..46c9b81 100644 --- a/src/main/java/it/cavallium/dbengine/database/LLDelta.java +++ b/src/main/java/it/cavallium/dbengine/database/LLDelta.java @@ -5,20 +5,66 @@ import io.net5.buffer.api.Drop; import io.net5.buffer.api.Owned; import io.net5.buffer.api.Send; import io.net5.buffer.api.internal.ResourceSupport; +import it.cavallium.dbengine.client.SearchResultKeys; import java.util.StringJoiner; import org.jetbrains.annotations.Nullable; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; public class LLDelta extends LiveResourceSupport { - @Nullable - private final Buffer previous; - @Nullable - private final Buffer current; - private LLDelta(@Nullable Send previous, @Nullable Send current, Drop drop) { - super(new LLDelta.CloseOnDrop(drop)); + private static final Logger logger = LoggerFactory.getLogger(LLDelta.class); + + private static final Drop DROP = new Drop<>() { + @Override + public void drop(LLDelta obj) { + try { + if (obj.previous != null) { + obj.previous.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close previous", ex); + } + try { + if (obj.current != null) { + obj.current.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close current", ex); + } + try { + if (obj.onClose != null) { + obj.onClose.run(); + } + } catch (Throwable ex) { + logger.error("Failed to close onDrop", ex); + } + } + + @Override + public Drop fork() { + return this; + } + + @Override + public void attach(LLDelta obj) { + + } + }; + + @Nullable + private Buffer previous; + @Nullable + private Buffer current; + @Nullable + private Runnable onClose; + + private LLDelta(@Nullable Send previous, @Nullable Send current, @Nullable Runnable onClose) { + super(DROP); assert isAllAccessible(); this.previous = previous != null ? previous.receive().makeReadOnly() : null; this.current = current != null ? current.receive().makeReadOnly() : null; + this.onClose = onClose; } private boolean isAllAccessible() { @@ -31,7 +77,7 @@ public class LLDelta extends LiveResourceSupport { public static LLDelta of(Send min, Send max) { assert (min == null && max == null) || (min != max); - return new LLDelta(min, max, d -> {}); + return new LLDelta(min, max, null); } public Send previous() { @@ -86,42 +132,28 @@ public class LLDelta extends LiveResourceSupport { .toString(); } - public LLDelta copy() { - ensureOwned(); - var prevCopy = previous != null ? previous.copy().send() : null; - Send curCopy = current != null ? current.copy().send() : null; - return new LLDelta(prevCopy, curCopy, d -> {}); - } - @Override protected RuntimeException createResourceClosedException() { return new IllegalStateException("Closed"); } + @Override + protected void makeInaccessible() { + this.current = null; + this.previous = null; + this.onClose = null; + } + @Override protected Owned prepareSend() { Send minSend = this.previous != null ? this.previous.send() : null; Send maxSend = this.current != null ? this.current.send() : null; - return drop -> new LLDelta(minSend, maxSend, drop); + Runnable onClose = this.onClose; + return drop -> { + var instance = new LLDelta(minSend, maxSend, onClose); + drop.attach(instance); + return instance; + }; } - private static class CloseOnDrop implements Drop { - - private final Drop delegate; - - public CloseOnDrop(Drop drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(LLDelta obj) { - if (obj.previous != null) obj.previous.close(); - if (obj.current != null) obj.current.close(); - delegate.drop(obj); - } - } } diff --git a/src/main/java/it/cavallium/dbengine/database/LLEntry.java b/src/main/java/it/cavallium/dbengine/database/LLEntry.java index 5b9a462..7f9fc9b 100644 --- a/src/main/java/it/cavallium/dbengine/database/LLEntry.java +++ b/src/main/java/it/cavallium/dbengine/database/LLEntry.java @@ -5,38 +5,74 @@ import io.net5.buffer.api.Drop; import io.net5.buffer.api.Owned; import io.net5.buffer.api.Send; import io.net5.buffer.api.internal.ResourceSupport; +import java.util.Objects; import java.util.StringJoiner; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; public class LLEntry extends LiveResourceSupport { - @NotNull - private final Buffer key; - @NotNull - private final Buffer value; - private LLEntry(@NotNull Send key, @NotNull Send value, Drop drop) { - super(new LLEntry.CloseOnDrop(drop)); + private static final Logger logger = LoggerFactory.getLogger(LLEntry.class); + + private static final Drop DROP = new Drop<>() { + @Override + public void drop(LLEntry obj) { + try { + if (obj.key != null) { + obj.key.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close key", ex); + } + try { + if (obj.value != null) { + obj.value.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close value", ex); + } + } + + @Override + public Drop fork() { + return this; + } + + @Override + public void attach(LLEntry obj) { + + } + }; + + @Nullable + private Buffer key; + @Nullable + private Buffer value; + + private LLEntry(@NotNull Send key, @NotNull Send value) { + super(DROP); this.key = key.receive().makeReadOnly(); this.value = value.receive().makeReadOnly(); assert isAllAccessible(); } private boolean isAllAccessible() { - assert key.isAccessible(); - assert value.isAccessible(); + assert key != null && key.isAccessible(); + assert value != null && value.isAccessible(); assert this.isAccessible(); assert this.isOwned(); return true; } public static LLEntry of(@NotNull Send key, @NotNull Send value) { - return new LLEntry(key, value, d -> {}); + return new LLEntry(key, value); } public Send getKey() { ensureOwned(); - return key.copy().send(); + return Objects.requireNonNull(key).copy().send(); } public Buffer getKeyUnsafe() { @@ -45,7 +81,7 @@ public class LLEntry extends LiveResourceSupport { public Send getValue() { ensureOwned(); - return value.copy().send(); + return Objects.requireNonNull(value).copy().send(); } @@ -64,6 +100,12 @@ public class LLEntry extends LiveResourceSupport { } } + @Override + protected void makeInaccessible() { + this.key = null; + this.value = null; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -91,11 +133,6 @@ public class LLEntry extends LiveResourceSupport { .toString(); } - public LLEntry copy() { - ensureOwned(); - return new LLEntry(key.copy().send(), value.copy().send(), d -> {}); - } - @Override protected RuntimeException createResourceClosedException() { return new IllegalStateException("Closed"); @@ -105,28 +142,12 @@ public class LLEntry extends LiveResourceSupport { protected Owned prepareSend() { Send keySend; Send valueSend; - keySend = this.key.send(); - valueSend = this.value.send(); - return drop -> new LLEntry(keySend, valueSend, drop); - } - - private static class CloseOnDrop implements Drop { - - private final Drop delegate; - - public CloseOnDrop(Drop drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(LLEntry obj) { - obj.key.close(); - obj.value.close(); - delegate.drop(obj); - } + keySend = Objects.requireNonNull(this.key).send(); + valueSend = Objects.requireNonNull(this.value).send(); + return drop -> { + var instance = new LLEntry(keySend, valueSend); + drop.attach(instance); + return instance; + }; } } diff --git a/src/main/java/it/cavallium/dbengine/database/LLRange.java b/src/main/java/it/cavallium/dbengine/database/LLRange.java index 6602b59..c45dc27 100644 --- a/src/main/java/it/cavallium/dbengine/database/LLRange.java +++ b/src/main/java/it/cavallium/dbengine/database/LLRange.java @@ -9,13 +9,54 @@ import io.net5.buffer.api.Send; import io.net5.buffer.api.internal.ResourceSupport; import java.util.StringJoiner; import org.jetbrains.annotations.Nullable; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; /** * Range of data, from min (inclusive),to max (exclusive) */ public class LLRange extends LiveResourceSupport { - private static final LLRange RANGE_ALL = new LLRange(null, null, null, d -> {}); + private static final Logger logger = LoggerFactory.getLogger(LLRange.class); + + private static final Drop DROP = new Drop<>() { + @Override + public void drop(LLRange obj) { + try { + if (obj.min != null) { + obj.min.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close min", ex); + } + try { + if (obj.max != null) { + obj.max.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close max", ex); + } + try { + if (obj.single != null) { + obj.single.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close single", ex); + } + } + + @Override + public Drop fork() { + return this; + } + + @Override + public void attach(LLRange obj) { + + } + }; + + private static final LLRange RANGE_ALL = new LLRange(null, null, null); @Nullable private Buffer min; @Nullable @@ -23,8 +64,8 @@ public class LLRange extends LiveResourceSupport { @Nullable private Buffer single; - private LLRange(Send min, Send max, Send single, Drop drop) { - super(new CloseOnDrop(drop)); + private LLRange(Send min, Send max, Send single) { + super(DROP); assert isAllAccessible(); assert single == null || (min == null && max == null); this.min = min != null ? min.receive().makeReadOnly() : null; @@ -46,19 +87,19 @@ public class LLRange extends LiveResourceSupport { } public static LLRange from(Send min) { - return new LLRange(min, null, null, d -> {}); + return new LLRange(min, null, null); } public static LLRange to(Send max) { - return new LLRange(null, max, null, d -> {}); + return new LLRange(null, max, null); } public static LLRange single(Send single) { - return new LLRange(null, null, single, d -> {}); + return new LLRange(null, null, single); } public static LLRange of(Send min, Send max) { - return new LLRange(min, max, null, d -> {}); + return new LLRange(min, max, null); } public boolean isAll() { @@ -179,8 +220,7 @@ public class LLRange extends LiveResourceSupport { ensureOwned(); return new LLRange(min != null ? min.copy().send() : null, max != null ? max.copy().send() : null, - single != null ? single.copy().send(): null, - d -> {} + single != null ? single.copy().send(): null ); } @@ -197,7 +237,11 @@ public class LLRange extends LiveResourceSupport { minSend = this.min != null ? this.min.send() : null; maxSend = this.max != null ? this.max.send() : null; singleSend = this.single != null ? this.single.send() : null; - return drop -> new LLRange(minSend, maxSend, singleSend, drop); + return drop -> { + var instance = new LLRange(minSend, maxSend, singleSend); + drop.attach(instance); + return instance; + }; } protected void makeInaccessible() { @@ -205,25 +249,4 @@ public class LLRange extends LiveResourceSupport { this.max = null; this.single = null; } - - private static class CloseOnDrop implements Drop { - - private final Drop delegate; - - public CloseOnDrop(Drop drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(LLRange obj) { - if (obj.min != null) obj.min.close(); - if (obj.max != null) obj.max.close(); - if (obj.single != null) obj.single.close(); - delegate.drop(obj); - } - } } diff --git a/src/main/java/it/cavallium/dbengine/database/LLSearchResultShard.java b/src/main/java/it/cavallium/dbengine/database/LLSearchResultShard.java index 2ffdbfc..34c3fbf 100644 --- a/src/main/java/it/cavallium/dbengine/database/LLSearchResultShard.java +++ b/src/main/java/it/cavallium/dbengine/database/LLSearchResultShard.java @@ -13,13 +13,38 @@ public final class LLSearchResultShard extends LiveResourceSupport DROP = new Drop<>() { + @Override + public void drop(LLSearchResultShard obj) { + try { + if (obj.onClose != null) { + obj.onClose.run(); + } + } catch (Throwable ex) { + logger.error("Failed to close onClose", ex); + } + } + + @Override + public Drop fork() { + return this; + } + + @Override + public void attach(LLSearchResultShard obj) { + + } + }; + private Flux results; private TotalHitsCount totalHitsCount; + private Runnable onClose; - public LLSearchResultShard(Flux results, TotalHitsCount totalHitsCount, Drop drop) { - super(drop); + public LLSearchResultShard(Flux results, TotalHitsCount totalHitsCount, Runnable onClose) { + super(DROP); this.results = results; this.totalHitsCount = totalHitsCount; + this.onClose = onClose; } public Flux results() { @@ -65,11 +90,13 @@ public final class LLSearchResultShard extends LiveResourceSupport prepareSend() { var results = this.results; var totalHitsCount = this.totalHitsCount; - return drop -> new LLSearchResultShard(results, totalHitsCount, drop); + var onClose = this.onClose; + return drop -> new LLSearchResultShard(results, totalHitsCount, onClose); } protected void makeInaccessible() { this.results = null; this.totalHitsCount = null; + this.onClose = null; } } diff --git a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseEmpty.java b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseEmpty.java index cdd9dbd..1d9b9ea 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseEmpty.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseEmpty.java @@ -36,10 +36,8 @@ public class DatabaseEmpty { private DatabaseEmpty() { } - public static DatabaseStageEntry create(LLDictionary dictionary, - Send key, - Drop> drop) { - return new DatabaseSingle<>(dictionary, key, nothingSerializer(dictionary.getAllocator()), drop); + public static DatabaseStageEntry create(LLDictionary dictionary, Send key, Runnable onClose) { + return new DatabaseSingle<>(dictionary, key, nothingSerializer(dictionary.getAllocator()), onClose); } public static final class Nothing { diff --git a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionary.java b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionary.java index 6c56ecd..c69537c 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionary.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionary.java @@ -44,26 +44,26 @@ public class DatabaseMapDictionary extends DatabaseMapDictionaryDeep prefixKey, SerializerFixedBinaryLength keySuffixSerializer, Serializer valueSerializer, - Drop>> drop) { + Runnable onClose) { // Do not retain or release or use the prefixKey here - super(dictionary, prefixKey, keySuffixSerializer, new SubStageGetterSingle<>(valueSerializer), 0, drop); + super(dictionary, prefixKey, keySuffixSerializer, new SubStageGetterSingle<>(valueSerializer), 0, onClose); this.valueSerializer = valueSerializer; } public static DatabaseMapDictionary simple(LLDictionary dictionary, SerializerFixedBinaryLength keySerializer, Serializer valueSerializer, - Drop>> drop) { + Runnable onClose) { return new DatabaseMapDictionary<>(dictionary, LLUtils.empty(dictionary.getAllocator()), keySerializer, - valueSerializer, drop); + valueSerializer, onClose); } public static DatabaseMapDictionary tail(LLDictionary dictionary, Send prefixKey, SerializerFixedBinaryLength keySuffixSerializer, Serializer valueSerializer, - Drop>> drop) { - return new DatabaseMapDictionary<>(dictionary, prefixKey, keySuffixSerializer, valueSerializer, drop); + Runnable onClose) { + return new DatabaseMapDictionary<>(dictionary, prefixKey, keySuffixSerializer, valueSerializer, onClose); } private Send toKey(Send suffixKeyToSend) { @@ -152,7 +152,7 @@ public class DatabaseMapDictionary extends DatabaseMapDictionaryDeep> at(@Nullable CompositeSnapshot snapshot, T keySuffix) { return Mono.fromCallable(() -> - new DatabaseSingle<>(dictionary, toKey(serializeSuffix(keySuffix)), valueSerializer, d -> {})); + new DatabaseSingle<>(dictionary, toKey(serializeSuffix(keySuffix)), valueSerializer, null)); } @Override @@ -401,7 +401,7 @@ public class DatabaseMapDictionary extends DatabaseMapDictionaryDeep(dictionary, toKey(keyBuf.send()), valueSerializer, d -> {}) + new DatabaseSingle<>(dictionary, toKey(keyBuf.send()), valueSerializer, null) )); } catch (SerializationException ex) { sink.error(ex); diff --git a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryDeep.java b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryDeep.java index 1642944..336f759 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryDeep.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryDeep.java @@ -25,6 +25,8 @@ import java.util.Map.Entry; import java.util.Objects; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -33,6 +35,45 @@ public class DatabaseMapDictionaryDeep> extend LiveResourceSupport>, DatabaseMapDictionaryDeep> implements DatabaseStageMap { + private static final Logger logger = LoggerFactory.getLogger(DatabaseMapDictionaryDeep.class); + + private static final Drop> DROP = new Drop<>() { + @Override + public void drop(DatabaseMapDictionaryDeep obj) { + try { + if (obj.range != null) { + obj.range.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close range", ex); + } + try { + if (obj.keyPrefix != null) { + obj.keyPrefix.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close keyPrefix", ex); + } + try { + if (obj.onClose != null) { + obj.onClose.run(); + } + } catch (Throwable ex) { + logger.error("Failed to close onClose", ex); + } + } + + @Override + public Drop> fork() { + return this; + } + + @Override + public void attach(DatabaseMapDictionaryDeep obj) { + + } + }; + protected final LLDictionary dictionary; private final BufferAllocator alloc; protected final SubStageGetter subStageGetter; @@ -44,6 +85,7 @@ public class DatabaseMapDictionaryDeep> extend protected LLRange range; protected Buffer keyPrefix; + protected Runnable onClose; private static void incrementPrefix(Buffer prefix, int prefixLength) { assert prefix.readableBytes() >= prefixLength; @@ -151,29 +193,30 @@ public class DatabaseMapDictionaryDeep> extend @Deprecated public static DatabaseMapDictionaryDeep> simple(LLDictionary dictionary, SerializerFixedBinaryLength keySerializer, SubStageGetterSingle subStageGetter, - Drop>> drop) { + Runnable onClose) { return new DatabaseMapDictionaryDeep<>(dictionary, LLUtils.empty(dictionary.getAllocator()), keySerializer, - subStageGetter, 0, drop); + subStageGetter, 0, onClose); } public static > DatabaseMapDictionaryDeep deepTail( LLDictionary dictionary, SerializerFixedBinaryLength keySerializer, int keyExtLength, - SubStageGetter subStageGetter, Drop> drop) { + SubStageGetter subStageGetter, Runnable onClose) { return new DatabaseMapDictionaryDeep<>(dictionary, LLUtils.empty(dictionary.getAllocator()), keySerializer, - subStageGetter, keyExtLength, drop); + subStageGetter, keyExtLength, onClose); } public static > DatabaseMapDictionaryDeep deepIntermediate( LLDictionary dictionary, Send prefixKey, SerializerFixedBinaryLength keySuffixSerializer, - SubStageGetter subStageGetter, int keyExtLength, Drop> drop) { + SubStageGetter subStageGetter, int keyExtLength, Runnable onClose) { return new DatabaseMapDictionaryDeep<>(dictionary, prefixKey, keySuffixSerializer, subStageGetter, - keyExtLength, drop); + keyExtLength, onClose); } + @SuppressWarnings({"unchecked", "rawtypes"}) protected DatabaseMapDictionaryDeep(LLDictionary dictionary, @NotNull Send prefixKeyToReceive, SerializerFixedBinaryLength keySuffixSerializer, SubStageGetter subStageGetter, int keyExtLength, - Drop> drop) { - super(new CloseOnDrop<>(drop)); + Runnable onClose) { + super((Drop>) (Drop) DROP); try (var prefixKey = prefixKeyToReceive.receive()) { this.dictionary = dictionary; this.alloc = dictionary.getAllocator(); @@ -198,9 +241,11 @@ public class DatabaseMapDictionaryDeep> extend } this.keyPrefix = prefixKey.send().receive(); + this.onClose = onClose; } } + @SuppressWarnings({"unchecked", "rawtypes"}) private DatabaseMapDictionaryDeep(LLDictionary dictionary, BufferAllocator alloc, SubStageGetter subStageGetter, @@ -211,8 +256,8 @@ public class DatabaseMapDictionaryDeep> extend Mono> rangeMono, Send range, Send keyPrefix, - Drop> drop) { - super(new CloseOnDrop<>(drop)); + Runnable onClose) { + super((Drop>) (Drop) DROP); this.dictionary = dictionary; this.alloc = alloc; this.subStageGetter = subStageGetter; @@ -224,6 +269,7 @@ public class DatabaseMapDictionaryDeep> extend this.range = range.receive(); this.keyPrefix = keyPrefix.receive(); + this.onClose = onClose; } @SuppressWarnings("unused") @@ -418,34 +464,29 @@ public class DatabaseMapDictionaryDeep> extend protected Owned> prepareSend() { var keyPrefix = this.keyPrefix.send(); var range = this.range.send(); - return drop -> new DatabaseMapDictionaryDeep<>(dictionary, alloc, subStageGetter, keySuffixSerializer, - keyPrefixLength, keySuffixLength, keyExtLength, rangeMono, range, keyPrefix, drop); + var onClose = this.onClose; + return drop -> { + var instance = new DatabaseMapDictionaryDeep<>(dictionary, + alloc, + subStageGetter, + keySuffixSerializer, + keyPrefixLength, + keySuffixLength, + keyExtLength, + rangeMono, + range, + keyPrefix, + onClose + ); + drop.attach(instance); + return instance; + }; } @Override protected void makeInaccessible() { this.keyPrefix = null; this.range = null; - } - - private static class CloseOnDrop> - implements Drop> { - - private final Drop> delegate; - - public CloseOnDrop(Drop> drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(DatabaseMapDictionaryDeep obj) { - obj.range.close(); - obj.keyPrefix.close(); - delegate.drop(obj); - } + this.onClose = null; } } diff --git a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryHashed.java b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryHashed.java index 57ab208..9c9437a 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryHashed.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryHashed.java @@ -25,6 +25,8 @@ import java.util.Set; import java.util.function.Function; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -33,19 +35,45 @@ public class DatabaseMapDictionaryHashed extends LiveResourceSupport>, DatabaseMapDictionaryHashed> implements DatabaseStageMap> { + private static final Logger logger = LoggerFactory.getLogger(DatabaseMapDictionaryHashed.class); + + private static final Drop> DROP = new Drop<>() { + @Override + public void drop(DatabaseMapDictionaryHashed obj) { + try { + if (obj.subDictionary != null) { + obj.subDictionary.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close subDictionary", ex); + } + } + + @Override + public Drop> fork() { + return this; + } + + @Override + public void attach(DatabaseMapDictionaryHashed obj) { + + } + }; + private final BufferAllocator alloc; private final Function keySuffixHashFunction; private DatabaseMapDictionary>> subDictionary; + @SuppressWarnings({"unchecked", "rawtypes"}) protected DatabaseMapDictionaryHashed(LLDictionary dictionary, @NotNull Send prefixKey, Serializer keySuffixSerializer, Serializer valueSerializer, Function keySuffixHashFunction, SerializerFixedBinaryLength keySuffixHashSerializer, - Drop> drop) { - super(new DatabaseMapDictionaryHashed.CloseOnDrop<>(drop)); + Runnable onClose) { + super((Drop>) (Drop) DROP); if (dictionary.getUpdateMode().block() != UpdateMode.ALLOW) { throw new IllegalArgumentException("Hashed maps only works when UpdateMode is ALLOW"); } @@ -55,15 +83,16 @@ public class DatabaseMapDictionaryHashed extends ValuesSetSerializer> valuesSetSerializer = new ValuesSetSerializer<>(alloc, valueWithHashSerializer); this.subDictionary = DatabaseMapDictionary.tail(dictionary, prefixKey, keySuffixHashSerializer, - valuesSetSerializer, d -> {}); + valuesSetSerializer, onClose); this.keySuffixHashFunction = keySuffixHashFunction; } + @SuppressWarnings({"unchecked", "rawtypes"}) private DatabaseMapDictionaryHashed(BufferAllocator alloc, Function keySuffixHashFunction, Send>>>> subDictionary, Drop> drop) { - super(new CloseOnDrop<>(drop)); + super((Drop>) (Drop) DROP); this.alloc = alloc; this.keySuffixHashFunction = keySuffixHashFunction; @@ -75,7 +104,7 @@ public class DatabaseMapDictionaryHashed extends Serializer valueSerializer, Function keyHashFunction, SerializerFixedBinaryLength keyHashSerializer, - Drop> drop) { + Runnable onClose) { return new DatabaseMapDictionaryHashed<>( dictionary, LLUtils.empty(dictionary.getAllocator()), @@ -83,7 +112,7 @@ public class DatabaseMapDictionaryHashed extends valueSerializer, keyHashFunction, keyHashSerializer, - drop + onClose ); } @@ -93,14 +122,14 @@ public class DatabaseMapDictionaryHashed extends Serializer valueSerializer, Function keySuffixHashFunction, SerializerFixedBinaryLength keySuffixHashSerializer, - Drop> drop) { + Runnable onClose) { return new DatabaseMapDictionaryHashed<>(dictionary, prefixKey, keySuffixSerializer, valueSerializer, keySuffixHashFunction, keySuffixHashSerializer, - drop + onClose ); } @@ -173,7 +202,7 @@ public class DatabaseMapDictionaryHashed extends private Mono> atPrivate(@Nullable CompositeSnapshot snapshot, T key, TH hash) { return subDictionary .at(snapshot, hash) - .map(entry -> new DatabaseSingleBucket(entry, key, d -> {})) + .map(entry -> new DatabaseSingleBucket(entry, key, null)) .doOnDiscard(Resource.class, Resource::close); } @@ -325,23 +354,4 @@ public class DatabaseMapDictionaryHashed extends protected void makeInaccessible() { this.subDictionary = null; } - - private static class CloseOnDrop implements Drop> { - - private final Drop> delegate; - - public CloseOnDrop(Drop> drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(DatabaseMapDictionaryHashed obj) { - obj.subDictionary.close(); - delegate.drop(obj); - } - } } diff --git a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionary.java b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionary.java index c082547..2df8d86 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionary.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionary.java @@ -20,21 +20,21 @@ public class DatabaseSetDictionary extends DatabaseMapDictionary protected DatabaseSetDictionary(LLDictionary dictionary, Send prefixKey, SerializerFixedBinaryLength keySuffixSerializer, - Drop>> drop) { - super(dictionary, prefixKey, keySuffixSerializer, DatabaseEmpty.nothingSerializer(dictionary.getAllocator()), drop); + Runnable onClose) { + super(dictionary, prefixKey, keySuffixSerializer, DatabaseEmpty.nothingSerializer(dictionary.getAllocator()), onClose); } public static DatabaseSetDictionary simple(LLDictionary dictionary, SerializerFixedBinaryLength keySerializer, - Drop>> drop) { - return new DatabaseSetDictionary<>(dictionary, LLUtils.empty(dictionary.getAllocator()), keySerializer, drop); + Runnable onClose) { + return new DatabaseSetDictionary<>(dictionary, LLUtils.empty(dictionary.getAllocator()), keySerializer, onClose); } public static DatabaseSetDictionary tail(LLDictionary dictionary, Send prefixKey, SerializerFixedBinaryLength keySuffixSerializer, - Drop>> drop) { - return new DatabaseSetDictionary<>(dictionary, prefixKey, keySuffixSerializer, drop); + Runnable onClose) { + return new DatabaseSetDictionary<>(dictionary, prefixKey, keySuffixSerializer, onClose); } public Mono> getKeySet(@Nullable CompositeSnapshot snapshot) { diff --git a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionaryHashed.java b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionaryHashed.java index 68bf230..89f944b 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionaryHashed.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionaryHashed.java @@ -25,14 +25,14 @@ public class DatabaseSetDictionaryHashed extends DatabaseMapDictionaryHas Serializer keySuffixSerializer, Function keySuffixHashFunction, SerializerFixedBinaryLength keySuffixHashSerializer, - Drop> drop) { + Runnable onClose) { super(dictionary, prefixKey, keySuffixSerializer, DatabaseEmpty.nothingSerializer(dictionary.getAllocator()), keySuffixHashFunction, keySuffixHashSerializer, - drop + onClose ); } @@ -40,13 +40,13 @@ public class DatabaseSetDictionaryHashed extends DatabaseMapDictionaryHas Serializer keySerializer, Function keyHashFunction, SerializerFixedBinaryLength keyHashSerializer, - Drop> drop) { + Runnable onClose) { return new DatabaseSetDictionaryHashed<>(dictionary, LLUtils.empty(dictionary.getAllocator()), keySerializer, keyHashFunction, keyHashSerializer, - drop + onClose ); } @@ -54,13 +54,14 @@ public class DatabaseSetDictionaryHashed extends DatabaseMapDictionaryHas Send prefixKey, Serializer keySuffixSerializer, Function keyHashFunction, - SerializerFixedBinaryLength keyHashSerializer, Drop> drop) { + SerializerFixedBinaryLength keyHashSerializer, + Runnable onClose) { return new DatabaseSetDictionaryHashed<>(dictionary, prefixKey, keySuffixSerializer, keyHashFunction, keyHashSerializer, - drop + onClose ); } diff --git a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingle.java b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingle.java index 9a00456..40efd77 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingle.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingle.java @@ -7,6 +7,7 @@ import io.net5.buffer.api.Send; import io.net5.buffer.api.internal.ResourceSupport; import it.cavallium.dbengine.client.BadBlock; import it.cavallium.dbengine.client.CompositeSnapshot; +import it.cavallium.dbengine.client.SearchResultKeys; import it.cavallium.dbengine.database.Delta; import it.cavallium.dbengine.database.LLDictionary; import it.cavallium.dbengine.database.LLDictionaryResultType; @@ -19,6 +20,8 @@ import it.cavallium.dbengine.database.serialization.SerializationException; import it.cavallium.dbengine.database.serialization.SerializationFunction; import it.cavallium.dbengine.database.serialization.Serializer; import org.jetbrains.annotations.Nullable; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.SynchronousSink; @@ -26,20 +29,49 @@ import reactor.core.publisher.SynchronousSink; public class DatabaseSingle extends ResourceSupport, DatabaseSingle> implements DatabaseStageEntry { + private static final Logger logger = LoggerFactory.getLogger(DatabaseSingle.class); + + private static final Drop> DROP = new Drop<>() { + @Override + public void drop(DatabaseSingle obj) { + try { + obj.key.close(); + } catch (Throwable ex) { + logger.error("Failed to close key", ex); + } + if (obj.onClose != null) { + obj.onClose.run(); + } + } + + @Override + public Drop> fork() { + return this; + } + + @Override + public void attach(DatabaseSingle obj) { + + } + }; + private final LLDictionary dictionary; private final Mono> keyMono; private final Serializer serializer; private Buffer key; + private Runnable onClose; + @SuppressWarnings({"unchecked", "rawtypes"}) public DatabaseSingle(LLDictionary dictionary, Send key, Serializer serializer, - Drop> drop) { - super(new CloseOnDrop<>(drop)); + Runnable onClose) { + super((Drop>) (Drop) DROP); try (key) { this.dictionary = dictionary; this.key = key.receive(); this.keyMono = LLUtils.lazyRetain(this.key); this.serializer = serializer; + this.onClose = onClose; } } @@ -144,30 +176,17 @@ public class DatabaseSingle extends ResourceSupport, Databas @Override protected Owned> prepareSend() { var key = this.key.send(); - return drop -> new DatabaseSingle<>(dictionary, key, serializer, drop); + var onClose = this.onClose; + return drop -> { + var instance = new DatabaseSingle<>(dictionary, key, serializer, onClose); + drop.attach(instance); + return instance; + }; } @Override protected void makeInaccessible() { this.key = null; - } - - private static class CloseOnDrop implements Drop> { - - private final Drop> delegate; - - public CloseOnDrop(Drop> drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(DatabaseSingle obj) { - obj.key.close(); - delegate.drop(obj); - } + this.onClose = null; } } \ No newline at end of file diff --git a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleBucket.java b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleBucket.java index 03a67d4..dd6b0c4 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleBucket.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleBucket.java @@ -24,6 +24,8 @@ import java.util.function.Function; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.warp.commonutils.functional.TriFunction; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -32,22 +34,58 @@ public class DatabaseSingleBucket extends LiveResourceSupport, DatabaseSingleBucket> implements DatabaseStageEntry { + private static final Logger logger = LoggerFactory.getLogger(DatabaseSingleBucket.class); + + private static final Drop> DROP = new Drop<>() { + @Override + public void drop(DatabaseSingleBucket obj) { + try { + if (obj.bucketStage != null) { + obj.bucketStage.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close bucketStage", ex); + } + try { + if (obj.onClose != null) { + obj.onClose.run(); + } + } catch (Throwable ex) { + logger.error("Failed to close onClose", ex); + } + } + + @Override + public Drop> fork() { + return this; + } + + @Override + public void attach(DatabaseSingleBucket obj) { + + } + }; + private final K key; private DatabaseStageEntry>> bucketStage; - public DatabaseSingleBucket(DatabaseStageEntry>> bucketStage, K key, - Drop> drop) { - super(new CloseOnDrop<>(drop)); + private Runnable onClose; + + @SuppressWarnings({"unchecked", "rawtypes"}) + public DatabaseSingleBucket(DatabaseStageEntry>> bucketStage, K key, Runnable onClose) { + super((Drop>) (Drop) DROP); this.key = key; this.bucketStage = bucketStage; + this.onClose = onClose; } - private DatabaseSingleBucket(Send>>> bucketStage, K key, - Drop> drop) { - super(new CloseOnDrop<>(drop)); + @SuppressWarnings({"unchecked", "rawtypes"}) + private DatabaseSingleBucket(Send>>> bucketStage, K key, Runnable onClose) { + super((Drop>) (Drop) DROP); this.key = key; this.bucketStage = (DatabaseStageEntry>>) bucketStage.receive(); + this.onClose = onClose; } @Override @@ -210,31 +248,17 @@ public class DatabaseSingleBucket @Override protected Owned> prepareSend() { var bucketStage = this.bucketStage.send(); - return drop -> new DatabaseSingleBucket<>(bucketStage, key, drop); + var onClose = this.onClose; + return drop -> { + var instance = new DatabaseSingleBucket(bucketStage, key, onClose); + drop.attach(instance); + return instance; + }; } @Override protected void makeInaccessible() { this.bucketStage = null; - } - - private static class CloseOnDrop implements - Drop> { - - private final Drop> delegate; - - public CloseOnDrop(Drop> drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(DatabaseSingleBucket obj) { - obj.bucketStage.close(); - delegate.drop(obj); - } + this.onClose = null; } } diff --git a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleMapped.java b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleMapped.java index 9b15b52..af62ef8 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleMapped.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleMapped.java @@ -14,6 +14,8 @@ import it.cavallium.dbengine.database.UpdateReturnMode; import it.cavallium.dbengine.database.serialization.SerializationException; import it.cavallium.dbengine.database.serialization.SerializationFunction; import org.jetbrains.annotations.Nullable; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.SynchronousSink; @@ -22,20 +24,47 @@ import reactor.core.publisher.SynchronousSink; public class DatabaseSingleMapped extends ResourceSupport, DatabaseSingleMapped> implements DatabaseStageEntry { + private static final Logger logger = LoggerFactory.getLogger(DatabaseSingleMapped.class); + + private static final Drop> DROP = new Drop<>() { + @Override + public void drop(DatabaseSingleMapped obj) { + try { + if (obj.serializedSingle != null) { + obj.serializedSingle.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close serializedSingle", ex); + } + } + + @Override + public Drop> fork() { + return this; + } + + @Override + public void attach(DatabaseSingleMapped obj) { + + } + }; + private final Mapper mapper; private DatabaseStageEntry serializedSingle; + @SuppressWarnings({"unchecked", "rawtypes"}) public DatabaseSingleMapped(DatabaseStageEntry serializedSingle, Mapper mapper, Drop> drop) { - super(new CloseOnDrop<>(drop)); + super((Drop>) (Drop) DROP); this.serializedSingle = serializedSingle; this.mapper = mapper; } + @SuppressWarnings({"unchecked", "rawtypes"}) private DatabaseSingleMapped(Send> serializedSingle, Mapper mapper, Drop> drop) { - super(new CloseOnDrop<>(drop)); + super((Drop>) (Drop) DROP); this.mapper = mapper; this.serializedSingle = (DatabaseStageEntry) serializedSingle.receive(); @@ -170,22 +199,4 @@ public class DatabaseSingleMapped extends ResourceSupport this.serializedSingle = null; } - private static class CloseOnDrop implements Drop> { - - private final Drop> delegate; - - public CloseOnDrop(Drop> drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(DatabaseSingleMapped obj) { - obj.serializedSingle.close(); - delegate.drop(obj); - } - } } diff --git a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashMap.java b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashMap.java index 38da00b..0365da2 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashMap.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashMap.java @@ -37,7 +37,7 @@ public class SubStageGetterHashMap implements Mono> prefixKeyMono) { return LLUtils.usingSend(prefixKeyMono, prefixKey -> Mono.just(DatabaseMapDictionaryHashed .tail(dictionary, prefixKey, keySerializer, valueSerializer, keyHashFunction, - keyHashSerializer, d -> {})), true); + keyHashSerializer, null)), true); } public int getKeyHashBinaryLength() { diff --git a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashSet.java b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashSet.java index 39c3279..09f36e0 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashSet.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashSet.java @@ -35,7 +35,7 @@ public class SubStageGetterHashSet implements return Mono.usingWhen(prefixKeyMono, prefixKey -> Mono .fromSupplier(() -> DatabaseSetDictionaryHashed.tail(dictionary, prefixKey, keySerializer, - keyHashFunction, keyHashSerializer, d -> {}) + keyHashFunction, keyHashSerializer, null) ), prefixKey -> Mono.fromRunnable(prefixKey::close) ); diff --git a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMap.java b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMap.java index 772f417..39f17a6 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMap.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMap.java @@ -28,7 +28,7 @@ public class SubStageGetterMap implements SubStageGetter, Databa Mono> prefixKeyMono) { return LLUtils.usingSend(prefixKeyMono, prefixKey -> Mono.fromSupplier(() -> DatabaseMapDictionary - .tail(dictionary, prefixKey, keySerializer, valueSerializer, d -> {})), true); + .tail(dictionary, prefixKey, keySerializer, valueSerializer, null)), true); } public int getKeyBinaryLength() { diff --git a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMapDeep.java b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMapDeep.java index 34e25c5..d07d11d 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMapDeep.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMapDeep.java @@ -42,8 +42,7 @@ public class SubStageGetterMapDeep> implements @Nullable CompositeSnapshot snapshot, Mono> prefixKeyMono) { return LLUtils.usingSend(prefixKeyMono, prefixKey -> Mono.just(DatabaseMapDictionaryDeep - .deepIntermediate(dictionary, prefixKey, keySerializer, subStageGetter, keyExtLength, - d -> {})), true); + .deepIntermediate(dictionary, prefixKey, keySerializer, subStageGetter, keyExtLength, null)), true); } public int getKeyBinaryLength() { diff --git a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSet.java b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSet.java index 6b6260c..6d4eed0 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSet.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSet.java @@ -24,7 +24,7 @@ public class SubStageGetterSet implements SubStageGetter, Dat Mono> prefixKeyMono) { return Mono.usingWhen(prefixKeyMono, prefixKey -> Mono - .fromSupplier(() -> DatabaseSetDictionary.tail(dictionary, prefixKey, keySerializer, d -> {})), + .fromSupplier(() -> DatabaseSetDictionary.tail(dictionary, prefixKey, keySerializer, null)), prefixKey -> Mono.fromRunnable(prefixKey::close) ); } diff --git a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSingle.java b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSingle.java index 533b4a6..7860c9c 100644 --- a/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSingle.java +++ b/src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSingle.java @@ -20,7 +20,7 @@ public class SubStageGetterSingle implements SubStageGetter> subStage(LLDictionary dictionary, @Nullable CompositeSnapshot snapshot, Mono> keyPrefixMono) { - return keyPrefixMono.map(keyPrefix -> new DatabaseSingle<>(dictionary, keyPrefix, serializer, d -> {})); + return keyPrefixMono.map(keyPrefix -> new DatabaseSingle<>(dictionary, keyPrefix, serializer, null)); } } diff --git a/src/main/java/it/cavallium/dbengine/database/disk/CachedIndexSearcherManager.java b/src/main/java/it/cavallium/dbengine/database/disk/CachedIndexSearcherManager.java index aa5992d..dbe0185 100644 --- a/src/main/java/it/cavallium/dbengine/database/disk/CachedIndexSearcherManager.java +++ b/src/main/java/it/cavallium/dbengine/database/disk/CachedIndexSearcherManager.java @@ -152,7 +152,7 @@ public class CachedIndexSearcherManager implements IndexSearcherManager { }); } - private void dropCachedIndexSearcher(LLIndexSearcher cachedIndexSearcher) { + private void dropCachedIndexSearcher() { // This shouldn't happen more than once per searcher. activeSearchers.arriveAndDeregister(); } diff --git a/src/main/java/it/cavallium/dbengine/database/disk/LLIndexSearcher.java b/src/main/java/it/cavallium/dbengine/database/disk/LLIndexSearcher.java index 4259e08..35daa21 100644 --- a/src/main/java/it/cavallium/dbengine/database/disk/LLIndexSearcher.java +++ b/src/main/java/it/cavallium/dbengine/database/disk/LLIndexSearcher.java @@ -3,18 +3,49 @@ package it.cavallium.dbengine.database.disk; import io.net5.buffer.api.Drop; import io.net5.buffer.api.Owned; import it.cavallium.dbengine.database.LiveResourceSupport; +import it.cavallium.dbengine.database.collections.DatabaseMapDictionaryHashed; import org.apache.lucene.index.IndexReader; import org.apache.lucene.search.IndexSearcher; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; public class LLIndexSearcher extends LiveResourceSupport { + private static final Logger logger = LoggerFactory.getLogger(LLIndexSearcher.class); + + private static final Drop DROP = new Drop<>() { + @Override + public void drop(LLIndexSearcher obj) { + try { + if (obj.onClose != null) { + obj.onClose.run(); + } + } catch (Throwable ex) { + logger.error("Failed to close onClose", ex); + } + } + + @Override + public Drop fork() { + return this; + } + + @Override + public void attach(LLIndexSearcher obj) { + + } + }; + private IndexSearcher indexSearcher; private final boolean decRef; - public LLIndexSearcher(IndexSearcher indexSearcher, boolean decRef, Drop drop) { - super(drop); + private Runnable onClose; + + public LLIndexSearcher(IndexSearcher indexSearcher, boolean decRef, Runnable onClose) { + super(DROP); this.indexSearcher = indexSearcher; this.decRef = decRef; + this.onClose = onClose; } public IndexReader getIndexReader() { @@ -39,11 +70,13 @@ public class LLIndexSearcher extends LiveResourceSupport prepareSend() { var indexSearcher = this.indexSearcher; - return drop -> new LLIndexSearcher(indexSearcher, decRef, drop); + var onClose = this.onClose; + return drop -> new LLIndexSearcher(indexSearcher, decRef, onClose); } protected void makeInaccessible() { this.indexSearcher = null; + this.onClose = null; } } diff --git a/src/main/java/it/cavallium/dbengine/database/disk/LLIndexSearchers.java b/src/main/java/it/cavallium/dbengine/database/disk/LLIndexSearchers.java index 2cb8de9..c21af59 100644 --- a/src/main/java/it/cavallium/dbengine/database/disk/LLIndexSearchers.java +++ b/src/main/java/it/cavallium/dbengine/database/disk/LLIndexSearchers.java @@ -6,6 +6,7 @@ import io.net5.buffer.api.Resource; import io.net5.buffer.api.Send; import io.net5.buffer.api.internal.ResourceSupport; import it.cavallium.dbengine.database.LLEntry; +import it.cavallium.dbengine.database.LLSearchResultShard; import it.cavallium.dbengine.database.LiveResourceSupport; import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap; import java.io.IOException; @@ -22,15 +23,17 @@ import org.apache.lucene.index.MultiReader; import org.apache.lucene.index.StoredFieldVisitor; import org.apache.lucene.index.Term; import org.apache.lucene.search.IndexSearcher; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; public interface LLIndexSearchers extends Resource { static LLIndexSearchers of(List> indexSearchers) { - return new ShardedIndexSearchers(indexSearchers, d -> {}); + return new ShardedIndexSearchers(indexSearchers, null); } static UnshardedIndexSearchers unsharded(Send indexSearcher) { - return new UnshardedIndexSearchers(indexSearcher, d -> {}); + return new UnshardedIndexSearchers(indexSearcher, null); } List shards(); @@ -42,11 +45,45 @@ public interface LLIndexSearchers extends Resource { class UnshardedIndexSearchers extends LiveResourceSupport implements LLIndexSearchers { - private LLIndexSearcher indexSearcher; + private static final Logger logger = LoggerFactory.getLogger(UnshardedIndexSearchers.class); - public UnshardedIndexSearchers(Send indexSearcher, Drop drop) { - super(new CloseOnDrop(drop)); + private static final Drop DROP = new Drop<>() { + @Override + public void drop(UnshardedIndexSearchers obj) { + try { + if (obj.indexSearcher != null) { + obj.indexSearcher.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close indexSearcher", ex); + } + try { + if (obj.onClose != null) { + obj.onClose.run(); + } + } catch (Throwable ex) { + logger.error("Failed to close onClose", ex); + } + } + + @Override + public Drop fork() { + return this; + } + + @Override + public void attach(UnshardedIndexSearchers obj) { + + } + }; + + private LLIndexSearcher indexSearcher; + private Runnable onClose; + + public UnshardedIndexSearchers(Send indexSearcher, Runnable onClose) { + super(DROP); this.indexSearcher = indexSearcher.receive(); + this.onClose = onClose; } @Override @@ -82,41 +119,61 @@ public interface LLIndexSearchers extends Resource { @Override protected Owned prepareSend() { Send indexSearcher = this.indexSearcher.send(); - return drop -> new UnshardedIndexSearchers(indexSearcher, drop); + var onClose = this.onClose; + return drop -> { + var instance = new UnshardedIndexSearchers(indexSearcher, onClose); + drop.attach(instance); + return instance; + }; } protected void makeInaccessible() { this.indexSearcher = null; - } - - private static class CloseOnDrop implements Drop { - - private final Drop delegate; - - public CloseOnDrop(Drop drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(UnshardedIndexSearchers obj) { - obj.indexSearcher.close(); - delegate.drop(obj); - } + this.onClose = null; } } class ShardedIndexSearchers extends LiveResourceSupport implements LLIndexSearchers { + private static final Logger logger = LoggerFactory.getLogger(ShardedIndexSearchers.class); + + private static final Drop DROP = new Drop<>() { + @Override + public void drop(ShardedIndexSearchers obj) { + try { + for (LLIndexSearcher indexSearcher : obj.indexSearchers) { + indexSearcher.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close indexSearcher", ex); + } + try { + if (obj.onClose != null) { + obj.onClose.run(); + } + } catch (Throwable ex) { + logger.error("Failed to close onClose", ex); + } + } + + @Override + public Drop fork() { + return this; + } + + @Override + public void attach(ShardedIndexSearchers obj) { + + } + }; + private List indexSearchers; private List indexSearchersVals; + private Runnable onClose; - public ShardedIndexSearchers(List> indexSearchers, Drop drop) { - super(new CloseOnDrop(drop)); + public ShardedIndexSearchers(List> indexSearchers, Runnable onClose) { + super(DROP); this.indexSearchers = new ArrayList<>(indexSearchers.size()); this.indexSearchersVals = new ArrayList<>(indexSearchers.size()); for (Send llIndexSearcher : indexSearchers) { @@ -124,6 +181,7 @@ public interface LLIndexSearchers extends Resource { this.indexSearchers.add(indexSearcher); this.indexSearchersVals.add(indexSearcher.getIndexSearcher()); } + this.onClose = onClose; } @Override @@ -177,33 +235,18 @@ public interface LLIndexSearchers extends Resource { for (LLIndexSearcher indexSearcher : this.indexSearchers) { indexSearchers.add(indexSearcher.send()); } - return drop -> new ShardedIndexSearchers(indexSearchers, drop); + var onClose = this.onClose; + return drop -> { + var instance = new ShardedIndexSearchers(indexSearchers, onClose); + drop.attach(instance); + return instance; + }; } protected void makeInaccessible() { this.indexSearchers = null; this.indexSearchersVals = null; - } - - private static class CloseOnDrop implements Drop { - - private final Drop delegate; - - public CloseOnDrop(Drop drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(ShardedIndexSearchers obj) { - for (LLIndexSearcher indexSearcher : obj.indexSearchers) { - indexSearcher.close(); - } - delegate.drop(obj); - } + this.onClose = null; } } } 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 df12df5..75c1ee3 100644 --- a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalLuceneIndex.java +++ b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalLuceneIndex.java @@ -332,7 +332,7 @@ public class LLLocalLuceneIndex implements LLLuceneIndex { return localSearcher.collect(searcher, localQueryParams, keyFieldName, transformer).map(resultToReceive -> { var result = resultToReceive.receive(); - return new LLSearchResultShard(result.results(), result.totalHitsCount(), d -> result.close()).send(); + return new LLSearchResultShard(result.results(), result.totalHitsCount(), result::close).send(); }).doOnDiscard(Send.class, Send::close); } @@ -344,7 +344,7 @@ public class LLLocalLuceneIndex implements LLLuceneIndex { return localSearcher.collect(searcher, localQueryParams, keyFieldName, NO_TRANSFORMATION).map(resultToReceive -> { var result = resultToReceive.receive(); - return new LLSearchResultShard(result.results(), result.totalHitsCount(), d -> result.close()).send(); + return new LLSearchResultShard(result.results(), result.totalHitsCount(), result::close).send(); }).doOnDiscard(Send.class, Send::close); } diff --git a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalMultiLuceneIndex.java b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalMultiLuceneIndex.java index a38b5d3..319551d 100644 --- a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalMultiLuceneIndex.java +++ b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalMultiLuceneIndex.java @@ -206,8 +206,7 @@ public class LLLocalMultiLuceneIndex implements LLLuceneIndex { // Transform the result type .map(resultToReceive -> { var result = resultToReceive.receive(); - return new LLSearchResultShard(result.results(), result.totalHitsCount(), - d -> result.close()).send(); + return new LLSearchResultShard(result.results(), result.totalHitsCount(), result::close).send(); }) .doOnDiscard(Send.class, Send::close); } @@ -225,8 +224,7 @@ public class LLLocalMultiLuceneIndex implements LLLuceneIndex { // Transform the result type .map(resultToReceive -> { var result = resultToReceive.receive(); - return new LLSearchResultShard(result.results(), result.totalHitsCount(), - d -> result.close()).send(); + return new LLSearchResultShard(result.results(), result.totalHitsCount(), result::close).send(); }) .doOnDiscard(Send.class, Send::close); } diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/CountLuceneLocalSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/CountLuceneLocalSearcher.java index 1e38688..845500b 100644 --- a/src/main/java/it/cavallium/dbengine/lucene/searcher/CountLuceneLocalSearcher.java +++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/CountLuceneLocalSearcher.java @@ -26,7 +26,7 @@ public class CountLuceneLocalSearcher implements LuceneLocalSearcher { }).subscribeOn(Schedulers.boundedElastic()), is -> Mono.empty() ) - .map(count -> new LuceneSearchResult(TotalHitsCount.of(count, true), Flux.empty(), drop -> {}).send()) + .map(count -> new LuceneSearchResult(TotalHitsCount.of(count, true), Flux.empty(), null).send()) .doOnDiscard(Send.class, Send::close); } } diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/LuceneSearchResult.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/LuceneSearchResult.java index f92d27f..73f5c7d 100644 --- a/src/main/java/it/cavallium/dbengine/lucene/searcher/LuceneSearchResult.java +++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/LuceneSearchResult.java @@ -19,13 +19,38 @@ public final class LuceneSearchResult extends LiveResourceSupport DROP = new Drop<>() { + @Override + public void drop(LuceneSearchResult obj) { + try { + if (obj.onClose != null) { + obj.onClose.run(); + } + } catch (Throwable ex) { + logger.error("Failed to close onClose", ex); + } + } + + @Override + public Drop fork() { + return this; + } + + @Override + public void attach(LuceneSearchResult obj) { + + } + }; + private TotalHitsCount totalHitsCount; private Flux results; + private Runnable onClose; - public LuceneSearchResult(TotalHitsCount totalHitsCount, Flux results, Drop drop) { - super(drop); + public LuceneSearchResult(TotalHitsCount totalHitsCount, Flux results, Runnable onClose) { + super(DROP); this.totalHitsCount = totalHitsCount; this.results = results; + this.onClose = onClose; } public TotalHitsCount totalHitsCount() { @@ -71,12 +96,18 @@ public final class LuceneSearchResult extends LiveResourceSupport prepareSend() { var totalHitsCount = this.totalHitsCount; var results = this.results; - return drop -> new LuceneSearchResult(totalHitsCount, results, drop); + var onClose = this.onClose; + return drop -> { + var instance = new LuceneSearchResult(totalHitsCount, results, onClose); + drop.attach(instance); + return instance; + }; } protected void makeInaccessible() { this.totalHitsCount = null; this.results = null; + this.onClose = null; } } diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneShardSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneShardSearcher.java index 46b9376..4720050 100644 --- a/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneShardSearcher.java +++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneShardSearcher.java @@ -111,7 +111,7 @@ public class ScoredSimpleLuceneShardSearcher implements LuceneMultiSearcher { List indexSearchers, LocalQueryParams queryParams, String keyFieldName, - Runnable drop) { + Runnable onClose) { var totalHitsCount = firstResult.totalHitsCount(); var firstPageHitsFlux = firstResult.firstPageHitsFlux(); var secondPageInfo = firstResult.nextPageInfo(); @@ -119,7 +119,7 @@ public class ScoredSimpleLuceneShardSearcher implements LuceneMultiSearcher { Flux nextHitsFlux = searchOtherPages(indexSearchers, queryParams, keyFieldName, secondPageInfo); Flux combinedFlux = firstPageHitsFlux.concatWith(nextHitsFlux); - return new LuceneSearchResult(totalHitsCount, combinedFlux, d -> drop.run()).send(); + return new LuceneSearchResult(totalHitsCount, combinedFlux, onClose).send(); } /** diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleLuceneLocalSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleLuceneLocalSearcher.java index b5cb26a..e69b01b 100644 --- a/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleLuceneLocalSearcher.java +++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleLuceneLocalSearcher.java @@ -102,7 +102,7 @@ public class SimpleLuceneLocalSearcher implements LuceneLocalSearcher { List indexSearchers, LocalQueryParams queryParams, String keyFieldName, - Runnable drop) { + Runnable onClose) { return firstResultMono.map(firstResult -> { var totalHitsCount = firstResult.totalHitsCount(); var firstPageHitsFlux = firstResult.firstPageHitsFlux(); @@ -111,7 +111,7 @@ public class SimpleLuceneLocalSearcher implements LuceneLocalSearcher { Flux nextHitsFlux = searchOtherPages(indexSearchers, queryParams, keyFieldName, secondPageInfo); Flux combinedFlux = firstPageHitsFlux.concatWith(nextHitsFlux); - return new LuceneSearchResult(totalHitsCount, combinedFlux, d -> drop.run()).send(); + return new LuceneSearchResult(totalHitsCount, combinedFlux, onClose).send(); }); } diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleUnsortedUnscoredLuceneMultiSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleUnsortedUnscoredLuceneMultiSearcher.java index e5bfed7..fcc8ade 100644 --- a/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleUnsortedUnscoredLuceneMultiSearcher.java +++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleUnsortedUnscoredLuceneMultiSearcher.java @@ -44,7 +44,7 @@ public class SimpleUnsortedUnscoredLuceneMultiSearcher implements LuceneMultiSea indexSearchers -> Flux .fromIterable(indexSearchers.shards()) .flatMap(searcher -> { - var llSearcher = Mono.fromCallable(() -> new LLIndexSearcher(searcher, false, d -> {}).send()); + var llSearcher = Mono.fromCallable(() -> new LLIndexSearcher(searcher, false, null).send()); return localSearcher.collect(llSearcher, localQueryParams, keyFieldName, transformer); }) .collectList() @@ -67,7 +67,7 @@ public class SimpleUnsortedUnscoredLuceneMultiSearcher implements LuceneMultiSea .skip(queryParams.offset()) .take(queryParams.limit(), true); - return new LuceneSearchResult(totalHitsCount, mergedFluxes, d -> { + return new LuceneSearchResult(totalHitsCount, mergedFluxes, () -> { for (LuceneSearchResult luceneSearchResult : resultsToDrop) { luceneSearchResult.close(); } diff --git a/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java b/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java index fe996df..f1f3eb7 100644 --- a/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java +++ b/src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java @@ -92,9 +92,7 @@ public class UnsortedUnscoredContinuousLuceneMultiSearcher implements LuceneMult .skip(queryParams.offset()) .take(queryParams.limit(), true); - return new LuceneSearchResult(totalHitsCount, mergedFluxes, d -> { - indexSearchers.close(); - }).send(); + return new LuceneSearchResult(totalHitsCount, mergedFluxes, indexSearchers::close).send(); }), false); } diff --git a/src/main/java/it/cavallium/dbengine/netty/JMXNettyMonitoringManager.java b/src/main/java/it/cavallium/dbengine/netty/JMXNettyMonitoringManager.java index c6dc415..72f5067 100644 --- a/src/main/java/it/cavallium/dbengine/netty/JMXNettyMonitoringManager.java +++ b/src/main/java/it/cavallium/dbengine/netty/JMXNettyMonitoringManager.java @@ -31,7 +31,7 @@ public class JMXNettyMonitoringManager { public static void initialize() { var instance = getInstance(); - instance.register("global", DefaultGlobalBufferAllocator.DEFAUL_GLOBAL_BUFFER_ALLOCATOR); + instance.register("global", DefaultGlobalBufferAllocator.DEFAULT_GLOBAL_BUFFER_ALLOCATOR); } public synchronized static JMXNettyMonitoringManager getInstance() { diff --git a/src/main/java/it/cavallium/dbengine/netty/NullableBuffer.java b/src/main/java/it/cavallium/dbengine/netty/NullableBuffer.java index 6b90063..b338533 100644 --- a/src/main/java/it/cavallium/dbengine/netty/NullableBuffer.java +++ b/src/main/java/it/cavallium/dbengine/netty/NullableBuffer.java @@ -6,22 +6,50 @@ import io.net5.buffer.api.Owned; import io.net5.buffer.api.Send; import io.net5.buffer.api.internal.ResourceSupport; import it.cavallium.dbengine.client.SearchResult; +import it.cavallium.dbengine.database.LLSearchResultShard; import it.cavallium.dbengine.database.LiveResourceSupport; import it.cavallium.dbengine.database.collections.DatabaseSingle; import org.jetbrains.annotations.Nullable; +import org.warp.commonutils.log.Logger; +import org.warp.commonutils.log.LoggerFactory; public class NullableBuffer extends LiveResourceSupport { + private static final Logger logger = LoggerFactory.getLogger(NullableBuffer.class); + + private static final Drop DROP = new Drop<>() { + @Override + public void drop(NullableBuffer obj) { + try { + if (obj.buffer != null) { + obj.buffer.close(); + } + } catch (Throwable ex) { + logger.error("Failed to close buffer", ex); + } + } + + @Override + public Drop fork() { + return this; + } + + @Override + public void attach(NullableBuffer obj) { + + } + }; + @Nullable private Buffer buffer; - public NullableBuffer(@Nullable Buffer buffer, Drop drop) { - super(new CloseOnDrop(drop)); + public NullableBuffer(@Nullable Buffer buffer) { + super(DROP); this.buffer = buffer == null ? null : buffer.send().receive(); } - public NullableBuffer(@Nullable Send buffer, Drop drop) { - super(new CloseOnDrop(drop)); + public NullableBuffer(@Nullable Send buffer) { + super(DROP); this.buffer = buffer == null ? null : buffer.receive(); } @@ -43,29 +71,10 @@ public class NullableBuffer extends LiveResourceSupport prepareSend() { var buffer = this.buffer == null ? null : this.buffer.send(); - return drop -> new NullableBuffer(buffer, drop); + return drop -> new NullableBuffer(buffer); } protected void makeInaccessible() { this.buffer = null; } - - private static class CloseOnDrop implements Drop { - - private final Drop delegate; - - public CloseOnDrop(Drop drop) { - if (drop instanceof CloseOnDrop closeOnDrop) { - this.delegate = closeOnDrop.delegate; - } else { - this.delegate = drop; - } - } - - @Override - public void drop(NullableBuffer obj) { - if (obj.buffer != null) obj.buffer.close(); - delegate.drop(obj); - } - } } diff --git a/src/test/java/it/cavallium/dbengine/DbTestUtils.java b/src/test/java/it/cavallium/dbengine/DbTestUtils.java index 3a06cfb..1af7174 100644 --- a/src/test/java/it/cavallium/dbengine/DbTestUtils.java +++ b/src/test/java/it/cavallium/dbengine/DbTestUtils.java @@ -180,7 +180,7 @@ public class DbTestUtils { return DatabaseMapDictionary.simple(dictionary, SerializerFixedBinaryLength.utf8(dictionary.getAllocator(), keyBytes), Serializer.utf8(dictionary.getAllocator()), - d -> {} + null ); } else { return DatabaseMapDictionaryHashed.simple(dictionary, @@ -211,7 +211,7 @@ public class DbTestUtils { } } }, - d -> {} + null ); } } @@ -227,7 +227,7 @@ public class DbTestUtils { new SubStageGetterMap<>(SerializerFixedBinaryLength.utf8(dictionary.getAllocator(), key2Bytes), Serializer.utf8(dictionary.getAllocator()) ), - d -> {} + null ); } @@ -243,7 +243,7 @@ public class DbTestUtils { String::hashCode, SerializerFixedBinaryLength.intSerializer(dictionary.getAllocator()) ), - d -> {} + null ); } @@ -254,7 +254,7 @@ public class DbTestUtils { Serializer.utf8(dictionary.getAllocator()), String::hashCode, SerializerFixedBinaryLength.intSerializer(dictionary.getAllocator()), - d -> {} + null ); } }