stepus53/dhfs
1
0
Fork 0
mirror of https://github.com/usatiuk/dhfs.git synced 2025-10-28 20:47:49 +01:00
Code Issues Packages Projects Releases Wiki Activity

Simplify ObjectPersistentStore

Browse Source
This commit is contained in:
Stepan Usatiuk
2025-04-26 11:11:51 +02:00
parent f7338f4e80
commit a8cf483eee
5 changed files with 47 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
dhfs-parent/objects/src/main/java/com/usatiuk/objects/stores/LmdbObjectPersistentStore.java
View File

@@ -10,7 +10,6 @@ import com.usatiuk.objects.iterators.IteratorStart;
import com.usatiuk.objects.iterators.KeyPredicateKvIterator;
import com.usatiuk.objects.iterators.ReversibleKvIterator;
import com.usatiuk.objects.snapshot.Snapshot;
import com.usatiuk.utils.RefcountedCloseable;
import io.quarkus.arc.properties.IfBuildProperty;
import io.quarkus.logging.Log;
import io.quarkus.runtime.ShutdownEvent;
@@ -104,53 +103,48 @@ public class LmdbObjectPersistentStore implements ObjectPersistentStore {
if (!_ready) throw new IllegalStateException("Wrong service order!");
}
@Nonnull
@Override
public Optional<ByteString> readObject(JObjectKey name) {
verifyReady();
try (Txn<ByteBuffer> txn = _env.txnRead()) {
var value = _db.get(txn, name.toByteBuffer());
return Optional.ofNullable(value).map(ByteString::copyFrom);
}
}
@Override
public Snapshot<JObjectKey, ByteString> getSnapshot() {
var txn = new RefcountedCloseable<>(_env.txnRead());
long commitId = readTxId(txn.get()).orElseThrow();
return new Snapshot<JObjectKey, ByteString>() {
private final RefcountedCloseable<Txn<ByteBuffer>> _txn = txn;
private final long _id = commitId;
private boolean _closed = false;
var txn = _env.txnRead();
try {
long commitId = readTxId(txn).orElseThrow();
return new Snapshot<JObjectKey, ByteString>() {
private final Txn<ByteBuffer> _txn = txn;
private final long _id = commitId;
private boolean _closed = false;
@Override
public CloseableKvIterator<JObjectKey, ByteString> getIterator(IteratorStart start, JObjectKey key) {
assert !_closed;
return new KeyPredicateKvIterator<>(new LmdbKvIterator(_txn.ref(), start, key), start, key, (k) -> !k.value().equals(DB_VER_OBJ_NAME_STR));
}
@Override
public CloseableKvIterator<JObjectKey, ByteString> getIterator(IteratorStart start, JObjectKey key) {
assert !_closed;
return new KeyPredicateKvIterator<>(new LmdbKvIterator(_txn, start, key), start, key, (k) -> !k.value().equals(DB_VER_OBJ_NAME_STR));
}
@Nonnull
@Override
public Optional<ByteString> readObject(JObjectKey name) {
assert !_closed;
var got = _db.get(_txn.get(), name.toByteBuffer());
var ret = Optional.ofNullable(got).map(UnsafeByteOperations::unsafeWrap);
return ret;
}
@Nonnull
@Override
public Optional<ByteString> readObject(JObjectKey name) {
assert !_closed;
var got = _db.get(_txn, name.toByteBuffer());
var ret = Optional.ofNullable(got).map(UnsafeByteOperations::unsafeWrap);
return ret;
}
@Override
public long id() {
assert !_closed;
return _id;
}
@Override
public long id() {
assert !_closed;
return _id;
}
@Override
public void close() {
assert !_closed;
_closed = true;
_txn.unref();
}
};
@Override
public void close() {
assert !_closed;
_closed = true;
_txn.close();
}
};
} catch (Exception e) {
txn.close();
throw e;
}
}
@Override
@@ -205,28 +199,28 @@ public class LmdbObjectPersistentStore implements ObjectPersistentStore {
private class LmdbKvIterator extends ReversibleKvIterator<JObjectKey, ByteString> {
private static final Cleaner CLEANER = Cleaner.create();
private final RefcountedCloseable<Txn<ByteBuffer>> _txn;
private final Txn<ByteBuffer> _txn; // Managed by the snapshot
private final Cursor<ByteBuffer> _cursor;
private final MutableObject<Boolean> _closed = new MutableObject<>(false);
// private final Exception _allocationStacktrace = new Exception();
private final Exception _allocationStacktrace = null;
// private final Exception _allocationStacktrace = null;
private boolean _hasNext = false;
private JObjectKey _peekedNextKey = null;
LmdbKvIterator(RefcountedCloseable<Txn<ByteBuffer>> txn, IteratorStart start, JObjectKey key) {
LmdbKvIterator(Txn<ByteBuffer> txn, IteratorStart start, JObjectKey key) {
_txn = txn;
_goingForward = true;
_cursor = _db.openCursor(_txn.get());
_cursor = _db.openCursor(_txn);
var closedRef = _closed;
var bt = _allocationStacktrace;
CLEANER.register(this, () -> {
if (!closedRef.getValue()) {
Log.error("Iterator was not closed before GC, allocated at: {0}", bt);
System.exit(-1);
}
});
// var bt = _allocationStacktrace;
// CLEANER.register(this, () -> {
// if (!closedRef.getValue()) {
// Log.error("Iterator was not closed before GC, allocated at: {0}", bt);
// System.exit(-1);
// }
// });
verifyReady();
@@ -308,7 +302,6 @@ public class LmdbObjectPersistentStore implements ObjectPersistentStore {
}
_closed.setValue(true);
_cursor.close();
_txn.unref();
}
@Override

8
dhfs-parent/objects/src/main/java/com/usatiuk/objects/stores/MemoryObjectPersistentStore.java
View File

@@ -21,14 +21,6 @@ public class MemoryObjectPersistentStore implements ObjectPersistentStore {
private TreePMap<JObjectKey, ByteString> _objects = TreePMap.empty();
private long _lastCommitId = 0;
@Nonnull
@Override
public Optional<ByteString> readObject(JObjectKey name) {
synchronized (this) {
return Optional.ofNullable(_objects.get(name));
}
}
@Override
public Snapshot<JObjectKey, ByteString> getSnapshot() {
synchronized (this) {

3
dhfs-parent/objects/src/main/java/com/usatiuk/objects/stores/ObjectPersistentStore.java
View File

@@ -10,9 +10,6 @@ import java.util.Optional;
// Persistent storage of objects
// All changes are written as sequential transactions
public interface ObjectPersistentStore {
@Nonnull
Optional<ByteString> readObject(JObjectKey name);
Snapshot<JObjectKey, ByteString> getSnapshot();
Runnable prepareTx(TxManifestRaw names, long txId);

5
dhfs-parent/objects/src/main/java/com/usatiuk/objects/stores/SerializingObjectPersistentStore.java
View File

@@ -23,11 +23,6 @@ public class SerializingObjectPersistentStore {
@Inject
ObjectPersistentStore delegateStore;
@Nonnull
Optional<JDataVersionedWrapper> readObject(JObjectKey name) {
return delegateStore.readObject(name).map(serializer::deserialize);
}
public Snapshot<JObjectKey, JDataVersionedWrapper> getSnapshot() {
return new Snapshot<JObjectKey, JDataVersionedWrapper>() {
private final Snapshot<JObjectKey, ByteString> _backing = delegateStore.getSnapshot();

53
dhfs-parent/utils/src/main/java/com/usatiuk/utils/RefcountedCloseable.java
View File

@@ -1,53 +0,0 @@
package com.usatiuk.utils;
import io.quarkus.logging.Log;
import org.apache.commons.lang3.mutable.MutableObject;
import java.lang.ref.Cleaner;
public class RefcountedCloseable<T extends AutoCloseable> {
private static final Cleaner CLEANER = Cleaner.create();
private final T _closeable;
private final MutableObject<Boolean> _closed = new MutableObject<>(false);
private int _refCount = 1;
public RefcountedCloseable(T closeable) {
_closeable = closeable;
var closedRef = _closed;
CLEANER.register(this, () -> {
if (!closedRef.getValue()) {
Log.error("RefcountedCloseable was not closed before GC");
System.exit(-1);
}
});
}
public RefcountedCloseable<T> ref() {
synchronized (this) {
if (_closed.getValue()) {
return null;
}
_refCount++;
return this;
}
}
public void unref() {
synchronized (this) {
_refCount--;
if (_refCount == 0) {
try {
assert !_closed.getValue();
_closed.setValue(true);
_closeable.close();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
}
public T get() {
return _closeable;
}
}