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

Implement closeRequested

Browse Source
This commit is contained in:
Andrea Cavalli 2022-06-08 18:52:15 +02:00
parent 3be0d3710c
commit 563defb2ff
1 changed files with 52 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
src/main/java/it/cavallium/dbengine/database/disk/LLLocalKeyValueDatabase.java
View File

@ -130,6 +130,7 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
private final ConcurrentHashMap<Long, Snapshot> snapshotsHandles = new ConcurrentHashMap<>(); private final ConcurrentHashMap<Long, Snapshot> snapshotsHandles = new ConcurrentHashMap<>();
private final AtomicLong nextSnapshotNumbers = new AtomicLong(1); private final AtomicLong nextSnapshotNumbers = new AtomicLong(1);
private final StampedLock closeLock = new StampedLock(); private final StampedLock closeLock = new StampedLock();
private volatile boolean closeRequested = false;
private volatile boolean closed = false; private volatile boolean closed = false;
@SuppressWarnings("SwitchStatementWithTooFewBranches") @SuppressWarnings("SwitchStatementWithTooFewBranches")
@ -189,6 +190,7 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
} }
} }
var rocksLogger = new RocksLog4jLogger(rocksdbOptions, logger);
this.persistentCaches = new HashMap<>(); this.persistentCaches = new HashMap<>();
for (Column column : columns) { for (Column column : columns) {
@ -361,7 +363,8 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
rocksdbOptions, rocksdbOptions,
databaseOptions.persistentCaches(), databaseOptions.persistentCaches(),
columnOptions.persistentCacheId(), columnOptions.persistentCacheId(),
refs refs,
rocksLogger
)); ));
columnFamilyOptions.setTableFormatConfig(tableOptions); columnFamilyOptions.setTableFormatConfig(tableOptions);
@ -560,7 +563,8 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
DBOptions rocksdbOptions, DBOptions rocksdbOptions,
List<it.cavallium.dbengine.rpc.current.data.PersistentCache> persistentCaches, List<it.cavallium.dbengine.rpc.current.data.PersistentCache> persistentCaches,
NullableString persistentCacheId, NullableString persistentCacheId,
RocksDBRefs refs) throws RocksDBException { RocksDBRefs refs,
RocksLog4jLogger rocksLogger) throws RocksDBException {
if (persistentCacheId.isEmpty()) { if (persistentCacheId.isEmpty()) {
return null; return null;
} }
@ -581,7 +585,7 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
var persistentCache = new PersistentCache(Env.getDefault(), var persistentCache = new PersistentCache(Env.getDefault(),
foundCache.path(), foundCache.path(),
foundCache.size(), foundCache.size(),
new RocksLog4jLogger(rocksdbOptions, logger), rocksLogger,
foundCache.optimizeForNvm() foundCache.optimizeForNvm()
); );
refs.track(persistentCache); refs.track(persistentCache);
@ -1057,14 +1061,13 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
List.of(Tag.of("db.name", dbName), Tag.of("db.statistics.name", tickerType.name())), List.of(Tag.of("db.name", dbName), Tag.of("db.statistics.name", tickerType.name())),
stats, stats,
statistics -> { statistics -> {
if (closed) { if (closeRequested || closed) return 0d;
return 0d; long closeReadLock = 0;
}
var closeReadLock = closeLock.readLock();
try { try {
if (closed) { closeReadLock = closeLock.tryReadLock(1, TimeUnit.SECONDS);
return 0d; } catch (InterruptedException ignored) {}
} try {
if (closeRequested || closed || closeReadLock == 0) return 0d;
return statistics.getTickerCount(tickerType); return statistics.getTickerCount(tickerType);
} finally { } finally {
closeLock.unlockRead(closeReadLock); closeLock.unlockRead(closeReadLock);
@ -1247,10 +1250,14 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
public Mono<MemoryStats> getMemoryStats() { public Mono<MemoryStats> getMemoryStats() {
return Mono return Mono
.fromCallable(() -> { .fromCallable(() -> {
if (closed) return null; if (closeRequested || closed) return null;
var closeReadLock = closeLock.readLock(); long closeReadLock = 0;
try { try {
if (!closed) { //noinspection BlockingMethodInNonBlockingContext
closeReadLock = closeLock.tryReadLock(1, TimeUnit.SECONDS);
} catch (InterruptedException ignored) {}
try {
if (closeRequested || closed || closeReadLock == 0) return null;
ensureOpen(); ensureOpen();
return new MemoryStats(db.getAggregatedLongProperty("rocksdb.estimate-table-readers-mem"), return new MemoryStats(db.getAggregatedLongProperty("rocksdb.estimate-table-readers-mem"),
db.getAggregatedLongProperty("rocksdb.size-all-mem-tables"), db.getAggregatedLongProperty("rocksdb.size-all-mem-tables"),
@ -1259,9 +1266,6 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
db.getAggregatedLongProperty("rocksdb.block-cache-usage") / this.handles.size(), db.getAggregatedLongProperty("rocksdb.block-cache-usage") / this.handles.size(),
db.getAggregatedLongProperty("rocksdb.block-cache-pinned-usage") / this.handles.size() db.getAggregatedLongProperty("rocksdb.block-cache-pinned-usage") / this.handles.size()
); );
} else {
return null;
}
} finally { } finally {
closeLock.unlockRead(closeReadLock); closeLock.unlockRead(closeReadLock);
} }
@ -1394,10 +1398,14 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
public Mono<String> getRocksDBStats() { public Mono<String> getRocksDBStats() {
return Mono return Mono
.fromCallable(() -> { .fromCallable(() -> {
if (closed) return null; if (closeRequested || closed) return null;
var closeReadLock = closeLock.readLock(); long closeReadLock = 0;
try { try {
if (!closed) { //noinspection BlockingMethodInNonBlockingContext
closeReadLock = closeLock.tryReadLock(1, TimeUnit.SECONDS);
} catch (InterruptedException ignored) {}
try {
if (closeRequested || closed || closeReadLock == 0) return null;
ensureOpen(); ensureOpen();
StringBuilder aggregatedStats = new StringBuilder(); StringBuilder aggregatedStats = new StringBuilder();
for (var entry : this.handles.entrySet()) { for (var entry : this.handles.entrySet()) {
@ -1408,9 +1416,6 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
.append("\n"); .append("\n");
} }
return aggregatedStats.toString(); return aggregatedStats.toString();
} else {
return null;
}
} finally { } finally {
closeLock.unlockRead(closeReadLock); closeLock.unlockRead(closeReadLock);
} }
@ -1425,10 +1430,14 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
.fromIterable(handles.entrySet()) .fromIterable(handles.entrySet())
.flatMapSequential(handle -> Mono .flatMapSequential(handle -> Mono
.fromCallable(() -> { .fromCallable(() -> {
if (closed) return null; if (closeRequested || closed) return null;
var closeReadLock = closeLock.readLock(); long closeReadLock = 0;
try { try {
if (closed) return null; //noinspection BlockingMethodInNonBlockingContext
closeReadLock = closeLock.tryReadLock(1, TimeUnit.SECONDS);
} catch (InterruptedException ignored) {}
try {
if (closeRequested || closed || closeReadLock == 0) return null;
ensureOpen(); ensureOpen();
return db.getPropertiesOfAllTables(handle.getValue()); return db.getPropertiesOfAllTables(handle.getValue());
} finally { } finally {
@ -1531,6 +1540,7 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
public Mono<Void> close() { public Mono<Void> close() {
return Mono return Mono
.<Void>fromCallable(() -> { .<Void>fromCallable(() -> {
closeRequested = true;
if (statistics != null) { if (statistics != null) {
statistics.close(); statistics.close();
statistics = null; statistics = null;