Read pages through a memory map of their segment

docs/operations/docker.md

@@ -216,7 +216,8 @@ The [`Dockerfile`](../../Dockerfile) has two stages:
 
 1. **build**: `ghcr.io/graalvm/native-image-community:25`. Runs `./mvnw install -DskipTests` and then
    `./mvnw -Pnative package -pl server`, producing a native `hstore` executable linked against glibc (Serial GC, `-O2`,
-   `-march=compatibility` so the image runs on any CPU of the target architecture). A BuildKit cache mount
+   `-march=compatibility` so the image runs on any CPU of the target architecture, and shared arenas so segment maps
+   can be closed when a segment is deleted). A BuildKit cache mount
    keeps `~/.m2` between builds.
 2. **runtime**: `debian:bookworm-slim` plus the binary, the entrypoint and the non-root user.
 

docs/storage/pages.md

@@ -193,14 +193,41 @@ Compared with fixed page slots, packing removes almost all internal fragmentatio
 
 ### Reading (`SegmentFile.read(position, maximum)`)
 
-The length of an image is not known before its header is read, so a read takes at most two system calls:
-
-1. Allocate one buffer of `min(pageSize, 4096)` bytes and read it at `offset × 64`. This is the only allocation for most images.
-2. If at least 80 bytes arrived, compute `length = 80 + payloadLength` from header offset 24:
-   * if `length` is already covered by the first read, or exceeds `pageSize` (a corrupt header), return the first buffer; `verify` then decides;
-   * otherwise, and only then, allocate a second buffer of exactly `length` bytes, copy the first read into it, and read the remainder.
-
-Most nodes fit in 4 KiB, so most reads are a single 4 KiB `pread`. A short read at end of file returns whatever arrived, and the header check then rejects it.
+Each segment file has a read-only memory map. If the map covers the image, `read` copies exactly
+`80 + payloadLength` bytes out of it into a new heap array and returns that. There is no system call, and the raw
+bytes stay in the OS page cache rather than on the Java heap, which only holds the decoded nodes in `NodeCache`.
+
+The copy is deliberate. `ByteCursor` reads one byte at a time through `MemorySegment`, and the JIT only makes that
+fast when every segment it sees is the same kind. Decoding straight from the map, while the WAL, catalog and
+feed decode from heap arrays, made decoding two to five times slower everywhere in a microbenchmark. Copying a
+page costs well under a microsecond and keeps every decoder on heap memory.
+
+If the image ends past the map, the file may need mapping again:
+
+* A sealed segment is remapped at its current size right away. The store seals a segment when it rolls to the
+  next one, and seals every segment except the active one when it opens.
+* The active segment is remapped only after the file has grown 4 MiB past the map. Until then, reads near the end
+  of the file use `pread`. This keeps a segment that is still filling up from being remapped on every new page.
+
+Each map has its own `Arena.ofShared()`. When a segment is remapped, the old map is closed as soon as the new one is
+in place, and deleting or closing a segment closes its map. A deleted segment's disk space therefore comes back
+straight away instead of whenever the garbage collector gets to the map. A read that was copying from a map when it
+was closed gets an `IllegalStateException`, sees that the map is no longer alive, and reads again from the current
+one. Readers only touch a map for the length of one copy, and compaction only deletes segments that no reader can
+reach, so this retry is rare. Closing a shared arena in a Native Image build needs `-H:+SharedArenaSupport`, which
+GraalVM still lists as experimental; the `native` profile in `server/pom.xml` turns it on. Segment files are only
+ever appended to or deleted, never truncated, so a map can't end up pointing past the end of its file.
+
+`pread` doesn't know how long an image is until it has the header, so it takes one or two system calls:
+
+1. Allocate a buffer of `min(pageSize, 4096)` bytes and read it at `offset × 64`.
+2. If at least 80 bytes came back, compute `length = 80 + payloadLength` from header offset 24:
+   * if `length` is larger than `pageSize`, the header is corrupt; return the buffer and let `verify` fail it;
+   * if the first read already has `length` bytes, return those;
+   * otherwise allocate a buffer of exactly `length` bytes, copy in what was read, and read the rest.
+
+A short read at the end of the file returns what it got, and the header check rejects it. Both paths return
+read-only memory, so nothing can write into a stored image through a read.
 
 ## Caching and I/O accounting
 

engine/src/main/java/io/hstore/engine/page/PageHeader.java

@@ -3,6 +3,8 @@
 import io.hstore.engine.HStoreException;
 
 import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.nio.ByteOrder;
 
 public record PageHeader(
         PageType type,
@@ -24,6 +26,7 @@ public record PageHeader(
     private static final int PAGE_ID_OFFSET = 8;
     private static final int PAYLOAD_LENGTH_OFFSET = 24;
     private static final int CHECKSUM_OFFSET = 72;
+    private static final ValueLayout.OfInt INT = ValueLayout.JAVA_INT_UNALIGNED.withOrder(ByteOrder.LITTLE_ENDIAN);
 
     public void writeTo(ByteCursor page) {
         page.position(0)
@@ -55,6 +58,10 @@ public static int payloadLength(MemorySegment page) {
         return ByteCursor.over(page).position(PAYLOAD_LENGTH_OFFSET).getInt();
     }
 
+    static int payloadLengthAt(MemorySegment file, long position) {
+        return file.get(INT, position + PAYLOAD_LENGTH_OFFSET);
+    }
+
     public static void seal(MemorySegment page, int payloadLength) {
         int checksum = Checksums.crc32c(page.asSlice(0, CHECKSUM_OFFSET), page.asSlice(SIZE, payloadLength));
         ByteCursor.over(page).position(CHECKSUM_OFFSET).putInt(checksum);

engine/src/main/java/io/hstore/engine/page/PageStore.java

@@ -67,6 +67,11 @@ private void restore(Collection<SegmentInfo> known) {
                     activeSegment = info.id();
                     nextUnit = Math.toIntExact(info.units());
                 });
+        files.forEach((id, file) -> {
+            if (id != activeSegment) {
+                file.seal();
+            }
+        });
     }
 
     public int pageSize() {
@@ -95,6 +100,10 @@ public long allocate(long epoch, int length) {
     private void rollSegment() {
         if (activeSegment != 0) {
             segments.computeIfPresent(activeSegment, (_, info) -> info.withState(SegmentState.SEALED, 0));
+            SegmentFile previous = files.get(activeSegment);
+            if (previous != null) {
+                previous.seal();
+            }
         }
         int id = segments.isEmpty() ? 1 : segments.lastKey() + 1;
         files.put(id, SegmentFile.open(id, SegmentFile.pathFor(directory, id)));
@@ -120,6 +129,11 @@ public void write(long pageId, MemorySegment image) {
         bytesWritten.addAndGet(image.byteSize());
     }
 
+    long mappedBytes(int segment) {
+        SegmentFile file = files.get(segment);
+        return file == null ? 0 : file.mappedBytes();
+    }
+
     public MemorySegment read(long pageId) {
         SegmentFile file = files.get(PageId.segmentOf(pageId));
         if (file == null) {

engine/src/main/java/io/hstore/engine/page/SegmentFile.java

@@ -4,22 +4,30 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantLock;
 
 final class SegmentFile implements AutoCloseable {
 
     private static final int FIRST_READ = 4096;
+    private static final long REMAP_BYTES = 4L << 20;
 
     private final int id;
     private final Path path;
     private final FileChannel channel;
     private final AtomicBoolean dirty = new AtomicBoolean();
+    private final ReentrantLock remapping = new ReentrantLock();
+    private volatile MemorySegment mapped = MemorySegment.NULL;
+    private Arena arena;
+    private volatile boolean sealed;
 
     private SegmentFile(int id, Path path, FileChannel channel) {
         this.id = id;
@@ -57,19 +65,102 @@ long unitsOnDisk() {
         }
     }
 
+    void seal() {
+        sealed = true;
+    }
+
     MemorySegment read(long position, int maximum) {
+        while (true) {
+            MemorySegment view = covering(position + PageHeader.SIZE);
+            if (view == null) {
+                return copy(position, maximum);
+            }
+            try {
+                return read(view, position, maximum);
+            } catch (IllegalStateException closed) {
+                if (view.scope().isAlive()) {
+                    throw closed;
+                }
+            }
+        }
+    }
+
+    private MemorySegment read(MemorySegment view, long position, int maximum) {
+        long length = PageHeader.SIZE + Integer.toUnsignedLong(PageHeader.payloadLengthAt(view, position));
+        if (length > maximum) {
+            return onHeap(view, position, Math.min(maximum, view.byteSize() - position));
+        }
+        if (position + length > view.byteSize()) {
+            view = covering(position + length);
+            if (view == null) {
+                return copy(position, maximum);
+            }
+        }
+        return onHeap(view, position, length);
+    }
+
+    long mappedBytes() {
+        return mapped.byteSize();
+    }
+
+    private static MemorySegment onHeap(MemorySegment view, long position, long length) {
+        byte[] image = new byte[Math.toIntExact(length)];
+        MemorySegment.copy(view, ValueLayout.JAVA_BYTE, position, image, 0, image.length);
+        return MemorySegment.ofArray(image).asReadOnly();
+    }
+
+    private MemorySegment covering(long required) {
+        MemorySegment current = mapped;
+        if (required <= current.byteSize()) {
+            return current;
+        }
+        if (!sealed && required - current.byteSize() < REMAP_BYTES) {
+            return null;
+        }
+        remapping.lock();
+        try {
+            current = mapped;
+            if (required <= current.byteSize()) {
+                return current;
+            }
+            long size = channel.size();
+            if (size < required) {
+                return null;
+            }
+            Arena replacement = Arena.ofShared();
+            try {
+                current = channel.map(FileChannel.MapMode.READ_ONLY, 0, size, replacement);
+            } catch (IOException | RuntimeException e) {
+                replacement.close();
+                throw e;
+            }
+            mapped = current;
+            unmap();
+            arena = replacement;
+            return current;
+        } catch (IOException e) {
+            throw HStoreException.io("cannot map segment " + id, e);
+        } finally {
+            remapping.unlock();
+        }
+    }
+
+    private MemorySegment copy(long position, int maximum) {
         ByteBuffer first = ByteBuffer.allocate(Math.min(maximum, FIRST_READ));
         fill(first, position);
         if (first.position() < PageHeader.SIZE) {
-            return MemorySegment.ofBuffer(first.flip());
+            return MemorySegment.ofBuffer(first.flip()).asReadOnly();
         }
         int length = PageHeader.SIZE + PageHeader.payloadLength(MemorySegment.ofArray(first.array()));
-        if (length <= first.position() || length > maximum) {
-            return MemorySegment.ofBuffer(first.flip());
+        if (length < PageHeader.SIZE || length > maximum) {
+            return MemorySegment.ofBuffer(first.flip()).asReadOnly();
+        }
+        if (length <= first.position()) {
+            return MemorySegment.ofBuffer(first.flip()).asSlice(0, length).asReadOnly();
         }
         ByteBuffer whole = ByteBuffer.allocate(length).put(first.flip());
         fill(whole, position);
-        return MemorySegment.ofBuffer(whole.flip());
+        return MemorySegment.ofBuffer(whole.flip()).asReadOnly();
     }
 
     private void fill(ByteBuffer buffer, long position) {
@@ -116,8 +207,22 @@ void delete() {
         }
     }
 
+    private void unmap() {
+        if (arena != null) {
+            arena.close();
+            arena = null;
+        }
+    }
+
     @Override
     public void close() {
+        remapping.lock();
+        try {
+            mapped = MemorySegment.NULL;
+            unmap();
+        } finally {
+            remapping.unlock();
+        }
         try {
             channel.close();
         } catch (IOException e) {