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

Bugfixes

Browse Source
This commit is contained in:
Andrea Cavalli 2021-08-22 23:50:50 +02:00
parent 2a24570512
commit c5552d2827
4 changed files with 31 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/main/java/it/cavallium/dbengine/database/LLUtils.java
View File

@ -531,7 +531,8 @@ public class LLUtils {
public static <T> Flux<T> handleDiscard(Flux<T> mono) { public static <T> Flux<T> handleDiscard(Flux<T> mono) {
return mono return mono
.doOnDiscard(ReferenceCounted.class, LLUtils::discardRefCounted) .doOnDiscard(ReferenceCounted.class, LLUtils::discardRefCounted)
.doOnDiscard(Map.Entry.class, LLUtils::discardEntry); .doOnDiscard(Map.Entry.class, LLUtils::discardEntry)
.doOnDiscard(Collection.class, LLUtils::discardCollection);
} }
private static void discardEntry(Map.Entry<?, ?> e) { private static void discardEntry(Map.Entry<?, ?> e) {
@ -552,4 +553,27 @@ public class LLUtils {
referenceCounted.release(); referenceCounted.release();
} }
} }
private static void discardCollection(Collection<?> collection) {
for (Object o : collection) {
if (o instanceof ReferenceCounted referenceCounted) {
if (referenceCounted.refCnt() > 0) {
referenceCounted.release();
}
} else if (o instanceof Map.Entry entry) {
if (entry.getKey() instanceof ReferenceCounted bb) {
if (bb.refCnt() > 0) {
bb.release();
}
}
if (entry.getValue() instanceof ReferenceCounted bb) {
if (bb.refCnt() > 0) {
bb.release();
}
}
} else {
break;
}
}
}
} }

17
src/main/java/it/cavallium/dbengine/database/disk/LLLocalDictionary.java
View File

@ -1093,22 +1093,7 @@ public class LLLocalDictionary implements LLDictionary {
} }
} }
), 2) // Max concurrency is 2 to read data while preparing the next segment ), 2) // Max concurrency is 2 to read data while preparing the next segment
.doOnDiscard(Entry.class, entry -> { .transform(LLUtils::handleDiscard);
if (entry.getKey() instanceof ByteBuf && entry.getValue() instanceof ByteBuf) {
//noinspection unchecked
var castedEntry = (Entry<ByteBuf, ByteBuf>) entry;
castedEntry.getKey().release();
castedEntry.getValue().release();
}
})
.doOnDiscard(Collection.class, obj -> {
//noinspection unchecked
var castedEntries = (Collection<Entry<ByteBuf, ByteBuf>>) obj;
for (Entry<ByteBuf, ByteBuf> entry : castedEntries) {
entry.getKey().release();
entry.getValue().release();
}
});
} }
@Override @Override

3
src/main/java/it/cavallium/dbengine/database/serialization/SerializationException.java
View File

@ -1,9 +1,8 @@
package it.cavallium.dbengine.database.serialization; package it.cavallium.dbengine.database.serialization;
import java.io.IOException; import java.io.IOException;
import java.io.UncheckedIOException;
public class SerializationException extends Exception { public class SerializationException extends IOException {
public SerializationException() { public SerializationException() {
super(); super();

6
src/test/java/it/cavallium/dbengine/TestDictionaryMap.java
View File

@ -432,8 +432,10 @@ public class TestDictionaryMap {
@ParameterizedTest @ParameterizedTest
@MethodSource("provideArgumentsPutMulti") @MethodSource("provideArgumentsPutMulti")
public void testSetAndGetChanged(DbType dbType, UpdateMode updateMode, Map<String, String> entries, boolean shouldFail) { public void testSetAndGetChanged(DbType dbType,
var remainingEntries = new ConcurrentHashMap<Entry<String, String>, Boolean>().keySet(true); UpdateMode updateMode,
Map<String, String> entries,
boolean shouldFail) {
Step<Boolean> stpVer = StepVerifier Step<Boolean> stpVer = StepVerifier
.create(tempDb(db -> tempDictionary(db, updateMode) .create(tempDb(db -> tempDictionary(db, updateMode)
.map(dict -> tempDatabaseMapDictionaryMap(dict, dbType, 5)) .map(dict -> tempDatabaseMapDictionaryMap(dict, dbType, 5))