dump

oops

dhfs-parent/.run/Main.run.xml

@@ -1,18 +1,16 @@
 <component name="ProjectRunConfigurationManager">
-    <configuration default="false" name="Main" type="QsApplicationConfigurationType" factoryName="QuarkusApplication"
-                   nameIsGenerated="true">
-        <option name="MAIN_CLASS_NAME" value="com.usatiuk.dhfsapp.Main"/>
-        <module name="dhfs-app"/>
-        <option name="VM_PARAMETERS"
-                value="-XX:+UnlockDiagnosticVMOptions -XX:+UseParallelGC -XX:+DebugNonSafepoints --add-exports java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-exports java.base/jdk.internal.access=ALL-UNNAMED -ea -Xmx2G -Ddhfs.webui.root=$ProjectFileDir$/../webui/dist -Ddhfs.fuse.root=${HOME}/dhfs_test/1/fuse -Ddhfs.objects.persistence.files.root=${HOME}/dhfs_test/1/data -Ddhfs.objects.persistence.stuff.root=${HOME}/dhfs_test/1/data/stuff -Ddhfs.objects.peerdiscovery.broadcast=true -Dquarkus.http.port=9010 -Dquarkus.http.ssl-port=9011 -Ddhfs.peerdiscovery.preset-uuid=11000000-0000-0000-0000-000000000000 -Ddhfs.peerdiscovery.static-peers=22000000-0000-0000-0000-000000000000:127.0.0.1:9020:9021 -Dquarkus.http.host=0.0.0.0"/>
-        <extension name="coverage">
-            <pattern>
-                <option name="PATTERN" value="com.usatiuk.dhfs.remoteobj.*"/>
-                <option name="ENABLED" value="true"/>
-            </pattern>
-        </extension>
-        <method v="2">
-            <option name="Make" enabled="true"/>
-        </method>
-    </configuration>
+  <configuration default="false" name="Main" type="QsApplicationConfigurationType" factoryName="QuarkusApplication" nameIsGenerated="true">
+    <option name="MAIN_CLASS_NAME" value="com.usatiuk.dhfsapp.Main" />
+    <module name="dhfs-app" />
+    <option name="VM_PARAMETERS" value="-XX:+UnlockDiagnosticVMOptions -XX:+UseParallelGC --enable-preview -XX:+DebugNonSafepoints --add-exports java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-exports java.base/jdk.internal.access=ALL-UNNAMED -ea -Xmx2G -Ddhfs.webui.root=$ProjectFileDir$/../webui/dist -Ddhfs.fuse.root=${HOME}/dhfs_test/1/fuse -Ddhfs.objects.persistence.files.root=${HOME}/dhfs_test/1/data -Ddhfs.objects.persistence.stuff.root=${HOME}/dhfs_test/1/data/stuff -Ddhfs.objects.peerdiscovery.broadcast=true -Dquarkus.http.port=9010 -Dquarkus.http.ssl-port=9011 -Ddhfs.peerdiscovery.preset-uuid=11000000-0000-0000-0000-000000000000 -Ddhfs.peerdiscovery.static-peers=22000000-0000-0000-0000-000000000000:127.0.0.1:9020:9021 -Dquarkus.http.host=0.0.0.0" />
+    <extension name="coverage">
+      <pattern>
+        <option name="PATTERN" value="com.usatiuk.dhfs.remoteobj.*" />
+        <option name="ENABLED" value="true" />
+      </pattern>
+    </extension>
+    <method v="2">
+      <option name="Make" enabled="true" />
+    </method>
+  </configuration>
 </component>

dhfs-parent/dhfs-app/src/main/resources/application.properties

@@ -14,8 +14,9 @@ dhfs.objects.persistence.stuff.root=${HOME}/dhfs_default/data/stuff
 dhfs.fuse.debug=false
 dhfs.fuse.enabled=true
 dhfs.files.allow_recursive_delete=false
-dhfs.files.target_chunk_size=2097152
-dhfs.files.target_chunk_alignment=19
+dhfs.files.target_chunk_size=524288
+dhfs.files.max_chunk_size=524288
+dhfs.files.target_chunk_alignment=17
 dhfs.objects.deletion.delay=1000
 dhfs.objects.deletion.can-delete-retry-delay=10000
 dhfs.objects.ref_verification=true
@@ -28,7 +29,7 @@ dhfs.objects.opsender.batch-size=100
 dhfs.objects.lock_timeout_secs=2
 dhfs.local-discovery=true
 dhfs.peerdiscovery.timeout=10000
-quarkus.log.category."com.usatiuk".min-level=TRACE
-quarkus.log.category."com.usatiuk".level=TRACE
+quarkus.log.category."com.usatiuk".min-level=INFO
+quarkus.log.category."com.usatiuk".level=INFO
 quarkus.http.insecure-requests=enabled
 quarkus.http.ssl.client-auth=required

dhfs-parent/dhfs-fs/src/main/java/com/usatiuk/dhfsfs/service/DhfsFileServiceImpl.java

@@ -55,6 +55,9 @@ public class DhfsFileServiceImpl implements DhfsFileService {
     @ConfigProperty(name = "dhfs.files.target_chunk_size")
     int targetChunkSize;
 
+    @ConfigProperty(name = "dhfs.files.max_chunk_size")
+    int maxChunkSize;
+
     @ConfigProperty(name = "dhfs.files.use_hash_for_chunks")
     boolean useHashForChunks;
 
@@ -83,7 +86,7 @@ public class DhfsFileServiceImpl implements DhfsFileService {
 
     private ChunkData createChunk(ByteString bytes) {
         var newChunk = new ChunkData(JObjectKey.of(UUID.randomUUID().toString()), bytes);
-        remoteTx.putData(newChunk);
+        remoteTx.putDataNew(newChunk);
         return newChunk;
     }
 
@@ -360,16 +363,10 @@ public class DhfsFileServiceImpl implements DhfsFileService {
 
             var file = remoteTx.getData(File.class, fileUuid, LockingStrategy.WRITE).orElse(null);
             if (file == null) {
-                Log.error("File not found when trying to write: " + fileUuid);
-                return -1L;
+                throw new StatusRuntimeException(Status.NOT_FOUND.withDescription("File not found when trying to write: " + fileUuid));
             }
 
-            if (writeLogging) {
-                Log.info("Writing to file: " + file.key() + " size=" + size(fileUuid) + " "
-                        + offset + " " + data.size());
-            }
-
-            NavigableMap<Long, JObjectKey> removedChunks = new TreeMap<>();
+            Map<Long, JObjectKey> removedChunks = new HashMap<>();
 
             long realOffset = targetChunkAlignment >= 0 ? alignDown(offset, targetChunkAlignment) : offset;
             long writeEnd = offset + data.size();
@@ -407,7 +404,7 @@ public class DhfsFileServiceImpl implements DhfsFileService {
             }
 
 
-            NavigableMap<Long, JObjectKey> newChunks = new TreeMap<>();
+            Map<Long, JObjectKey> newChunks = new HashMap<>();
 
             if (existingEnd < offset) {
                 if (!pendingPrefix.isEmpty()) {
@@ -424,12 +421,13 @@ public class DhfsFileServiceImpl implements DhfsFileService {
             int combinedSize = pendingWrites.size();
 
             {
-                int targetChunkSize = 1 << targetChunkAlignment;
                 int cur = 0;
                 while (cur < combinedSize) {
                     int end;
 
-                    if (targetChunkAlignment < 0)
+                    if (combinedSize - cur < maxChunkSize)
+                        end = combinedSize;
+                    else if (targetChunkAlignment < 0)
                         end = combinedSize;
                     else
                         end = Math.min(cur + targetChunkSize, combinedSize);
@@ -550,7 +548,7 @@ public class DhfsFileServiceImpl implements DhfsFileService {
         });
     }
 
-    private void fillZeros(long fillStart, long length, NavigableMap<Long, JObjectKey> newChunks) {
+    private void fillZeros(long fillStart, long length, Map<Long, JObjectKey> newChunks) {
         long combinedSize = (length - fillStart);
 
         long start = fillStart;

dhfs-parent/dhfs-fuse/src/main/java/com/usatiuk/dhfsfuse/DhfsFuse.java

@@ -7,6 +7,8 @@ import com.usatiuk.dhfsfs.service.DirectoryNotEmptyException;
 import com.usatiuk.dhfsfs.service.GetattrRes;
 import com.usatiuk.kleppmanntree.AlreadyExistsException;
 import com.usatiuk.objects.JObjectKey;
+import com.usatiuk.utils.UninitializedByteBuffer;
+import com.usatiuk.utils.UnsafeAccessor;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.quarkus.logging.Log;
@@ -27,7 +29,6 @@ import ru.serce.jnrfuse.struct.FuseFileInfo;
 import ru.serce.jnrfuse.struct.Statvfs;
 import ru.serce.jnrfuse.struct.Timespec;
 
-import java.nio.ByteBuffer;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Optional;
@@ -51,8 +52,6 @@ public class DhfsFuse extends FuseStubFS {
     @ConfigProperty(name = "dhfs.files.target_chunk_size")
     int targetChunkSize;
     @Inject
-    JnrPtrByteOutputAccessors jnrPtrByteOutputAccessors;
-    @Inject
     DhfsFileService fileService;
 
     private long allocateHandle(JObjectKey key) {
@@ -231,7 +230,7 @@ public class DhfsFuse extends FuseStubFS {
             var fileKey = getFromHandle(fi.fh.get());
             var read = fileService.read(fileKey, offset, (int) size);
             if (read.isEmpty()) return 0;
-            UnsafeByteOperations.unsafeWriteTo(read, new JnrPtrByteOutput(jnrPtrByteOutputAccessors, buf, size));
+            UnsafeByteOperations.unsafeWriteTo(read, new JnrPtrByteOutput(buf, size));
             return read.size();
         } catch (Throwable e) {
             Log.error("When reading " + path, e);
@@ -244,21 +243,17 @@ public class DhfsFuse extends FuseStubFS {
         if (offset < 0) return -ErrorCodes.EINVAL();
         try {
             var fileKey = getFromHandle(fi.fh.get());
-            var buffer = ByteBuffer.allocateDirect((int) size);
+            var buffer = UninitializedByteBuffer.allocate((int) size);
 
-            if (buffer.isDirect()) {
-                jnrPtrByteOutputAccessors.getUnsafe().copyMemory(
-                        buf.address(),
-                        jnrPtrByteOutputAccessors.getNioAccess().getBufferAddress(buffer),
-                        size
-                );
-            } else {
-                buf.get(0, buffer.array(), 0, (int) size);
-            }
+            UnsafeAccessor.get().getUnsafe().copyMemory(
+                    buf.address(),
+                    UnsafeAccessor.get().getNioAccess().getBufferAddress(buffer),
+                    size
+            );
 
             var written = fileService.write(fileKey, offset, UnsafeByteOperations.unsafeWrap(buffer));
             return written.intValue();
-        } catch (Throwable e) {
+        } catch (Exception e) {
             Log.error("When writing " + path, e);
             return -ErrorCodes.EIO();
         }
@@ -394,7 +389,7 @@ public class DhfsFuse extends FuseStubFS {
             var file = fileOpt.get();
             var read = fileService.readlinkBS(fileOpt.get());
             if (read.isEmpty()) return 0;
-            UnsafeByteOperations.unsafeWriteTo(read, new JnrPtrByteOutput(jnrPtrByteOutputAccessors, buf, size));
+            UnsafeByteOperations.unsafeWriteTo(read, new JnrPtrByteOutput(buf, size));
             buf.putByte(Math.min(size - 1, read.size()), (byte) 0);
             return 0;
         } catch (Throwable e) {

dhfs-parent/dhfs-fuse/src/main/java/com/usatiuk/dhfsfuse/JnrPtrByteOutput.java

@@ -1,6 +1,7 @@
 package com.usatiuk.dhfsfuse;
 
 import com.google.protobuf.ByteOutput;
+import com.usatiuk.utils.UnsafeAccessor;
 import jnr.ffi.Pointer;
 
 import java.nio.ByteBuffer;
@@ -9,14 +10,12 @@ import java.nio.MappedByteBuffer;
 public class JnrPtrByteOutput extends ByteOutput {
     private final Pointer _backing;
     private final long _size;
-    private final JnrPtrByteOutputAccessors _accessors;
     private long _pos;
 
-    public JnrPtrByteOutput(JnrPtrByteOutputAccessors accessors, Pointer backing, long size) {
+    public JnrPtrByteOutput(Pointer backing, long size) {
         _backing = backing;
         _size = size;
         _pos = 0;
-        _accessors = accessors;
     }
 
     @Override
@@ -47,9 +46,9 @@ public class JnrPtrByteOutput extends ByteOutput {
             if (value instanceof MappedByteBuffer mb) {
                 mb.load();
             }
-            long addr = _accessors.getNioAccess().getBufferAddress(value) + value.position();
+            long addr = UnsafeAccessor.get().getNioAccess().getBufferAddress(value) + value.position();
             var out = _backing.address() + _pos;
-            _accessors.getUnsafe().copyMemory(addr, out, rem);
+            UnsafeAccessor.get().getUnsafe().copyMemory(addr, out, rem);
         } else {
             _backing.put(_pos, value.array(), value.arrayOffset() + value.position(), rem);
         }

dhfs-parent/dhfs-fuse/src/main/java/com/usatiuk/dhfsfuse/JnrPtrByteOutputAccessors.java

@@ -1,29 +0,0 @@
-package com.usatiuk.dhfsfuse;
-
-import jakarta.inject.Singleton;
-import jdk.internal.access.JavaNioAccess;
-import jdk.internal.access.SharedSecrets;
-import sun.misc.Unsafe;
-
-import java.lang.reflect.Field;
-
-@Singleton
-class JnrPtrByteOutputAccessors {
-    JavaNioAccess _nioAccess;
-    Unsafe _unsafe;
-
-    JnrPtrByteOutputAccessors() throws NoSuchFieldException, IllegalAccessException {
-        _nioAccess = SharedSecrets.getJavaNioAccess();
-        Field f = Unsafe.class.getDeclaredField("theUnsafe");
-        f.setAccessible(true);
-        _unsafe = (Unsafe) f.get(null);
-    }
-
-    public JavaNioAccess getNioAccess() {
-        return _nioAccess;
-    }
-
-    public Unsafe getUnsafe() {
-        return _unsafe;
-    }
-}

dhfs-parent/objects/src/main/java/com/usatiuk/objects/JObjectKeyImpl.java

@@ -1,5 +1,7 @@
 package com.usatiuk.objects;
 
+import com.usatiuk.utils.UninitializedByteBuffer;
+
 import java.io.Serial;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
@@ -46,7 +48,7 @@ public final class JObjectKeyImpl implements JObjectKey {
         synchronized (this) {
             if (_bb != null) return _bb;
             var bytes = value.getBytes(StandardCharsets.ISO_8859_1);
-            var directBb = ByteBuffer.allocateDirect(bytes.length);
+            var directBb = UninitializedByteBuffer.allocate(bytes.length);
             directBb.put(bytes);
             directBb.flip();
             _bb = directBb;
@@ -69,7 +71,7 @@ public final class JObjectKeyImpl implements JObjectKey {
 
     @Override
     public int hashCode() {
-        return Objects.hash(value);
+        return value.hashCode();
     }
 
 }

dhfs-parent/objects/src/main/java/com/usatiuk/objects/iterators/IterProdFn2.java

@@ -0,0 +1,8 @@
+package com.usatiuk.objects.iterators;
+
+import java.util.stream.Stream;
+
+@FunctionalInterface
+public interface IterProdFn2<K extends Comparable<K>, V> {
+    Stream<CloseableKvIterator<K, MaybeTombstone<V>>> get(IteratorStart start, K key);
+}

dhfs-parent/objects/src/main/java/com/usatiuk/objects/iterators/MergingKvIterator.java

@@ -13,6 +13,7 @@ public class MergingKvIterator<K extends Comparable<K>, V> extends ReversibleKvI
     private final NavigableMap<K, IteratorEntry<K, V>> _sortedIterators = new TreeMap<>();
     private final String _name;
     private final List<IteratorEntry<K, V>> _iterators;
+
     public MergingKvIterator(String name, IteratorStart startType, K startKey, List<IterProdFn<K, V>> iterators) {
         _goingForward = true;
         _name = name;
@@ -73,6 +74,84 @@ public class MergingKvIterator<K extends Comparable<K>, V> extends ReversibleKvI
             advanceIterator(iterator);
         }
 
+//        Log.tracev("{0} Initialized: {1}", _name, _sortedIterators);
+//        switch (startType) {
+////            case LT -> {
+////                assert _sortedIterators.isEmpty() || _sortedIterators.firstKey().compareTo(initialStartKey) < 0;
+////            }
+////            case LE -> {
+////                assert _sortedIterators.isEmpty() || _sortedIterators.firstKey().compareTo(initialStartKey) <= 0;
+////            }
+//            case GT -> {
+//                assert _sortedIterators.isEmpty() || _sortedIterators.firstKey().compareTo(startKey) > 0;
+//            }
+//            case GE -> {
+//                assert _sortedIterators.isEmpty() || _sortedIterators.firstKey().compareTo(startKey) >= 0;
+//            }
+//        }
+    }
+
+    public MergingKvIterator(String name, IteratorStart startType, K startKey, IterProdFn2<K, V> iteratorsProd) {
+        _goingForward = true;
+        _name = name;
+
+        // Why streams are so slow?
+        {
+            var iterators = iteratorsProd.get(startType, startKey).toList();
+            IteratorEntry<K, V>[] iteratorEntries = new IteratorEntry[iterators.size()];
+            for (int i = 0; i < iterators.size(); i++) {
+                iteratorEntries[i] = new IteratorEntry<>(i, (CloseableKvIterator<K, V>) iterators.get(i));
+            }
+            _iterators = List.of(iteratorEntries);
+        }
+
+        if (startType == IteratorStart.LT || startType == IteratorStart.LE) {
+            // Starting at a greatest key less than/less or equal than:
+            // We have a bunch of iterators that have given us theirs "greatest LT/LE key"
+            // now we need to pick the greatest of those to start with
+            // But if some of them don't have a lesser key, we need to pick the smallest of those
+
+            K greatestLess = null;
+            K smallestMore = null;
+
+            for (var ite : _iterators) {
+                var it = ite.iterator();
+                if (it.hasNext()) {
+                    var peeked = it.peekNextKey();
+                    if (startType == IteratorStart.LE ? peeked.compareTo(startKey) <= 0 : peeked.compareTo(startKey) < 0) {
+                        if (greatestLess == null || peeked.compareTo(greatestLess) > 0) {
+                            greatestLess = peeked;
+                        }
+                    } else {
+                        if (smallestMore == null || peeked.compareTo(smallestMore) < 0) {
+                            smallestMore = peeked;
+                        }
+                    }
+                }
+            }
+
+            K initialMaxValue;
+            if (greatestLess != null)
+                initialMaxValue = greatestLess;
+            else
+                initialMaxValue = smallestMore;
+
+            if (initialMaxValue == null) {
+                // Empty iterators
+            }
+
+            for (var ite : _iterators) {
+                var iterator = ite.iterator();
+                while (iterator.hasNext() && iterator.peekNextKey().compareTo(initialMaxValue) < 0) {
+                    iterator.skip();
+                }
+            }
+        }
+
+        for (IteratorEntry<K, V> iterator : _iterators) {
+            advanceIterator(iterator);
+        }
+
 //        Log.tracev("{0} Initialized: {1}", _name, _sortedIterators);
 //        switch (startType) {
 ////            case LT -> {

dhfs-parent/objects/src/main/java/com/usatiuk/objects/iterators/TombstoneMergingKvIterator.java

@@ -19,4 +19,18 @@ public abstract class TombstoneMergingKvIterator {
     public static <K extends Comparable<K>, V> CloseableKvIterator<K, V> of(String name, IteratorStart startType, K startKey, IterProdFn<K, MaybeTombstone<V>>... iterators) {
         return of(name, startType, startKey, List.of(iterators));
     }
+
+    public static <K extends Comparable<K>, V> CloseableKvIterator<K, V> of(String name, IteratorStart startType, K startKey, IterProdFn2<K, V> itProd) {
+        return new PredicateKvIterator<K, MaybeTombstone<V>, V>(
+                new MergingKvIterator<K, MaybeTombstone<V>>(name + "-merging", startType, startKey, (IterProdFn2<K, MaybeTombstone<V>>) itProd),
+                startType, startKey,
+                pair -> {
+//                    Log.tracev("{0} - Processing pair {1}", name, pair);
+                    if (pair instanceof Tombstone<V>) {
+                        return null;
+                    }
+                    return ((Data<V>) pair).value();
+                });
+    }
+
 }

dhfs-parent/objects/src/main/java/com/usatiuk/objects/snapshot/Snapshot.java

@@ -2,13 +2,16 @@ package com.usatiuk.objects.snapshot;
 
 import com.usatiuk.objects.iterators.CloseableKvIterator;
 import com.usatiuk.objects.iterators.IteratorStart;
+import com.usatiuk.objects.iterators.MaybeTombstone;
+import com.usatiuk.objects.iterators.Tombstone;
 import com.usatiuk.utils.AutoCloseableNoThrow;
 
 import javax.annotation.Nonnull;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 public interface Snapshot<K extends Comparable<K>, V> extends AutoCloseableNoThrow {
-    CloseableKvIterator<K, V> getIterator(IteratorStart start, K key);
+    Stream<CloseableKvIterator<K, MaybeTombstone<V>>> getIterator(IteratorStart start, K key);
 
     @Nonnull
     Optional<V> readObject(K name);

dhfs-parent/objects/src/main/java/com/usatiuk/objects/stores/CachingObjectPersistentStore.java

@@ -21,6 +21,7 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
 
 @ApplicationScoped
 public class CachingObjectPersistentStore {
@@ -33,6 +34,7 @@ public class CachingObjectPersistentStore {
     private ExecutorService _statusExecutor;
     private AtomicLong _cached = new AtomicLong();
     private AtomicLong _cacheTries = new AtomicLong();
+
     public CachingObjectPersistentStore(@ConfigProperty(name = "dhfs.objects.lru.limit") int sizeLimit) {
         _cache = new AtomicReference<>(
                 new Cache(TreePMap.empty(), 0, -1, sizeLimit)
@@ -150,10 +152,11 @@ public class CachingObjectPersistentStore {
                     }
 
                     @Override
-                    public CloseableKvIterator<JObjectKey, JDataVersionedWrapper> getIterator(IteratorStart start, JObjectKey key) {
-                        return TombstoneMergingKvIterator.<JObjectKey, JDataVersionedWrapper>of("cache", start, key,
-                                (mS, mK) -> new NavigableMapKvIterator<JObjectKey, MaybeTombstone<JDataVersionedWrapper>>(_curCache.map(), mS, mK),
-                                (mS, mK) -> new CachingKvIterator(_backing.getIterator(start, key)));
+                    public Stream<CloseableKvIterator<JObjectKey, MaybeTombstone<JDataVersionedWrapper>>> getIterator(IteratorStart start, JObjectKey key) {
+                        return Stream.concat(
+                                Stream.of(new NavigableMapKvIterator<JObjectKey, MaybeTombstone<JDataVersionedWrapper>>(_curCache.map(), start, key)),
+                                _backing.getIterator(start, key).map(i -> new CachingKvIterator((CloseableKvIterator<JObjectKey, JDataVersionedWrapper>) (CloseableKvIterator<JObjectKey, ?>) i))
+                        );
                     }
 
                     @Nonnull

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

@@ -3,10 +3,7 @@ package com.usatiuk.objects.stores;
 import com.usatiuk.objects.JObjectKey;
 import com.usatiuk.objects.JObjectKeyMax;
 import com.usatiuk.objects.JObjectKeyMin;
-import com.usatiuk.objects.iterators.CloseableKvIterator;
-import com.usatiuk.objects.iterators.IteratorStart;
-import com.usatiuk.objects.iterators.KeyPredicateKvIterator;
-import com.usatiuk.objects.iterators.ReversibleKvIterator;
+import com.usatiuk.objects.iterators.*;
 import com.usatiuk.objects.snapshot.Snapshot;
 import io.quarkus.arc.properties.IfBuildProperty;
 import io.quarkus.logging.Log;
@@ -28,6 +25,7 @@ import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.util.NoSuchElementException;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 import static org.lmdbjava.DbiFlags.MDB_CREATE;
 import static org.lmdbjava.Env.create;
@@ -112,9 +110,9 @@ public class LmdbObjectPersistentStore implements ObjectPersistentStore {
                 private boolean _closed = false;
 
                 @Override
-                public CloseableKvIterator<JObjectKey, ByteBuffer> getIterator(IteratorStart start, JObjectKey key) {
+                public Stream<CloseableKvIterator<JObjectKey, MaybeTombstone<ByteBuffer>>> 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));
+                    return Stream.of(new KeyPredicateKvIterator<>(new LmdbKvIterator(_txn, start, key), start, key, (k) -> !k.value().equals(DB_VER_OBJ_NAME_STR)));
                 }
 
                 @Nonnull
@@ -195,7 +193,7 @@ public class LmdbObjectPersistentStore implements ObjectPersistentStore {
         return _root.toFile().getUsableSpace();
     }
 
-    private class LmdbKvIterator extends ReversibleKvIterator<JObjectKey, ByteBuffer> {
+    private class LmdbKvIterator extends ReversibleKvIterator<JObjectKey, MaybeTombstone<ByteBuffer>> {
         private static final Cleaner CLEANER = Cleaner.create();
         private final Txn<ByteBuffer> _txn; // Managed by the snapshot
         private final Cursor<ByteBuffer> _cursor;
@@ -350,13 +348,13 @@ public class LmdbObjectPersistentStore implements ObjectPersistentStore {
         }
 
         @Override
-        protected Pair<JObjectKey, ByteBuffer> nextImpl() {
+        protected Pair<JObjectKey, MaybeTombstone<ByteBuffer>> nextImpl() {
             if (!_hasNext) {
                 throw new NoSuchElementException("No more elements");
             }
             // TODO: Right now with java serialization it doesn't matter, it's all copied to arrays anyway
             var val = _cursor.val();
-            var ret = Pair.of(JObjectKey.fromByteBuffer(_cursor.key()), val.asReadOnlyBuffer());
+            Pair<JObjectKey, MaybeTombstone<ByteBuffer>> ret = Pair.of(JObjectKey.fromByteBuffer(_cursor.key()), new DataWrapper<>(val.asReadOnlyBuffer()));
             if (_goingForward)
                 _hasNext = _cursor.next();
             else

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

@@ -2,10 +2,7 @@ package com.usatiuk.objects.stores;
 
 import com.google.protobuf.ByteString;
 import com.usatiuk.objects.JObjectKey;
-import com.usatiuk.objects.iterators.CloseableKvIterator;
-import com.usatiuk.objects.iterators.IteratorStart;
-import com.usatiuk.objects.iterators.MappingKvIterator;
-import com.usatiuk.objects.iterators.NavigableMapKvIterator;
+import com.usatiuk.objects.iterators.*;
 import com.usatiuk.objects.snapshot.Snapshot;
 import io.quarkus.arc.properties.IfBuildProperty;
 import jakarta.enterprise.context.ApplicationScoped;
@@ -15,6 +12,7 @@ import javax.annotation.Nonnull;
 import java.nio.ByteBuffer;
 import java.util.Optional;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Stream;
 
 @ApplicationScoped
 @IfBuildProperty(name = "dhfs.objects.persistence", stringValue = "memory")
@@ -31,8 +29,8 @@ public class MemoryObjectPersistentStore implements ObjectPersistentStore {
                 private final long _lastCommitId = MemoryObjectPersistentStore.this._lastCommitId;
 
                 @Override
-                public CloseableKvIterator<JObjectKey, ByteBuffer> getIterator(IteratorStart start, JObjectKey key) {
-                    return new MappingKvIterator<>(new NavigableMapKvIterator<>(_objects, start, key), ByteString::asReadOnlyByteBuffer);
+                public Stream<CloseableKvIterator<JObjectKey, MaybeTombstone<ByteBuffer>>> getIterator(IteratorStart start, JObjectKey key) {
+                    return Stream.of(new MappingKvIterator<>(new NavigableMapKvIterator<>(_objects, start, key), s -> new DataWrapper<>(s.asReadOnlyByteBuffer())));
                 }
 
                 @Nonnull

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

@@ -1,13 +1,9 @@
 package com.usatiuk.objects.stores;
 
-import com.google.protobuf.ByteString;
 import com.usatiuk.objects.JDataVersionedWrapper;
 import com.usatiuk.objects.JDataVersionedWrapperSerializer;
 import com.usatiuk.objects.JObjectKey;
-import com.usatiuk.objects.ObjectSerializer;
-import com.usatiuk.objects.iterators.CloseableKvIterator;
-import com.usatiuk.objects.iterators.IteratorStart;
-import com.usatiuk.objects.iterators.MappingKvIterator;
+import com.usatiuk.objects.iterators.*;
 import com.usatiuk.objects.snapshot.Snapshot;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -16,6 +12,7 @@ import org.apache.commons.lang3.tuple.Pair;
 import javax.annotation.Nonnull;
 import java.nio.ByteBuffer;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 @ApplicationScoped
 public class SerializingObjectPersistentStore {
@@ -30,8 +27,9 @@ public class SerializingObjectPersistentStore {
             private final Snapshot<JObjectKey, ByteBuffer> _backing = delegateStore.getSnapshot();
 
             @Override
-            public CloseableKvIterator<JObjectKey, JDataVersionedWrapper> getIterator(IteratorStart start, JObjectKey key) {
-                return new MappingKvIterator<>(_backing.getIterator(start, key), d -> serializer.deserialize(d));
+            public Stream<CloseableKvIterator<JObjectKey, MaybeTombstone<JDataVersionedWrapper>>> getIterator(IteratorStart start, JObjectKey key) {
+                return _backing.getIterator(start, key).map(i -> new MappingKvIterator<JObjectKey, MaybeTombstone<ByteBuffer>, MaybeTombstone<JDataVersionedWrapper>>(i,
+                        d -> serializer.deserialize(((DataWrapper<ByteBuffer>) d).value())));
             }
 
             @Nonnull

dhfs-parent/objects/src/main/java/com/usatiuk/objects/stores/WritebackObjectPersistentStore.java

@@ -3,7 +3,10 @@ package com.usatiuk.objects.stores;
 import com.usatiuk.objects.JDataVersionedWrapper;
 import com.usatiuk.objects.JDataVersionedWrapperImpl;
 import com.usatiuk.objects.JObjectKey;
-import com.usatiuk.objects.iterators.*;
+import com.usatiuk.objects.iterators.CloseableKvIterator;
+import com.usatiuk.objects.iterators.IteratorStart;
+import com.usatiuk.objects.iterators.MaybeTombstone;
+import com.usatiuk.objects.iterators.NavigableMapKvIterator;
 import com.usatiuk.objects.snapshot.Snapshot;
 import com.usatiuk.objects.transaction.TxCommitException;
 import com.usatiuk.objects.transaction.TxRecord;
@@ -27,6 +30,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
+import java.util.stream.Stream;
 
 @ApplicationScoped
 public class WritebackObjectPersistentStore {
@@ -140,7 +144,7 @@ public class WritebackObjectPersistentStore {
 
                 Log.tracev("Bundle {0} committed", bundle.id());
 
-                while (true) {
+                synchronized (_pendingWrites) {
                     var curPw = _pendingWrites.get();
                     var curPwMap = curPw.pendingWrites();
                     for (var e : bundle._entries.values()) {
@@ -153,8 +157,7 @@ public class WritebackObjectPersistentStore {
                             bundle.id(),
                             curPw.lastCommittedId()
                     );
-                    if (_pendingWrites.compareAndSet(curPw, newCurPw))
-                        break;
+                    _pendingWrites.compareAndSet(curPw, newCurPw);
                 }
 
                 List<List<Runnable>> callbacks = new ArrayList<>();
@@ -251,7 +254,7 @@ public class WritebackObjectPersistentStore {
                     }
                 }
 
-                while (true) {
+                synchronized (_pendingWrites) {
                     var curPw = _pendingWrites.get();
                     var curPwMap = curPw.pendingWrites();
                     for (var e : ((TxBundle) bundle)._entries.values()) {
@@ -272,18 +275,17 @@ public class WritebackObjectPersistentStore {
                             bundle.id()
                     );
 
-                    if (!_pendingWrites.compareAndSet(curPw, newCurPw))
-                        continue;
-
-                    ((TxBundle) bundle).setReady();
-                    if (_pendingBundles.peek() == bundle)
-                        _pendingBundles.notify();
-                    synchronized (_flushWaitSynchronizer) {
-                        currentSize += ((TxBundle) bundle).size();
-                    }
-
-                    return bundle.id();
+                    _pendingWrites.compareAndSet(curPw, newCurPw);
                 }
+
+                ((TxBundle) bundle).setReady();
+                if (_pendingBundles.peek() == bundle)
+                    _pendingBundles.notify();
+                synchronized (_flushWaitSynchronizer) {
+                    currentSize += ((TxBundle) bundle).size();
+                }
+
+                return bundle.id();
             }
         }
     }
@@ -341,10 +343,8 @@ public class WritebackObjectPersistentStore {
                 private final long txId = finalPw.lastCommittedId();
 
                 @Override
-                public CloseableKvIterator<JObjectKey, JDataVersionedWrapper> getIterator(IteratorStart start, JObjectKey key) {
-                    return TombstoneMergingKvIterator.<JObjectKey, JDataVersionedWrapper>of("writeback-ps", start, key,
-                            (tS, tK) -> new NavigableMapKvIterator<>(_pendingWrites, tS, tK),
-                            (tS, tK) -> (CloseableKvIterator<JObjectKey, MaybeTombstone<JDataVersionedWrapper>>) (CloseableKvIterator<JObjectKey, ?>) _cache.getIterator(tS, tK));
+                public Stream<CloseableKvIterator<JObjectKey, MaybeTombstone<JDataVersionedWrapper>>> getIterator(IteratorStart start, JObjectKey key) {
+                    return Stream.concat(Stream.of(new NavigableMapKvIterator<>(_pendingWrites, start, key)), _cache.getIterator(start, key));
                 }
 
                 @Nonnull
@@ -390,7 +390,7 @@ public class WritebackObjectPersistentStore {
     }
 
     private static class TxBundle {
-        private final LinkedHashMap<JObjectKey, BundleEntry> _entries = new LinkedHashMap<>();
+        private final HashMap<JObjectKey, BundleEntry> _entries = new HashMap<>();
         private final ArrayList<Runnable> _callbacks = new ArrayList<>();
         private long _txId;
         private volatile boolean _ready = false;

dhfs-parent/objects/src/main/java/com/usatiuk/objects/transaction/CurrentTransaction.java

@@ -43,4 +43,9 @@ public class CurrentTransaction implements Transaction {
     public <T extends JData> void put(JData obj) {
         transactionManager.current().put(obj);
     }
+
+    @Override
+    public <T extends JData> void putNew(JData obj) {
+        transactionManager.current().putNew(obj);
+    }
 }

dhfs-parent/objects/src/main/java/com/usatiuk/objects/transaction/JObjectManager.java

@@ -11,6 +11,7 @@ import io.quarkus.runtime.StartupEvent;
 import jakarta.annotation.Priority;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.enterprise.event.Observes;
+import jakarta.enterprise.inject.Instance;
 import jakarta.enterprise.inject.spi.CDI;
 import jakarta.inject.Inject;
 import org.apache.commons.lang3.tuple.Pair;
@@ -35,8 +36,8 @@ public class JObjectManager {
         _preCommitTxHooks = List.copyOf(CDI.current().select(PreCommitTxHook.class).stream().sorted(Comparator.comparingInt(PreCommitTxHook::getPriority)).toList());
     }
 
-    JObjectManager() {
-        Log.infov("Pre-commit hooks: {0}", String.join("->", _preCommitTxHooks.stream().map(Objects::toString).toList()));
+    JObjectManager(Instance<PreCommitTxHook> preCommitTxHooks) {
+        Log.debugv("Pre-commit hooks: {0}", String.join("->", _preCommitTxHooks.stream().map(Objects::toString).toList()));
     }
 
     private void verifyReady() {
@@ -164,7 +165,8 @@ public class JObjectManager {
                     }
                 }
                 for (var write : writes.entrySet()) {
-                    toLock.add(write.getKey());
+                    if (!readSet.containsKey(write.getKey()))
+                        toLock.add(write.getKey());
                 }
                 Collections.sort(toLock);
                 for (var key : toLock) {

dhfs-parent/objects/src/main/java/com/usatiuk/objects/transaction/Transaction.java

@@ -14,6 +14,7 @@ public interface Transaction extends TransactionHandle {
     <T extends JData> Optional<T> get(Class<T> type, JObjectKey key, LockingStrategy strategy);
 
     <T extends JData> void put(JData obj);
+    <T extends JData> void putNew(JData obj);
 
     void delete(JObjectKey key);
 

dhfs-parent/objects/src/main/java/com/usatiuk/objects/transaction/TransactionFactoryImpl.java

@@ -13,6 +13,7 @@ import org.apache.commons.lang3.tuple.Pair;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 
 import java.util.*;
+import java.util.stream.Stream;
 
 @Singleton
 public class TransactionFactoryImpl implements TransactionFactory {
@@ -59,6 +60,7 @@ public class TransactionFactoryImpl implements TransactionFactory {
         private final NavigableMap<JObjectKey, TxRecord.TxObjectRecord<?>> _writes = new TreeMap<>();
         private final List<Runnable> _onCommit = new ArrayList<>();
         private final List<Runnable> _onFlush = new ArrayList<>();
+        private final HashSet<JObjectKey> _knownNew = new HashSet<>();
         private final Snapshot<JObjectKey, JDataVersionedWrapper> _snapshot;
         private boolean _closed = false;
         private Map<JObjectKey, TxRecord.TxObjectRecord<?>> _newWrites = new HashMap<>();
@@ -94,6 +96,9 @@ public class TransactionFactoryImpl implements TransactionFactory {
 
         @Override
         public <T extends JData> Optional<T> getFromSource(Class<T> type, JObjectKey key) {
+            if (_knownNew.contains(key)) {
+                return Optional.empty();
+            }
             return _readSet.computeIfAbsent(key, k -> {
                         var read = _snapshot.readObject(k);
                         return new TransactionObjectNoLock<>(read);
@@ -158,16 +163,21 @@ public class TransactionFactoryImpl implements TransactionFactory {
         @Override
         public CloseableKvIterator<JObjectKey, JData> getIterator(IteratorStart start, JObjectKey key) {
             Log.tracev("Getting tx iterator with start={0}, key={1}", start, key);
-            return new ReadTrackingIterator(TombstoneMergingKvIterator.<JObjectKey, ReadTrackingInternalCrap>of("tx", start, key,
-                    (tS, tK) -> new MappingKvIterator<>(new NavigableMapKvIterator<>(_writes, tS, tK),
+            return new ReadTrackingIterator(TombstoneMergingKvIterator.<JObjectKey, ReadTrackingInternalCrap>of("tx", start, key, (tS, tK) ->
+                    Stream.concat(Stream.of(new MappingKvIterator<>(new NavigableMapKvIterator<>(_writes, tS, tK),
                             t -> switch (t) {
                                 case TxRecord.TxObjectRecordWrite<?> write ->
-                                        new DataWrapper<>(new ReadTrackingInternalCrapTx(write.data()));
-                                case TxRecord.TxObjectRecordDeleted deleted -> new TombstoneImpl<>();
+                                        new DataWrapper<ReadTrackingInternalCrap>(new ReadTrackingInternalCrapTx(write.data()));
+                                case TxRecord.TxObjectRecordDeleted deleted ->
+                                        new TombstoneImpl<ReadTrackingInternalCrap>();
                                 case null, default -> null;
-                            }),
-                    (tS, tK) -> new MappingKvIterator<>(_snapshot.getIterator(tS, tK),
-                            d -> new DataWrapper<ReadTrackingInternalCrap>(new ReadTrackingInternalCrapSource(d)))));
+                            })), _snapshot.getIterator(tS, tK).map(itin -> new MappingKvIterator<JObjectKey, MaybeTombstone<JDataVersionedWrapper>, MaybeTombstone<ReadTrackingInternalCrap>>(itin,
+                            d -> switch (d) {
+                                case Data<JDataVersionedWrapper> w ->
+                                        new DataWrapper<>(new ReadTrackingInternalCrapSource(w.value()));
+                                case Tombstone<JDataVersionedWrapper> t -> new TombstoneImpl<>();
+                                case null, default -> null;
+                            })))));
         }
 
         @Override
@@ -181,6 +191,14 @@ public class TransactionFactoryImpl implements TransactionFactory {
             _newWrites.put(obj.key(), new TxRecord.TxObjectRecordWrite<>(obj));
         }
 
+        @Override
+        public void putNew(JData obj) {
+            _knownNew.add(obj.key());
+
+            _writes.put(obj.key(), new TxRecord.TxObjectRecordWrite<>(obj));
+            _newWrites.put(obj.key(), new TxRecord.TxObjectRecordWrite<>(obj));
+        }
+
         @Override
         public Collection<TxRecord.TxObjectRecord<?>> drainNewWrites() {
             var ret = _newWrites;

dhfs-parent/objects/src/test/java/com/usatiuk/objects/stores/LmdbKvIteratorTest.java

@@ -13,6 +13,7 @@ import org.apache.commons.lang3.tuple.Pair;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.RepeatedTest;
 
+import java.nio.ByteBuffer;
 import java.util.List;
 
 class Profiles {
@@ -46,49 +47,49 @@ public class LmdbKvIteratorTest {
 
         try (var snapshot = store.getSnapshot()) {
             var iterator = snapshot.getIterator(IteratorStart.GE, JObjectKey.of(""));
-            Just.checkIterator(iterator, List.of(Pair.of(JObjectKey.of(Long.toString(1)), ByteString.copyFrom(new byte[]{2})),
-                    Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})),
-                    Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4}))));
+            Just.checkIterator(iterator, List.of(Pair.of(JObjectKey.of(Long.toString(1)), ByteBuffer.wrap(new byte[]{2})),
+                    Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})),
+                    Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4}))));
             Assertions.assertFalse(iterator.hasNext());
             iterator.close();
 
             iterator = snapshot.getIterator(IteratorStart.LE, JObjectKey.of(Long.toString(3)));
-            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})));
+            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})));
             Assertions.assertFalse(iterator.hasNext());
             iterator.close();
 
             iterator = snapshot.getIterator(IteratorStart.LE, JObjectKey.of(Long.toString(2)));
-            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})));
+            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})));
             Assertions.assertFalse(iterator.hasNext());
             iterator.close();
 
             iterator = snapshot.getIterator(IteratorStart.GE, JObjectKey.of(Long.toString(2)));
-            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})));
+            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})));
             Assertions.assertFalse(iterator.hasNext());
             iterator.close();
 
             iterator = snapshot.getIterator(IteratorStart.GT, JObjectKey.of(Long.toString(2)));
-            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})));
+            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})));
             Assertions.assertFalse(iterator.hasNext());
             iterator.close();
 
             iterator = snapshot.getIterator(IteratorStart.LT, JObjectKey.of(Long.toString(3)));
-            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})));
+            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})));
             Assertions.assertFalse(iterator.hasNext());
             iterator.close();
 
             iterator = snapshot.getIterator(IteratorStart.LT, JObjectKey.of(Long.toString(2)));
-            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(1)), ByteString.copyFrom(new byte[]{2})), Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})));
+            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(1)), ByteBuffer.wrap(new byte[]{2})), Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})));
             Assertions.assertFalse(iterator.hasNext());
             iterator.close();
 
             iterator = snapshot.getIterator(IteratorStart.LT, JObjectKey.of(Long.toString(1)));
-            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(1)), ByteString.copyFrom(new byte[]{2})), Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})));
+            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(1)), ByteBuffer.wrap(new byte[]{2})), Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})));
             Assertions.assertFalse(iterator.hasNext());
             iterator.close();
 
             iterator = snapshot.getIterator(IteratorStart.LE, JObjectKey.of(Long.toString(1)));
-            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(1)), ByteString.copyFrom(new byte[]{2})), Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})));
+            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(1)), ByteBuffer.wrap(new byte[]{2})), Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})));
             Assertions.assertFalse(iterator.hasNext());
             iterator.close();
 
@@ -101,7 +102,7 @@ public class LmdbKvIteratorTest {
             iterator.close();
 
             iterator = snapshot.getIterator(IteratorStart.LE, JObjectKey.of(Long.toString(0)));
-            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(1)), ByteString.copyFrom(new byte[]{2})), Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})));
+            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(1)), ByteBuffer.wrap(new byte[]{2})), Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})));
             Assertions.assertFalse(iterator.hasNext());
             iterator.close();
 
@@ -111,11 +112,11 @@ public class LmdbKvIteratorTest {
             Assertions.assertEquals(JObjectKey.of(Long.toString(1)), iterator.peekPrevKey());
             Assertions.assertEquals(JObjectKey.of(Long.toString(2)), iterator.peekNextKey());
             Assertions.assertEquals(JObjectKey.of(Long.toString(1)), iterator.peekPrevKey());
-            Just.checkIterator(iterator.reversed(), Pair.of(JObjectKey.of(Long.toString(1)), ByteString.copyFrom(new byte[]{2})));
-            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(1)), ByteString.copyFrom(new byte[]{2})), Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})));
-            Assertions.assertEquals(Pair.of(JObjectKey.of(Long.toString(3)), ByteString.copyFrom(new byte[]{4})), iterator.prev());
-            Assertions.assertEquals(Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})), iterator.prev());
-            Assertions.assertEquals(Pair.of(JObjectKey.of(Long.toString(2)), ByteString.copyFrom(new byte[]{3})), iterator.next());
+            Just.checkIterator(iterator.reversed(), Pair.of(JObjectKey.of(Long.toString(1)), ByteBuffer.wrap(new byte[]{2})));
+            Just.checkIterator(iterator, Pair.of(JObjectKey.of(Long.toString(1)), ByteBuffer.wrap(new byte[]{2})), Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})), Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})));
+            Assertions.assertEquals(Pair.of(JObjectKey.of(Long.toString(3)), ByteBuffer.wrap(new byte[]{4})), iterator.prev());
+            Assertions.assertEquals(Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})), iterator.prev());
+            Assertions.assertEquals(Pair.of(JObjectKey.of(Long.toString(2)), ByteBuffer.wrap(new byte[]{3})), iterator.next());
             iterator.close();
         }
 

dhfs-parent/pom.xml

@@ -102,6 +102,7 @@
                         <arg>-parameters</arg>
                         <arg>--add-exports</arg>
                         <arg>java.base/jdk.internal.access=ALL-UNNAMED</arg>
+                        <arg>--enable-preview</arg>
                     </compilerArgs>
                 </configuration>
             </plugin>
@@ -119,6 +120,7 @@
                         --add-exports java.base/sun.nio.ch=ALL-UNNAMED
                         --add-exports java.base/jdk.internal.access=ALL-UNNAMED
                         --add-opens=java.base/java.nio=ALL-UNNAMED
+                        --enable-preview
                     </argLine>
                     <skipTests>${skip.unit}</skipTests>
                     <redirectTestOutputToFile>true</redirectTestOutputToFile>

dhfs-parent/sync-base/src/main/java/com/usatiuk/dhfs/remoteobj/RemoteObjPusherTxHook.java

@@ -19,8 +19,8 @@ public class RemoteObjPusherTxHook implements PreCommitTxHook {
     @Override
     public void onChange(JObjectKey key, JData old, JData cur) {
         boolean invalidate = switch (cur) {
-            case RemoteObjectMeta remote -> !remote.changelog().equals(((RemoteObjectMeta) old).changelog());
-            case JKleppmannTreePersistentData pd -> !pd.queues().equals(((JKleppmannTreePersistentData) old).queues());
+            case RemoteObjectMeta remote -> remote.changelog() != ((RemoteObjectMeta) old).changelog();
+            case JKleppmannTreePersistentData pd -> pd.queues() != ((JKleppmannTreePersistentData) old).queues();
             default -> false;
         };
 

dhfs-parent/sync-base/src/main/java/com/usatiuk/dhfs/remoteobj/RemoteTransaction.java

@@ -93,6 +93,11 @@ public class RemoteTransaction {
         curTx.put(newData);
     }
 
+    public <T extends JDataRemote> void putDataNew(T obj) {
+        curTx.putNew(new RemoteObjectMeta(obj, persistentPeerDataService.getSelfUuid()));
+        curTx.putNew(new RemoteObjectDataWrapper<>(obj));
+    }
+
     public <T extends JDataRemote> void putData(T obj) {
         var curMeta = getMeta(obj.key()).orElse(null);
 

dhfs-parent/utils/src/main/java/com/usatiuk/utils/DataLocker.java

@@ -4,93 +4,52 @@ import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
 
 import java.lang.ref.Cleaner;
+import java.lang.ref.WeakReference;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 public class DataLocker {
-    private static final AutoCloseableNoThrow DUMMY_LOCK = () -> {
-    };
-    private final ConcurrentHashMap<Object, LockTag> _locks = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<Object, WeakReference<ReentrantLock>> _locks = new ConcurrentHashMap<>();
+    private static final Cleaner CLEANER = Cleaner.create();
+
+    private Lock getTag(Object data) {
+        var newTag = new ReentrantLock();
+        var newTagRef = new WeakReference<>(newTag);
+
+        while (true) {
+            var oldTagRef = _locks.putIfAbsent(data, newTagRef);
+            var oldTag = oldTagRef != null ? oldTagRef.get() : null;
+
+            if (oldTag == null && oldTagRef != null) {
+                _locks.remove(data, oldTagRef);
+                continue;
+            }
+
+            if (oldTag != null)
+                return oldTag;
+
+            CLEANER.register(newTag, () -> {
+                _locks.remove(data, newTagRef);
+            });
+            return newTag;
+        }
+    }
 
     @Nonnull
     public AutoCloseableNoThrow lock(Object data) {
-        while (true) {
-            var newTag = new LockTag();
-            var oldTag = _locks.putIfAbsent(data, newTag);
-            if (oldTag == null) {
-                return new Lock(data, newTag);
-            }
-            try {
-                synchronized (oldTag) {
-                    while (!oldTag.released) {
-                        if (oldTag.owner == Thread.currentThread()) {
-                            return DUMMY_LOCK;
-                        }
-                        oldTag.wait();
-//                            tag.wait(4000L);
-//                            if (!tag.released) {
-//                                System.out.println("Timeout waiting for lock: " + data);
-//                                System.exit(1);
-//                                throw new InterruptedException();
-//                            }
-                    }
-                }
-            } catch (InterruptedException ignored) {
-            }
-        }
+        var lock = getTag(data);
+        lock.lock();
+        return lock::unlock;
     }
 
     @Nullable
     public AutoCloseableNoThrow tryLock(Object data) {
-        while (true) {
-            var newTag = new LockTag();
-            var oldTag = _locks.putIfAbsent(data, newTag);
-            if (oldTag == null) {
-                return new Lock(data, newTag);
-            }
-            synchronized (oldTag) {
-                if (!oldTag.released) {
-                    if (oldTag.owner == Thread.currentThread()) {
-                        return DUMMY_LOCK;
-                    }
-                    return null;
-                }
-            }
+        var lock = getTag(data);
+        if (lock.tryLock()) {
+            return lock::unlock;
+        } else {
+            return null;
         }
     }
-
-    private static class LockTag {
-        final Thread owner = Thread.currentThread();
-        //        final StackTraceElement[] _creationStack = Thread.currentThread().getStackTrace();
-        boolean released = false;
-    }
-
-    private class Lock implements AutoCloseableNoThrow {
-        private static final Cleaner CLEANER = Cleaner.create();
-        private final Object _key;
-        private final LockTag _tag;
-
-        public Lock(Object key, LockTag tag) {
-            _key = key;
-            _tag = tag;
-//            CLEANER.register(this, () -> {
-//                if (!tag.released) {
-//                    Log.error("Lock collected without release: " + key);
-//                }
-//            });
-        }
-
-        @Override
-        public void close() {
-            synchronized (_tag) {
-                if (_tag.released)
-                    return;
-                _tag.released = true;
-                // Notify all because when the object is locked again,
-                // it's a different lock tag
-                _tag.notifyAll();
-                _locks.remove(_key, _tag);
-            }
-        }
-    }
-
 }

dhfs-parent/utils/src/main/java/com/usatiuk/utils/UninitializedByteBuffer.java

@@ -0,0 +1,42 @@
+package com.usatiuk.utils;
+
+import java.lang.foreign.*;
+import java.lang.invoke.MethodHandle;
+import java.nio.ByteBuffer;
+import java.util.function.Consumer;
+
+public class UninitializedByteBuffer {
+
+    private static final Linker LINKER = Linker.nativeLinker();
+    private static final MethodHandle malloc = LINKER.downcallHandle(
+            LINKER.defaultLookup().find("malloc").orElseThrow(),
+            FunctionDescriptor.of(ValueLayout.ADDRESS, ValueLayout.JAVA_LONG)
+    );
+    private static final MethodHandle free = LINKER.downcallHandle(
+            LINKER.defaultLookup().find("free").orElseThrow(),
+            FunctionDescriptor.ofVoid(ValueLayout.ADDRESS)
+    );
+
+    public static ByteBuffer allocate(int capacity) {
+        UnsafeAccessor.get().getNioAccess().reserveMemory(capacity, capacity);
+        // Invoke malloc(), which returns a pointer
+        MemorySegment segment = null;
+        try {
+            segment = (MemorySegment) malloc.invokeExact((long) capacity);
+        } catch (Throwable e) {
+            throw new RuntimeException(e);
+        }
+
+
+        Consumer<MemorySegment> cleanup = s -> {
+            try {
+                free.invokeExact(s);
+                UnsafeAccessor.get().getNioAccess().unreserveMemory(capacity, capacity);
+            } catch (Throwable e) {
+                throw new RuntimeException(e);
+            }
+        };
+        var reint = segment.reinterpret(capacity, Arena.ofAuto(), cleanup);
+        return reint.asByteBuffer();
+    }
+}

dhfs-parent/utils/src/main/java/com/usatiuk/utils/UnsafeAccessor.java

@@ -0,0 +1,41 @@
+package com.usatiuk.utils;
+
+import jdk.internal.access.JavaNioAccess;
+import jdk.internal.access.SharedSecrets;
+import sun.misc.Unsafe;
+
+import java.lang.reflect.Field;
+
+public class UnsafeAccessor {
+    private static final UnsafeAccessor INSTANCE;
+
+    static {
+        try {
+            INSTANCE = new UnsafeAccessor();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public static UnsafeAccessor get() {
+        return INSTANCE;
+    }
+
+    private JavaNioAccess _nioAccess;
+    private Unsafe _unsafe;
+
+    private UnsafeAccessor() throws NoSuchFieldException, IllegalAccessException {
+        _nioAccess = SharedSecrets.getJavaNioAccess();
+        Field f = Unsafe.class.getDeclaredField("theUnsafe");
+        f.setAccessible(true);
+        _unsafe = (Unsafe) f.get(null);
+    }
+
+    public JavaNioAccess getNioAccess() {
+        return _nioAccess;
+    }
+
+    public Unsafe getUnsafe() {
+        return _unsafe;
+    }
+}