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

Update netty

Browse Source
This commit is contained in:
Andrea Cavalli 2021-10-01 19:17:33 +02:00
parent 9646ddcb28
commit 818bd20ed9
37 changed files with 756 additions and 401 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
pom.xml
View File

@ -99,10 +99,6 @@
<groupId>io.net5</groupId>
<artifactId>netty-buffer</artifactId>
</dependency>
<dependency>
<groupId>io.net5.incubator</groupId>
<artifactId>netty-incubator-buffer-memseg</artifactId>
</dependency>
<dependency>
<groupId>javax.xml.bind</groupId>
<artifactId>jaxb-api</artifactId>
@ -275,11 +271,6 @@
<artifactId>netty-buffer</artifactId>
<version>5.0.0.Final-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>io.net5.incubator</groupId>
<artifactId>netty-incubator-buffer-memseg</artifactId>
<version>0.0.1.Final-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>javax.xml.bind</groupId>
<artifactId>jaxb-api</artifactId>

9
src/main/java/it/cavallium/dbengine/client/LuceneIndexImpl.java
View File

@ -94,7 +94,7 @@ public class LuceneIndexImpl<T, U> implements LuceneIndex<T, U> {
.fromCallable(signal::key)
.map(indicizer::getKey), signal.score())),
llSearchResult.totalHitsCount(),
d -> llSearchResult.close()
llSearchResult::close
).send();
});
}
@ -107,7 +107,7 @@ public class LuceneIndexImpl<T, U> implements LuceneIndex<T, U> {
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<T, U> implements LuceneIndex<T, U> {
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();
});
}

37
src/main/java/it/cavallium/dbengine/client/SearchResult.java
View File

@ -14,18 +14,39 @@ import reactor.core.publisher.Mono;
public final class SearchResult<T, U> extends LiveResourceSupport<SearchResult<T, U>, SearchResult<T, U>> {
private static final Drop<SearchResult<?, ?>> DROP = new Drop<>() {
@Override
public void drop(SearchResult<?, ?> obj) {
if (obj.onClose != null) {
obj.onClose.run();
}
}
@Override
public Drop<SearchResult<?, ?>> fork() {
return this;
}
@Override
public void attach(SearchResult<?, ?> obj) {
}
};
private Flux<SearchResultItem<T, U>> results;
private TotalHitsCount totalHitsCount;
private Runnable onClose;
public SearchResult(Flux<SearchResultItem<T, U>> results, TotalHitsCount totalHitsCount,
Drop<SearchResult<T, U>> drop) {
super(drop);
@SuppressWarnings({"unchecked", "rawtypes"})
public SearchResult(Flux<SearchResultItem<T, U>> results, TotalHitsCount totalHitsCount, Runnable onClose) {
super((Drop<SearchResult<T,U>>) (Drop) DROP);
this.results = results;
this.totalHitsCount = totalHitsCount;
this.onClose = onClose;
}
public static <T, U> SearchResult<T, U> empty() {
return new SearchResult<T, U>(Flux.empty(), TotalHitsCount.of(0, true), d -> {});
return new SearchResult<T, U>(Flux.empty(), TotalHitsCount.of(0, true), null);
}
public Flux<SearchResultItem<T, U>> results() {
@ -50,11 +71,17 @@ public final class SearchResult<T, U> extends LiveResourceSupport<SearchResult<T
protected Owned<SearchResult<T, U>> 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;
}
}

44
src/main/java/it/cavallium/dbengine/client/SearchResultKeys.java
View File

@ -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<T> extends LiveResourceSupport<SearchResultKeys<T>, SearchResultKeys<T>> {
private static final Logger logger = LoggerFactory.getLogger(SearchResultKeys.class);
private static final Drop<SearchResultKeys<?>> DROP = new Drop<>() {
@Override
public void drop(SearchResultKeys<?> obj) {
if (obj.onClose != null) {
obj.onClose.run();
}
}
@Override
public Drop<SearchResultKeys<?>> fork() {
return this;
}
@Override
public void attach(SearchResultKeys<?> obj) {
}
};
private Flux<SearchResultKey<T>> results;
private TotalHitsCount totalHitsCount;
private Runnable onClose;
public SearchResultKeys(Flux<SearchResultKey<T>> results, TotalHitsCount totalHitsCount,
Drop<SearchResultKeys<T>> drop) {
super(drop);
@SuppressWarnings({"unchecked", "rawtypes"})
public SearchResultKeys(Flux<SearchResultKey<T>> results, TotalHitsCount totalHitsCount, Runnable onClose) {
super((Drop<SearchResultKeys<T>>) (Drop) DROP);
this.results = results;
this.totalHitsCount = totalHitsCount;
this.onClose = onClose;
}
public static <T> SearchResultKeys<T> empty() {
return new SearchResultKeys<T>(Flux.empty(), TotalHitsCount.of(0, true), d -> {});
return new SearchResultKeys<>(Flux.empty(), TotalHitsCount.of(0, true), null);
}
public <U> SearchResult<T, U> withValues(ValueGetter<T, U> 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<SearchResultKey<T>> results() {
@ -62,13 +78,19 @@ public final class SearchResultKeys<T> extends LiveResourceSupport<SearchResultK
protected Owned<SearchResultKeys<T>> 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;
}
}

100
src/main/java/it/cavallium/dbengine/database/LLDelta.java
View File

@ -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<LLDelta, LLDelta> {
@Nullable
private final Buffer previous;
@Nullable
private final Buffer current;
private LLDelta(@Nullable Send<Buffer> previous, @Nullable Send<Buffer> current, Drop<LLDelta> drop) {
super(new LLDelta.CloseOnDrop(drop));
private static final Logger logger = LoggerFactory.getLogger(LLDelta.class);
private static final Drop<LLDelta> 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<LLDelta> 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<Buffer> previous, @Nullable Send<Buffer> 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<LLDelta, LLDelta> {
public static LLDelta of(Send<Buffer> min, Send<Buffer> max) {
assert (min == null && max == null) || (min != max);
return new LLDelta(min, max, d -> {});
return new LLDelta(min, max, null);
}
public Send<Buffer> previous() {
@ -86,42 +132,28 @@ public class LLDelta extends LiveResourceSupport<LLDelta, LLDelta> {
.toString();
}
public LLDelta copy() {
ensureOwned();
var prevCopy = previous != null ? previous.copy().send() : null;
Send<Buffer> 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<LLDelta> prepareSend() {
Send<Buffer> minSend = this.previous != null ? this.previous.send() : null;
Send<Buffer> 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<LLDelta> {
private final Drop<LLDelta> delegate;
public CloseOnDrop(Drop<LLDelta> 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);
}
}
}

99
src/main/java/it/cavallium/dbengine/database/LLEntry.java
View File

@ -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<LLEntry, LLEntry> {
@NotNull
private final Buffer key;
@NotNull
private final Buffer value;
private LLEntry(@NotNull Send<Buffer> key, @NotNull Send<Buffer> value, Drop<LLEntry> drop) {
super(new LLEntry.CloseOnDrop(drop));
private static final Logger logger = LoggerFactory.getLogger(LLEntry.class);
private static final Drop<LLEntry> 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<LLEntry> fork() {
return this;
}
@Override
public void attach(LLEntry obj) {
}
};
@Nullable
private Buffer key;
@Nullable
private Buffer value;
private LLEntry(@NotNull Send<Buffer> key, @NotNull Send<Buffer> 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<Buffer> key, @NotNull Send<Buffer> value) {
return new LLEntry(key, value, d -> {});
return new LLEntry(key, value);
}
public Send<Buffer> getKey() {
ensureOwned();
return key.copy().send();
return Objects.requireNonNull(key).copy().send();
}
public Buffer getKeyUnsafe() {
@ -45,7 +81,7 @@ public class LLEntry extends LiveResourceSupport<LLEntry, LLEntry> {
public Send<Buffer> getValue() {
ensureOwned();
return value.copy().send();
return Objects.requireNonNull(value).copy().send();
}
@ -64,6 +100,12 @@ public class LLEntry extends LiveResourceSupport<LLEntry, LLEntry> {
}
}
@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<LLEntry, LLEntry> {
.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<LLEntry, LLEntry> {
protected Owned<LLEntry> prepareSend() {
Send<Buffer> keySend;
Send<Buffer> valueSend;
keySend = this.key.send();
valueSend = this.value.send();
return drop -> new LLEntry(keySend, valueSend, drop);
}
private static class CloseOnDrop implements Drop<LLEntry> {
private final Drop<LLEntry> delegate;
public CloseOnDrop(Drop<LLEntry> 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;
};
}
}

85
src/main/java/it/cavallium/dbengine/database/LLRange.java
View File

@ -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<LLRange, LLRange> {
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<LLRange> 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<LLRange> 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<LLRange, LLRange> {
@Nullable
private Buffer single;
private LLRange(Send<Buffer> min, Send<Buffer> max, Send<Buffer> single, Drop<LLRange> drop) {
super(new CloseOnDrop(drop));
private LLRange(Send<Buffer> min, Send<Buffer> max, Send<Buffer> 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<LLRange, LLRange> {
}
public static LLRange from(Send<Buffer> min) {
return new LLRange(min, null, null, d -> {});
return new LLRange(min, null, null);
}
public static LLRange to(Send<Buffer> max) {
return new LLRange(null, max, null, d -> {});
return new LLRange(null, max, null);
}
public static LLRange single(Send<Buffer> single) {
return new LLRange(null, null, single, d -> {});
return new LLRange(null, null, single);
}
public static LLRange of(Send<Buffer> min, Send<Buffer> 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<LLRange, LLRange> {
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<LLRange, LLRange> {
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<LLRange, LLRange> {
this.max = null;
this.single = null;
}
private static class CloseOnDrop implements Drop<LLRange> {
private final Drop<LLRange> delegate;
public CloseOnDrop(Drop<LLRange> 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);
}
}
}

33
src/main/java/it/cavallium/dbengine/database/LLSearchResultShard.java
View File

@ -13,13 +13,38 @@ public final class LLSearchResultShard extends LiveResourceSupport<LLSearchResul
private static final Logger logger = LoggerFactory.getLogger(LLSearchResultShard.class);
private static final Drop<LLSearchResultShard> 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<LLSearchResultShard> fork() {
return this;
}
@Override
public void attach(LLSearchResultShard obj) {
}
};
private Flux<LLKeyScore> results;
private TotalHitsCount totalHitsCount;
private Runnable onClose;
public LLSearchResultShard(Flux<LLKeyScore> results, TotalHitsCount totalHitsCount, Drop<LLSearchResultShard> drop) {
super(drop);
public LLSearchResultShard(Flux<LLKeyScore> results, TotalHitsCount totalHitsCount, Runnable onClose) {
super(DROP);
this.results = results;
this.totalHitsCount = totalHitsCount;
this.onClose = onClose;
}
public Flux<LLKeyScore> results() {
@ -65,11 +90,13 @@ public final class LLSearchResultShard extends LiveResourceSupport<LLSearchResul
protected Owned<LLSearchResultShard> 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;
}
}

6
src/main/java/it/cavallium/dbengine/database/collections/DatabaseEmpty.java
View File

@ -36,10 +36,8 @@ public class DatabaseEmpty {
private DatabaseEmpty() {
}
public static DatabaseStageEntry<Nothing> create(LLDictionary dictionary,
Send<Buffer> key,
Drop<DatabaseSingle<Nothing>> drop) {
return new DatabaseSingle<>(dictionary, key, nothingSerializer(dictionary.getAllocator()), drop);
public static DatabaseStageEntry<Nothing> create(LLDictionary dictionary, Send<Buffer> key, Runnable onClose) {
return new DatabaseSingle<>(dictionary, key, nothingSerializer(dictionary.getAllocator()), onClose);
}
public static final class Nothing {

16
src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionary.java
View File

@ -44,26 +44,26 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
@NotNull Send<Buffer> prefixKey,
SerializerFixedBinaryLength<T> keySuffixSerializer,
Serializer<U> valueSerializer,
Drop<DatabaseMapDictionaryDeep<T, U, DatabaseStageEntry<U>>> 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 <T, U> DatabaseMapDictionary<T, U> simple(LLDictionary dictionary,
SerializerFixedBinaryLength<T> keySerializer,
Serializer<U> valueSerializer,
Drop<DatabaseMapDictionaryDeep<T, U, DatabaseStageEntry<U>>> drop) {
Runnable onClose) {
return new DatabaseMapDictionary<>(dictionary, LLUtils.empty(dictionary.getAllocator()), keySerializer,
valueSerializer, drop);
valueSerializer, onClose);
}
public static <T, U> DatabaseMapDictionary<T, U> tail(LLDictionary dictionary,
Send<Buffer> prefixKey,
SerializerFixedBinaryLength<T> keySuffixSerializer,
Serializer<U> valueSerializer,
Drop<DatabaseMapDictionaryDeep<T, U, DatabaseStageEntry<U>>> drop) {
return new DatabaseMapDictionary<>(dictionary, prefixKey, keySuffixSerializer, valueSerializer, drop);
Runnable onClose) {
return new DatabaseMapDictionary<>(dictionary, prefixKey, keySuffixSerializer, valueSerializer, onClose);
}
private Send<Buffer> toKey(Send<Buffer> suffixKeyToSend) {
@ -152,7 +152,7 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
@Override
public Mono<DatabaseStageEntry<U>> 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<T, U> extends DatabaseMapDictionaryDeep<T, U,
removePrefix(keyBuf);
suffixKeyConsistency(keyBuf.readableBytes());
sink.next(Map.entry(deserializeSuffix(keyBuf.copy().send()),
new DatabaseSingle<>(dictionary, toKey(keyBuf.send()), valueSerializer, d -> {})
new DatabaseSingle<>(dictionary, toKey(keyBuf.send()), valueSerializer, null)
));
} catch (SerializationException ex) {
sink.error(ex);

107
src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryDeep.java
View File

@ -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<T, U, US extends DatabaseStage<U>> extend
LiveResourceSupport<DatabaseStage<Map<T, U>>, DatabaseMapDictionaryDeep<T, U, US>>
implements DatabaseStageMap<T, U, US> {
private static final Logger logger = LoggerFactory.getLogger(DatabaseMapDictionaryDeep.class);
private static final Drop<DatabaseMapDictionaryDeep<?, ?, ?>> 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<DatabaseMapDictionaryDeep<?, ?, ?>> fork() {
return this;
}
@Override
public void attach(DatabaseMapDictionaryDeep<?, ?, ?> obj) {
}
};
protected final LLDictionary dictionary;
private final BufferAllocator alloc;
protected final SubStageGetter<U, US> subStageGetter;
@ -44,6 +85,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> 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<T, U, US extends DatabaseStage<U>> extend
@Deprecated
public static <T, U> DatabaseMapDictionaryDeep<T, U, DatabaseStageEntry<U>> simple(LLDictionary dictionary,
SerializerFixedBinaryLength<T> keySerializer, SubStageGetterSingle<U> subStageGetter,
Drop<DatabaseMapDictionaryDeep<T, U, DatabaseStageEntry<U>>> drop) {
Runnable onClose) {
return new DatabaseMapDictionaryDeep<>(dictionary, LLUtils.empty(dictionary.getAllocator()), keySerializer,
subStageGetter, 0, drop);
subStageGetter, 0, onClose);
}
public static <T, U, US extends DatabaseStage<U>> DatabaseMapDictionaryDeep<T, U, US> deepTail(
LLDictionary dictionary, SerializerFixedBinaryLength<T> keySerializer, int keyExtLength,
SubStageGetter<U, US> subStageGetter, Drop<DatabaseMapDictionaryDeep<T, U, US>> drop) {
SubStageGetter<U, US> subStageGetter, Runnable onClose) {
return new DatabaseMapDictionaryDeep<>(dictionary, LLUtils.empty(dictionary.getAllocator()), keySerializer,
subStageGetter, keyExtLength, drop);
subStageGetter, keyExtLength, onClose);
}
public static <T, U, US extends DatabaseStage<U>> DatabaseMapDictionaryDeep<T, U, US> deepIntermediate(
LLDictionary dictionary, Send<Buffer> prefixKey, SerializerFixedBinaryLength<T> keySuffixSerializer,
SubStageGetter<U, US> subStageGetter, int keyExtLength, Drop<DatabaseMapDictionaryDeep<T, U, US>> drop) {
SubStageGetter<U, US> 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<Buffer> prefixKeyToReceive,
SerializerFixedBinaryLength<T> keySuffixSerializer, SubStageGetter<U, US> subStageGetter, int keyExtLength,
Drop<DatabaseMapDictionaryDeep<T, U, US>> drop) {
super(new CloseOnDrop<>(drop));
Runnable onClose) {
super((Drop<DatabaseMapDictionaryDeep<T, U, US>>) (Drop) DROP);
try (var prefixKey = prefixKeyToReceive.receive()) {
this.dictionary = dictionary;
this.alloc = dictionary.getAllocator();
@ -198,9 +241,11 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
}
this.keyPrefix = prefixKey.send().receive();
this.onClose = onClose;
}
}
@SuppressWarnings({"unchecked", "rawtypes"})
private DatabaseMapDictionaryDeep(LLDictionary dictionary,
BufferAllocator alloc,
SubStageGetter<U, US> subStageGetter,
@ -211,8 +256,8 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
Mono<Send<LLRange>> rangeMono,
Send<LLRange> range,
Send<Buffer> keyPrefix,
Drop<DatabaseMapDictionaryDeep<T, U, US>> drop) {
super(new CloseOnDrop<>(drop));
Runnable onClose) {
super((Drop<DatabaseMapDictionaryDeep<T,U,US>>) (Drop) DROP);
this.dictionary = dictionary;
this.alloc = alloc;
this.subStageGetter = subStageGetter;
@ -224,6 +269,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
this.range = range.receive();
this.keyPrefix = keyPrefix.receive();
this.onClose = onClose;
}
@SuppressWarnings("unused")
@ -418,34 +464,29 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
protected Owned<DatabaseMapDictionaryDeep<T, U, US>> 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<T, U, US extends DatabaseStage<U>>
implements Drop<DatabaseMapDictionaryDeep<T, U, US>> {
private final Drop<DatabaseMapDictionaryDeep<T,U,US>> delegate;
public CloseOnDrop(Drop<DatabaseMapDictionaryDeep<T, U, US>> drop) {
if (drop instanceof CloseOnDrop<T, U, US> closeOnDrop) {
this.delegate = closeOnDrop.delegate;
} else {
this.delegate = drop;
}
}
@Override
public void drop(DatabaseMapDictionaryDeep<T, U, US> obj) {
obj.range.close();
obj.keyPrefix.close();
delegate.drop(obj);
}
this.onClose = null;
}
}

66
src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryHashed.java
View File

@ -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<T, U, TH> extends
LiveResourceSupport<DatabaseStage<Map<T, U>>, DatabaseMapDictionaryHashed<T, U, TH>>
implements DatabaseStageMap<T, U, DatabaseStageEntry<U>> {
private static final Logger logger = LoggerFactory.getLogger(DatabaseMapDictionaryHashed.class);
private static final Drop<DatabaseMapDictionaryHashed<?, ?, ?>> 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<DatabaseMapDictionaryHashed<?, ?, ?>> fork() {
return this;
}
@Override
public void attach(DatabaseMapDictionaryHashed<?, ?, ?> obj) {
}
};
private final BufferAllocator alloc;
private final Function<T, TH> keySuffixHashFunction;
private DatabaseMapDictionary<TH, ObjectArraySet<Entry<T, U>>> subDictionary;
@SuppressWarnings({"unchecked", "rawtypes"})
protected DatabaseMapDictionaryHashed(LLDictionary dictionary,
@NotNull Send<Buffer> prefixKey,
Serializer<T> keySuffixSerializer,
Serializer<U> valueSerializer,
Function<T, TH> keySuffixHashFunction,
SerializerFixedBinaryLength<TH> keySuffixHashSerializer,
Drop<DatabaseMapDictionaryHashed<T, U, TH>> drop) {
super(new DatabaseMapDictionaryHashed.CloseOnDrop<>(drop));
Runnable onClose) {
super((Drop<DatabaseMapDictionaryHashed<T, U, TH>>) (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<T, U, TH> extends
ValuesSetSerializer<Entry<T, U>> 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<T, TH> keySuffixHashFunction,
Send<DatabaseStage<Map<TH, ObjectArraySet<Entry<T, U>>>>> subDictionary,
Drop<DatabaseMapDictionaryHashed<T, U, TH>> drop) {
super(new CloseOnDrop<>(drop));
super((Drop<DatabaseMapDictionaryHashed<T, U, TH>>) (Drop) DROP);
this.alloc = alloc;
this.keySuffixHashFunction = keySuffixHashFunction;
@ -75,7 +104,7 @@ public class DatabaseMapDictionaryHashed<T, U, TH> extends
Serializer<U> valueSerializer,
Function<T, UH> keyHashFunction,
SerializerFixedBinaryLength<UH> keyHashSerializer,
Drop<DatabaseMapDictionaryHashed<T, U, UH>> drop) {
Runnable onClose) {
return new DatabaseMapDictionaryHashed<>(
dictionary,
LLUtils.empty(dictionary.getAllocator()),
@ -83,7 +112,7 @@ public class DatabaseMapDictionaryHashed<T, U, TH> extends
valueSerializer,
keyHashFunction,
keyHashSerializer,
drop
onClose
);
}
@ -93,14 +122,14 @@ public class DatabaseMapDictionaryHashed<T, U, TH> extends
Serializer<U> valueSerializer,
Function<T, UH> keySuffixHashFunction,
SerializerFixedBinaryLength<UH> keySuffixHashSerializer,
Drop<DatabaseMapDictionaryHashed<T, U, UH>> drop) {
Runnable onClose) {
return new DatabaseMapDictionaryHashed<>(dictionary,
prefixKey,
keySuffixSerializer,
valueSerializer,
keySuffixHashFunction,
keySuffixHashSerializer,
drop
onClose
);
}
@ -173,7 +202,7 @@ public class DatabaseMapDictionaryHashed<T, U, TH> extends
private Mono<DatabaseSingleBucket<T, U, TH>> atPrivate(@Nullable CompositeSnapshot snapshot, T key, TH hash) {
return subDictionary
.at(snapshot, hash)
.map(entry -> new DatabaseSingleBucket<T, U, TH>(entry, key, d -> {}))
.map(entry -> new DatabaseSingleBucket<T, U, TH>(entry, key, null))
.doOnDiscard(Resource.class, Resource::close);
}
@ -325,23 +354,4 @@ public class DatabaseMapDictionaryHashed<T, U, TH> extends
protected void makeInaccessible() {
this.subDictionary = null;
}
private static class CloseOnDrop<T, U, TH> implements Drop<DatabaseMapDictionaryHashed<T,U,TH>> {
private final Drop<DatabaseMapDictionaryHashed<T,U,TH>> delegate;
public CloseOnDrop(Drop<DatabaseMapDictionaryHashed<T,U,TH>> drop) {
if (drop instanceof CloseOnDrop<T, U, TH> closeOnDrop) {
this.delegate = closeOnDrop.delegate;
} else {
this.delegate = drop;
}
}
@Override
public void drop(DatabaseMapDictionaryHashed<T, U, TH> obj) {
obj.subDictionary.close();
delegate.drop(obj);
}
}
}

12
src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionary.java
View File

@ -20,21 +20,21 @@ public class DatabaseSetDictionary<T> extends DatabaseMapDictionary<T, Nothing>
protected DatabaseSetDictionary(LLDictionary dictionary,
Send<Buffer> prefixKey,
SerializerFixedBinaryLength<T> keySuffixSerializer,
Drop<DatabaseMapDictionaryDeep<T, Nothing, DatabaseStageEntry<Nothing>>> drop) {
super(dictionary, prefixKey, keySuffixSerializer, DatabaseEmpty.nothingSerializer(dictionary.getAllocator()), drop);
Runnable onClose) {
super(dictionary, prefixKey, keySuffixSerializer, DatabaseEmpty.nothingSerializer(dictionary.getAllocator()), onClose);
}
public static <T> DatabaseSetDictionary<T> simple(LLDictionary dictionary,
SerializerFixedBinaryLength<T> keySerializer,
Drop<DatabaseMapDictionaryDeep<T, Nothing, DatabaseStageEntry<Nothing>>> 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 <T> DatabaseSetDictionary<T> tail(LLDictionary dictionary,
Send<Buffer> prefixKey,
SerializerFixedBinaryLength<T> keySuffixSerializer,
Drop<DatabaseMapDictionaryDeep<T, Nothing, DatabaseStageEntry<Nothing>>> drop) {
return new DatabaseSetDictionary<>(dictionary, prefixKey, keySuffixSerializer, drop);
Runnable onClose) {
return new DatabaseSetDictionary<>(dictionary, prefixKey, keySuffixSerializer, onClose);
}
public Mono<Set<T>> getKeySet(@Nullable CompositeSnapshot snapshot) {

13
src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionaryHashed.java
View File

@ -25,14 +25,14 @@ public class DatabaseSetDictionaryHashed<T, TH> extends DatabaseMapDictionaryHas
Serializer<T> keySuffixSerializer,
Function<T, TH> keySuffixHashFunction,
SerializerFixedBinaryLength<TH> keySuffixHashSerializer,
Drop<DatabaseMapDictionaryHashed<T, Nothing, TH>> drop) {
Runnable onClose) {
super(dictionary,
prefixKey,
keySuffixSerializer,
DatabaseEmpty.nothingSerializer(dictionary.getAllocator()),
keySuffixHashFunction,
keySuffixHashSerializer,
drop
onClose
);
}
@ -40,13 +40,13 @@ public class DatabaseSetDictionaryHashed<T, TH> extends DatabaseMapDictionaryHas
Serializer<T> keySerializer,
Function<T, TH> keyHashFunction,
SerializerFixedBinaryLength<TH> keyHashSerializer,
Drop<DatabaseMapDictionaryHashed<T, Nothing, TH>> drop) {
Runnable onClose) {
return new DatabaseSetDictionaryHashed<>(dictionary,
LLUtils.empty(dictionary.getAllocator()),
keySerializer,
keyHashFunction,
keyHashSerializer,
drop
onClose
);
}
@ -54,13 +54,14 @@ public class DatabaseSetDictionaryHashed<T, TH> extends DatabaseMapDictionaryHas
Send<Buffer> prefixKey,
Serializer<T> keySuffixSerializer,
Function<T, TH> keyHashFunction,
SerializerFixedBinaryLength<TH> keyHashSerializer, Drop<DatabaseMapDictionaryHashed<T, Nothing, TH>> drop) {
SerializerFixedBinaryLength<TH> keyHashSerializer,
Runnable onClose) {
return new DatabaseSetDictionaryHashed<>(dictionary,
prefixKey,
keySuffixSerializer,
keyHashFunction,
keyHashSerializer,
drop
onClose
);
}

63
src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingle.java
View File

@ -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<U> extends ResourceSupport<DatabaseStage<U>, DatabaseSingle<U>> implements
DatabaseStageEntry<U> {
private static final Logger logger = LoggerFactory.getLogger(DatabaseSingle.class);
private static final Drop<DatabaseSingle<?>> 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<DatabaseSingle<?>> fork() {
return this;
}
@Override
public void attach(DatabaseSingle<?> obj) {
}
};
private final LLDictionary dictionary;
private final Mono<Send<Buffer>> keyMono;
private final Serializer<U> serializer;
private Buffer key;
private Runnable onClose;
@SuppressWarnings({"unchecked", "rawtypes"})
public DatabaseSingle(LLDictionary dictionary, Send<Buffer> key, Serializer<U> serializer,
Drop<DatabaseSingle<U>> drop) {
super(new CloseOnDrop<>(drop));
Runnable onClose) {
super((Drop<DatabaseSingle<U>>) (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<U> extends ResourceSupport<DatabaseStage<U>, Databas
@Override
protected Owned<DatabaseSingle<U>> 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<U> implements Drop<DatabaseSingle<U>> {
private final Drop<DatabaseSingle<U>> delegate;
public CloseOnDrop(Drop<DatabaseSingle<U>> drop) {
if (drop instanceof CloseOnDrop<U> closeOnDrop) {
this.delegate = closeOnDrop.delegate;
} else {
this.delegate = drop;
}
}
@Override
public void drop(DatabaseSingle<U> obj) {
obj.key.close();
delegate.drop(obj);
}
this.onClose = null;
}
}

78
src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleBucket.java
View File

@ -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<K, V, TH>
extends LiveResourceSupport<DatabaseStage<V>, DatabaseSingleBucket<K, V, TH>>
implements DatabaseStageEntry<V> {
private static final Logger logger = LoggerFactory.getLogger(DatabaseSingleBucket.class);
private static final Drop<DatabaseSingleBucket<?, ?, ?>> 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<DatabaseSingleBucket<?, ?, ?>> fork() {
return this;
}
@Override
public void attach(DatabaseSingleBucket<?, ?, ?> obj) {
}
};
private final K key;
private DatabaseStageEntry<ObjectArraySet<Entry<K, V>>> bucketStage;
public DatabaseSingleBucket(DatabaseStageEntry<ObjectArraySet<Entry<K, V>>> bucketStage, K key,
Drop<DatabaseSingleBucket<K, V, TH>> drop) {
super(new CloseOnDrop<>(drop));
private Runnable onClose;
@SuppressWarnings({"unchecked", "rawtypes"})
public DatabaseSingleBucket(DatabaseStageEntry<ObjectArraySet<Entry<K, V>>> bucketStage, K key, Runnable onClose) {
super((Drop<DatabaseSingleBucket<K,V,TH>>) (Drop) DROP);
this.key = key;
this.bucketStage = bucketStage;
this.onClose = onClose;
}
private DatabaseSingleBucket(Send<DatabaseStage<ObjectArraySet<Entry<K, V>>>> bucketStage, K key,
Drop<DatabaseSingleBucket<K, V, TH>> drop) {
super(new CloseOnDrop<>(drop));
@SuppressWarnings({"unchecked", "rawtypes"})
private DatabaseSingleBucket(Send<DatabaseStage<ObjectArraySet<Entry<K, V>>>> bucketStage, K key, Runnable onClose) {
super((Drop<DatabaseSingleBucket<K,V,TH>>) (Drop) DROP);
this.key = key;
this.bucketStage = (DatabaseStageEntry<ObjectArraySet<Entry<K, V>>>) bucketStage.receive();
this.onClose = onClose;
}
@Override
@ -210,31 +248,17 @@ public class DatabaseSingleBucket<K, V, TH>
@Override
protected Owned<DatabaseSingleBucket<K, V, TH>> prepareSend() {
var bucketStage = this.bucketStage.send();
return drop -> new DatabaseSingleBucket<>(bucketStage, key, drop);
var onClose = this.onClose;
return drop -> {
var instance = new DatabaseSingleBucket<K, V, TH>(bucketStage, key, onClose);
drop.attach(instance);
return instance;
};
}
@Override
protected void makeInaccessible() {
this.bucketStage = null;
}
private static class CloseOnDrop<K, V, TH> implements
Drop<DatabaseSingleBucket<K, V, TH>> {
private final Drop<DatabaseSingleBucket<K, V, TH>> delegate;
public CloseOnDrop(Drop<DatabaseSingleBucket<K, V, TH>> drop) {
if (drop instanceof CloseOnDrop<K, V, TH> closeOnDrop) {
this.delegate = closeOnDrop.delegate;
} else {
this.delegate = drop;
}
}
@Override
public void drop(DatabaseSingleBucket<K, V, TH> obj) {
obj.bucketStage.close();
delegate.drop(obj);
}
this.onClose = null;
}
}

51
src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleMapped.java
View File

@ -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<A, B> extends ResourceSupport<DatabaseStage<A>, DatabaseSingleMapped<A, B>>
implements DatabaseStageEntry<A> {
private static final Logger logger = LoggerFactory.getLogger(DatabaseSingleMapped.class);
private static final Drop<DatabaseSingleMapped<?, ?>> 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<DatabaseSingleMapped<?, ?>> fork() {
return this;
}
@Override
public void attach(DatabaseSingleMapped<?, ?> obj) {
}
};
private final Mapper<A, B> mapper;
private DatabaseStageEntry<B> serializedSingle;
@SuppressWarnings({"unchecked", "rawtypes"})
public DatabaseSingleMapped(DatabaseStageEntry<B> serializedSingle, Mapper<A, B> mapper,
Drop<DatabaseSingleMapped<A, B>> drop) {
super(new CloseOnDrop<>(drop));
super((Drop<DatabaseSingleMapped<A,B>>) (Drop) DROP);
this.serializedSingle = serializedSingle;
this.mapper = mapper;
}
@SuppressWarnings({"unchecked", "rawtypes"})
private DatabaseSingleMapped(Send<DatabaseStage<B>> serializedSingle, Mapper<A, B> mapper,
Drop<DatabaseSingleMapped<A, B>> drop) {
super(new CloseOnDrop<>(drop));
super((Drop<DatabaseSingleMapped<A,B>>) (Drop) DROP);
this.mapper = mapper;
this.serializedSingle = (DatabaseStageEntry<B>) serializedSingle.receive();
@ -170,22 +199,4 @@ public class DatabaseSingleMapped<A, B> extends ResourceSupport<DatabaseStage<A>
this.serializedSingle = null;
}
private static class CloseOnDrop<A, B> implements Drop<DatabaseSingleMapped<A, B>> {
private final Drop<DatabaseSingleMapped<A, B>> delegate;
public CloseOnDrop(Drop<DatabaseSingleMapped<A, B>> drop) {
if (drop instanceof CloseOnDrop<A, B> closeOnDrop) {
this.delegate = closeOnDrop.delegate;
} else {
this.delegate = drop;
}
}
@Override
public void drop(DatabaseSingleMapped<A, B> obj) {
obj.serializedSingle.close();
delegate.drop(obj);
}
}
}

2
src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashMap.java
View File

@ -37,7 +37,7 @@ public class SubStageGetterHashMap<T, U, TH> implements
Mono<Send<Buffer>> prefixKeyMono) {
return LLUtils.usingSend(prefixKeyMono, prefixKey -> Mono.just(DatabaseMapDictionaryHashed
.tail(dictionary, prefixKey, keySerializer, valueSerializer, keyHashFunction,
keyHashSerializer, d -> {})), true);
keyHashSerializer, null)), true);
}
public int getKeyHashBinaryLength() {

2
src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashSet.java
View File

@ -35,7 +35,7 @@ public class SubStageGetterHashSet<T, TH> implements
return Mono.usingWhen(prefixKeyMono,
prefixKey -> Mono
.fromSupplier(() -> DatabaseSetDictionaryHashed.tail(dictionary, prefixKey, keySerializer,
keyHashFunction, keyHashSerializer, d -> {})
keyHashFunction, keyHashSerializer, null)
),
prefixKey -> Mono.fromRunnable(prefixKey::close)
);

2
src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMap.java
View File

@ -28,7 +28,7 @@ public class SubStageGetterMap<T, U> implements SubStageGetter<Map<T, U>, Databa
Mono<Send<Buffer>> 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() {

3
src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMapDeep.java
View File

@ -42,8 +42,7 @@ public class SubStageGetterMapDeep<T, U, US extends DatabaseStage<U>> implements
@Nullable CompositeSnapshot snapshot,
Mono<Send<Buffer>> 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() {

2
src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSet.java
View File

@ -24,7 +24,7 @@ public class SubStageGetterSet<T> implements SubStageGetter<Map<T, Nothing>, Dat
Mono<Send<Buffer>> 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)
);
}

2
src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSingle.java
View File

@ -20,7 +20,7 @@ public class SubStageGetterSingle<T> implements SubStageGetter<T, DatabaseStageE
public Mono<DatabaseStageEntry<T>> subStage(LLDictionary dictionary,
@Nullable CompositeSnapshot snapshot,
Mono<Send<Buffer>> keyPrefixMono) {
return keyPrefixMono.map(keyPrefix -> new DatabaseSingle<>(dictionary, keyPrefix, serializer, d -> {}));
return keyPrefixMono.map(keyPrefix -> new DatabaseSingle<>(dictionary, keyPrefix, serializer, null));
}
}

2
src/main/java/it/cavallium/dbengine/database/disk/CachedIndexSearcherManager.java
View File

@ -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();
}

39
src/main/java/it/cavallium/dbengine/database/disk/LLIndexSearcher.java
View File

@ -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<LLIndexSearcher, LLIndexSearcher> {
private static final Logger logger = LoggerFactory.getLogger(LLIndexSearcher.class);
private static final Drop<LLIndexSearcher> 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<LLIndexSearcher> fork() {
return this;
}
@Override
public void attach(LLIndexSearcher obj) {
}
};
private IndexSearcher indexSearcher;
private final boolean decRef;
public LLIndexSearcher(IndexSearcher indexSearcher, boolean decRef, Drop<LLIndexSearcher> 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<LLIndexSearcher, LLInde
@Override
protected Owned<LLIndexSearcher> 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;
}
}

141
src/main/java/it/cavallium/dbengine/database/disk/LLIndexSearchers.java
View File

@ -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<LLIndexSearchers> {
static LLIndexSearchers of(List<Send<LLIndexSearcher>> indexSearchers) {
return new ShardedIndexSearchers(indexSearchers, d -> {});
return new ShardedIndexSearchers(indexSearchers, null);
}
static UnshardedIndexSearchers unsharded(Send<LLIndexSearcher> indexSearcher) {
return new UnshardedIndexSearchers(indexSearcher, d -> {});
return new UnshardedIndexSearchers(indexSearcher, null);
}
List<IndexSearcher> shards();
@ -42,11 +45,45 @@ public interface LLIndexSearchers extends Resource<LLIndexSearchers> {
class UnshardedIndexSearchers extends LiveResourceSupport<LLIndexSearchers, UnshardedIndexSearchers>
implements LLIndexSearchers {
private LLIndexSearcher indexSearcher;
private static final Logger logger = LoggerFactory.getLogger(UnshardedIndexSearchers.class);
public UnshardedIndexSearchers(Send<LLIndexSearcher> indexSearcher, Drop<UnshardedIndexSearchers> drop) {
super(new CloseOnDrop(drop));
private static final Drop<UnshardedIndexSearchers> 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<UnshardedIndexSearchers> fork() {
return this;
}
@Override
public void attach(UnshardedIndexSearchers obj) {
}
};
private LLIndexSearcher indexSearcher;
private Runnable onClose;
public UnshardedIndexSearchers(Send<LLIndexSearcher> indexSearcher, Runnable onClose) {
super(DROP);
this.indexSearcher = indexSearcher.receive();
this.onClose = onClose;
}
@Override
@ -82,41 +119,61 @@ public interface LLIndexSearchers extends Resource<LLIndexSearchers> {
@Override
protected Owned<UnshardedIndexSearchers> prepareSend() {
Send<LLIndexSearcher> 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<UnshardedIndexSearchers> {
private final Drop<UnshardedIndexSearchers> delegate;
public CloseOnDrop(Drop<UnshardedIndexSearchers> 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<LLIndexSearchers, ShardedIndexSearchers>
implements LLIndexSearchers {
private static final Logger logger = LoggerFactory.getLogger(ShardedIndexSearchers.class);
private static final Drop<ShardedIndexSearchers> 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<ShardedIndexSearchers> fork() {
return this;
}
@Override
public void attach(ShardedIndexSearchers obj) {
}
};
private List<LLIndexSearcher> indexSearchers;
private List<IndexSearcher> indexSearchersVals;
private Runnable onClose;
public ShardedIndexSearchers(List<Send<LLIndexSearcher>> indexSearchers, Drop<ShardedIndexSearchers> drop) {
super(new CloseOnDrop(drop));
public ShardedIndexSearchers(List<Send<LLIndexSearcher>> indexSearchers, Runnable onClose) {
super(DROP);
this.indexSearchers = new ArrayList<>(indexSearchers.size());
this.indexSearchersVals = new ArrayList<>(indexSearchers.size());
for (Send<LLIndexSearcher> llIndexSearcher : indexSearchers) {
@ -124,6 +181,7 @@ public interface LLIndexSearchers extends Resource<LLIndexSearchers> {
this.indexSearchers.add(indexSearcher);
this.indexSearchersVals.add(indexSearcher.getIndexSearcher());
}
this.onClose = onClose;
}
@Override
@ -177,33 +235,18 @@ public interface LLIndexSearchers extends Resource<LLIndexSearchers> {
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<ShardedIndexSearchers> {
private final Drop<ShardedIndexSearchers> delegate;
public CloseOnDrop(Drop<ShardedIndexSearchers> 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;
}
}
}

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

@ -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);
}

6
src/main/java/it/cavallium/dbengine/database/disk/LLLocalMultiLuceneIndex.java
View File

@ -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);
}

2
src/main/java/it/cavallium/dbengine/lucene/searcher/CountLuceneLocalSearcher.java
View File

@ -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);
}
}

37
src/main/java/it/cavallium/dbengine/lucene/searcher/LuceneSearchResult.java
View File

@ -19,13 +19,38 @@ public final class LuceneSearchResult extends LiveResourceSupport<LuceneSearchRe
private static final Logger logger = LoggerFactory.getLogger(LuceneSearchResult.class);
private static final Drop<LuceneSearchResult> 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<LuceneSearchResult> fork() {
return this;
}
@Override
public void attach(LuceneSearchResult obj) {
}
};
private TotalHitsCount totalHitsCount;
private Flux<LLKeyScore> results;
private Runnable onClose;
public LuceneSearchResult(TotalHitsCount totalHitsCount, Flux<LLKeyScore> results, Drop<LuceneSearchResult> drop) {
super(drop);
public LuceneSearchResult(TotalHitsCount totalHitsCount, Flux<LLKeyScore> 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<LuceneSearchRe
protected Owned<LuceneSearchResult> 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;
}
}

4
src/main/java/it/cavallium/dbengine/lucene/searcher/ScoredSimpleLuceneShardSearcher.java
View File

@ -111,7 +111,7 @@ public class ScoredSimpleLuceneShardSearcher implements LuceneMultiSearcher {
List<IndexSearcher> 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<LLKeyScore> nextHitsFlux = searchOtherPages(indexSearchers, queryParams, keyFieldName, secondPageInfo);
Flux<LLKeyScore> combinedFlux = firstPageHitsFlux.concatWith(nextHitsFlux);
return new LuceneSearchResult(totalHitsCount, combinedFlux, d -> drop.run()).send();
return new LuceneSearchResult(totalHitsCount, combinedFlux, onClose).send();
}
/**

4
src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleLuceneLocalSearcher.java
View File

@ -102,7 +102,7 @@ public class SimpleLuceneLocalSearcher implements LuceneLocalSearcher {
List<IndexSearcher> 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<LLKeyScore> nextHitsFlux = searchOtherPages(indexSearchers, queryParams, keyFieldName, secondPageInfo);
Flux<LLKeyScore> combinedFlux = firstPageHitsFlux.concatWith(nextHitsFlux);
return new LuceneSearchResult(totalHitsCount, combinedFlux, d -> drop.run()).send();
return new LuceneSearchResult(totalHitsCount, combinedFlux, onClose).send();
});
}

4
src/main/java/it/cavallium/dbengine/lucene/searcher/SimpleUnsortedUnscoredLuceneMultiSearcher.java
View File

@ -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();
}

4
src/main/java/it/cavallium/dbengine/lucene/searcher/UnsortedUnscoredContinuousLuceneMultiSearcher.java
View File

@ -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);
}

2
src/main/java/it/cavallium/dbengine/netty/JMXNettyMonitoringManager.java
View File

@ -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() {

57
src/main/java/it/cavallium/dbengine/netty/NullableBuffer.java
View File

@ -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<NullableBuffer, NullableBuffer> {
private static final Logger logger = LoggerFactory.getLogger(NullableBuffer.class);
private static final Drop<NullableBuffer> 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<NullableBuffer> fork() {
return this;
}
@Override
public void attach(NullableBuffer obj) {
}
};
@Nullable
private Buffer buffer;
public NullableBuffer(@Nullable Buffer buffer, Drop<NullableBuffer> 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> buffer, Drop<NullableBuffer> drop) {
super(new CloseOnDrop(drop));
public NullableBuffer(@Nullable Send<Buffer> buffer) {
super(DROP);
this.buffer = buffer == null ? null : buffer.receive();
}
@ -43,29 +71,10 @@ public class NullableBuffer extends LiveResourceSupport<NullableBuffer, Nullable
@Override
protected Owned<NullableBuffer> 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<NullableBuffer> {
private final Drop<NullableBuffer> delegate;
public CloseOnDrop(Drop<NullableBuffer> 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);
}
}
}

10
src/test/java/it/cavallium/dbengine/DbTestUtils.java
View File

@ -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
);
}
}