Lazy retain generated suffixes
This commit is contained in:
parent
bc759c344d
commit
906379923f
@ -16,6 +16,8 @@ import java.util.Collection;
|
|||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map.Entry;
|
import java.util.Map.Entry;
|
||||||
import java.util.Objects;
|
import java.util.Objects;
|
||||||
|
import java.util.concurrent.Callable;
|
||||||
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
import java.util.function.Function;
|
import java.util.function.Function;
|
||||||
import java.util.function.ToIntFunction;
|
import java.util.function.ToIntFunction;
|
||||||
import org.apache.lucene.document.Document;
|
import org.apache.lucene.document.Document;
|
||||||
@ -493,7 +495,23 @@ public class LLUtils {
|
|||||||
return Mono.just(buf).map(ByteBuf::retain);
|
return Mono.just(buf).map(ByteBuf::retain);
|
||||||
}
|
}
|
||||||
|
|
||||||
public static Mono<LLRange> lazyRetain(LLRange range) {
|
public static Mono<LLRange> lazyRetainRange(LLRange range) {
|
||||||
return Mono.just(range).map(LLRange::retain);
|
return Mono.just(range).map(LLRange::retain);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public static Mono<ByteBuf> lazyRetain(Callable<ByteBuf> bufCallable) {
|
||||||
|
return Mono.fromCallable(bufCallable).cacheInvalidateIf(byteBuf -> {
|
||||||
|
// Retain if the value has been cached previously
|
||||||
|
byteBuf.retain();
|
||||||
|
return false;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
public static Mono<LLRange> lazyRetainRange(Callable<LLRange> rangeCallable) {
|
||||||
|
return Mono.fromCallable(rangeCallable).cacheInvalidateIf(range -> {
|
||||||
|
// Retain if the value has been cached previously
|
||||||
|
range.retain();
|
||||||
|
return false;
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
@ -269,7 +269,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> implem
|
|||||||
assert keyPrefix.refCnt() > 0;
|
assert keyPrefix.refCnt() > 0;
|
||||||
assert keyPrefixLength == 0 || !LLUtils.equals(firstKey, nextRangeKey);
|
assert keyPrefixLength == 0 || !LLUtils.equals(firstKey, nextRangeKey);
|
||||||
this.range = keyPrefixLength == 0 ? LLRange.all() : LLRange.of(firstKey.retain(), nextRangeKey.retain());
|
this.range = keyPrefixLength == 0 ? LLRange.all() : LLRange.of(firstKey.retain(), nextRangeKey.retain());
|
||||||
this.rangeMono = LLUtils.lazyRetain(this.range);
|
this.rangeMono = LLUtils.lazyRetainRange(this.range);
|
||||||
assert subStageKeysConsistency(keyPrefixLength + keySuffixLength + keyExtLength);
|
assert subStageKeysConsistency(keyPrefixLength + keySuffixLength + keyExtLength);
|
||||||
} finally {
|
} finally {
|
||||||
nextRangeKey.release();
|
nextRangeKey.release();
|
||||||
@ -389,34 +389,33 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> implem
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Mono<US> at(@Nullable CompositeSnapshot snapshot, T keySuffix) {
|
public Mono<US> at(@Nullable CompositeSnapshot snapshot, T keySuffix) {
|
||||||
return Mono
|
return Mono.using(
|
||||||
.using(
|
() -> serializeSuffix(keySuffix),
|
||||||
() -> serializeSuffix(keySuffix),
|
keySuffixData -> {
|
||||||
keySuffixData -> {
|
Flux<ByteBuf> debuggingKeysFlux = Mono.<List<ByteBuf>>defer(() -> {
|
||||||
Flux<ByteBuf> debuggingKeysFlux = Mono
|
if (LLLocalDictionary.DEBUG_PREFIXES_WHEN_ASSERTIONS_ARE_ENABLED
|
||||||
.<List<ByteBuf>>defer(() -> {
|
&& this.subStageGetter.needsDebuggingKeyFlux()) {
|
||||||
if (LLLocalDictionary.DEBUG_PREFIXES_WHEN_ASSERTIONS_ARE_ENABLED
|
return Flux
|
||||||
&& this.subStageGetter.needsDebuggingKeyFlux()) {
|
.using(
|
||||||
return Flux
|
() -> toExtRange(keySuffixData.retain()),
|
||||||
.using(
|
extRangeBuf -> this.dictionary
|
||||||
() -> toExtRange(keySuffixData.retain()),
|
.getRangeKeys(resolveSnapshot(snapshot), LLUtils.lazyRetainRange(extRangeBuf)),
|
||||||
extRangeBuf -> this.dictionary
|
LLRange::release
|
||||||
.getRangeKeys(resolveSnapshot(snapshot), LLUtils.lazyRetain(extRangeBuf)),
|
)
|
||||||
LLRange::release
|
.collectList();
|
||||||
)
|
} else {
|
||||||
.collectList();
|
return Mono.just(List.of());
|
||||||
} else {
|
}
|
||||||
return Mono.just(List.of());
|
}).flatMapIterable(it -> it);
|
||||||
}
|
return Mono.using(
|
||||||
})
|
() -> toKeyWithoutExt(keySuffixData.retain()),
|
||||||
.flatMapIterable(it -> it);
|
keyWithoutExt -> this.subStageGetter
|
||||||
Mono<ByteBuf> keyBufMono = LLUtils.lazyRetain(toKeyWithoutExt(keySuffixData.retain()));
|
.subStage(dictionary, snapshot, LLUtils.lazyRetain(keyWithoutExt), debuggingKeysFlux),
|
||||||
return this.subStageGetter
|
ReferenceCounted::release
|
||||||
.subStage(dictionary, snapshot, keyBufMono, debuggingKeysFlux);
|
);
|
||||||
},
|
},
|
||||||
ReferenceCounted::release
|
ReferenceCounted::release
|
||||||
)
|
).doOnDiscard(DatabaseStage.class, DatabaseStage::release);
|
||||||
.doOnDiscard(DatabaseStage.class, DatabaseStage::release);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -538,7 +537,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> implem
|
|||||||
.doOnNext(ReferenceCounted::release)
|
.doOnNext(ReferenceCounted::release)
|
||||||
.then();
|
.then();
|
||||||
} else {
|
} else {
|
||||||
return dictionary.setRange(LLUtils.lazyRetain(range), Flux.empty());
|
return dictionary.setRange(LLUtils.lazyRetainRange(range), Flux.empty());
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user