perf: stop allocating to pin a generation on every read

database/pom.xml

@@ -31,9 +31,9 @@
                         <configuration>
                             <compilerArgs combine.children="append">
                                 <arg>--add-modules</arg>
-                                <arg>jdk.httpserver</arg>
+                                <arg>jdk.httpserver,jdk.management</arg>
                                 <arg>--add-reads</arg>
-                                <arg>io.hstore.db=jdk.httpserver</arg>
+                                <arg>io.hstore.db=jdk.httpserver,java.management,jdk.management</arg>
                             </compilerArgs>
                         </configuration>
                     </execution>
@@ -43,7 +43,7 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-surefire-plugin</artifactId>
                 <configuration>
-                    <argLine>--add-modules jdk.httpserver --add-reads io.hstore.db=jdk.httpserver</argLine>
+                    <argLine>--add-modules jdk.httpserver,jdk.management --add-reads io.hstore.db=jdk.httpserver,java.management,jdk.management</argLine>
                 </configuration>
             </plugin>
         </plugins>

database/src/test/java/io/hstore/db/ReadAllocationTest.java

@@ -0,0 +1,49 @@
+// SPDX-FileCopyrightText: Metacog Labs
+// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0
+
+package io.hstore.db;
+
+import io.hstore.engine.EngineOptions;
+import io.hstore.engine.StorageEngine;
+
+import com.sun.management.ThreadMXBean;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.lang.management.ManagementFactory;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ReadAllocationTest {
+
+    private static final ThreadMXBean THREADS = (ThreadMXBean) ManagementFactory.getThreadMXBean();
+
+    @TempDir
+    Path directory;
+
+    @Test
+    void openingAndClosingASnapshotAllocatesOnlyTheSnapshot() {
+        try (HypergraphDatabase database = HypergraphDatabase.open(directory, DatabaseOptions.defaults()
+                .withEngine(EngineOptions.defaults().withPageSize(4096)))) {
+            database.write(writer -> writer.defineNode("Person", List.of()));
+            for (int i = 0; i < 200; i++) {
+                int index = i;
+                database.write(writer -> writer.node("Person", "p" + index, Map.of()));
+            }
+            StorageEngine engine = database.engine();
+            for (int i = 0; i < 100_000; i++) {
+                engine.snapshot().close();
+            }
+            int rounds = 10_000;
+            long before = THREADS.getCurrentThreadAllocatedBytes();
+            for (int i = 0; i < rounds; i++) {
+                engine.snapshot().close();
+            }
+            long perSnapshot = (THREADS.getCurrentThreadAllocatedBytes() - before) / rounds;
+            assertTrue(perSnapshot <= 48, perSnapshot + " bytes allocated per snapshot");
+        }
+    }
+}

docs/architecture/overview.md

@@ -88,7 +88,7 @@ The engine has no dependencies outside `java.base`. All three modules build into
 | `engine.topology` | Hyperedge representation: `Hyperedge` (members tree plus order index), `Incidence` (one member's role set, weight, validity, payload and qualifier), `IncidenceCodec` (columnar leaf encoding), `MemberChange`, `Weight` (fixed point) and `Validity`. |
 | `engine.index` | `PostingIndex`, a two-level index from a key to a postings set. Small sets are inlined and large ones are promoted to their own tree. Used by the reverse incidence, type, canonical-key and symbol indexes. |
 | `engine.catalog` | `AtomRecord` (`NodeRecord`/`EdgeRecord`) and its codec, `EngineSlots` (built-in slots, schemas and key functions), `Slot`/`SlotRegistry`, `RootVector`, `Generation`, `Branch`, `Symbol`, and `CatalogImage`/`CatalogStore` (checkpoint files). |
-| `engine.txn` | `TransactionManager` (commit lock, append, publication, pins, history, write buffer), `ChangeLog` (the key changes a commit logs, and their replay), `Transaction` (op log, read sets, rebase), `Snapshot`, `View` (read API shared by both), `Workspace` (mutable root vector plus derivations), `GroupCommitter`, `Op`/`EdgeAction`, `Derivation`/`EngineDerivations`, `CrashPoint`, `Durability`, `WalMode`, `Isolation` and `TxnOptions`. |
+| `engine.txn` | `TransactionManager` (commit lock, append, publication, history and its pin counts, write buffer), `ChangeLog` (the key changes a commit logs, and their replay), `Transaction` (op log, read sets, rebase), `Snapshot`, `View` (read API shared by both), `Workspace` (mutable root vector plus derivations), `GroupCommitter`, `Op`/`EdgeAction`, `Derivation`/`EngineDerivations`, `CrashPoint`, `Durability`, `WalMode`, `Isolation` and `TxnOptions`. |
 | `engine.wal` | `WriteAheadLog` (segmented, checksummed records) and `WalRecord` (`Begin`, `Page`, `PageRef`, `Changes`, `BranchMeta`, `Feed`, `Commit`, `Abort`, `Checkpoint`, and `Root`, which is still read but no longer written). |
 | `engine.feed` | `ChangeFeed` (segmented append-only `feed/*.feed` files, sparse generation index, retention, tailing subscriptions), `CommitEvent` and `FeedCodec`; see [change-feed.md](../storage/change-feed.md). |
 | `engine.maintenance` | `Checkpointer` (writes a published generation's trees to pages and publishes the catalog image), `ReferenceCounter` (counts references at each checkpoint, releases pages no reader can reach), `Compactor` (moving live pages out of victim segments, reclamation), `Recovery` (catalog discovery, replay of logged key changes, durable-prefix rule) and `ChangeLogCheck` (replays the log against the generations in memory; the tests use it). |
@@ -114,7 +114,7 @@ The engine has no dependencies outside `java.base`. All three modules build into
 
 ## Life of a read
 
-`HypergraphDatabase.read` opens a `Snapshot` (`engine/src/main/java/io/hstore/engine/txn/Snapshot.java`). Opening it pins the current generation (`TransactionManager.pinCurrent` increments a counter in `pins`) and wraps the branch's `RootVector`. The snapshot never changes, and no lock is held while it is read. `Snapshot.close` unpins.
+`HypergraphDatabase.read` opens a `Snapshot` (`engine/src/main/java/io/hstore/engine/txn/Snapshot.java`). Opening it pins the current generation (`TransactionManager.pinCurrent` increments the reader count of the current generation's `Retained` entry in `history`) and wraps the branch's `RootVector`. The snapshot never changes, and no lock is held while it is read. `Snapshot.close` unpins.
 
 ```mermaid
 sequenceDiagram
@@ -129,7 +129,7 @@ sequenceDiagram
     participant PS as PageStore
     C->>HD: read(branch, principal, work)
     HD->>TM: snapshot(branch)
-    TM->>TM: pins[current.id]++
+    TM->>TM: current.readers++
     TM-->>HD: Snapshot(generation, branch.roots)
     HD->>C: work.apply(Reader)
     C->>S: atom(id) / incident(atom) / edge(id)
@@ -149,7 +149,7 @@ sequenceDiagram
     end
     T-->>S: Optional value
     C-->>HD: result
-    HD->>TM: Snapshot.close(): pins[id]--
+    HD->>TM: Snapshot.close(): readers--
 ```
 
 Notes:

docs/storage/catalog-and-generations.md

@@ -92,11 +92,11 @@ sequenceDiagram
 
 ### Pins
 
-Every `Snapshot` and `Transaction` pins its generation id in `TransactionManager.pins`, a `ConcurrentSkipListMap<Long, Integer>` reference count. `oldestPinned()` is the smallest pinned id. The compactor consults it before deleting retired segments (see [maintenance.md](maintenance.md)), and `trimHistory` never drops a pinned generation.
+Every `Snapshot` and `Transaction` pins its generation by incrementing the reader count of the generation's `Retained` entry in `TransactionManager.history`, and decrements it when it closes. `oldestPinned()` is the smallest id whose count is above zero. The compactor consults it before deleting retired segments (see [maintenance.md](maintenance.md)), and `trimHistory` never drops a pinned generation.
 
 ## History retention and time travel
 
-Published generations are kept in `TransactionManager.history` (a `ConcurrentSkipListMap` by id).
+Published generations are kept in `TransactionManager.history` (a `ConcurrentSkipListMap` by id, each value a `Retained` holding the generation and its reader count).
 
 * `trimHistory` runs on every publication. While there are more than `historyLimit` entries (`history_limit`, default 64), it drops the oldest generations that are neither pinned nor current, skipping over pinned ones rather than stopping at them. A long-running snapshot therefore keeps only its own generation alive beyond the limit; older unpinned generations are still trimmed.
 * `snapshotAt(generation, branch)` pins the generation and opens a snapshot over its root vector. A generation that has been trimmed fails with `generation N is no longer retained`.

docs/storage/maintenance.md

@@ -247,7 +247,7 @@ victim's scan finds and moves them like any other page, and the transaction comm
 
 Retained history no longer holds a segment back. Old generations reach moved pages through the directory like everything else, so `AT GENERATION` and `AS OF` keep working after the victims are gone.
 
-**Why it must wait for readers.** A reader looks up a page's address and then reads the page. One that started before the move may have looked up an old address and not read it yet. Every reader and writer pins its generation, and anyone pinned at `retiredAt` or earlier may have started before the moves finished. Deleting the file early would turn that reader's next read into `segment does not exist`. Pins are reference counts per generation (`TransactionManager.pins`). Every `Snapshot.close()` and transaction end decrements them, and the maintenance thread retries `reclaim` on every pass (`maintenanceInterval`, 500 ms by default). A retired segment therefore disappears within one interval of its last reader closing.
+**Why it must wait for readers.** A reader looks up a page's address and then reads the page. One that started before the move may have looked up an old address and not read it yet. Every reader and writer pins its generation, and anyone pinned at `retiredAt` or earlier may have started before the moves finished. Deleting the file early would turn that reader's next read into `segment does not exist`. Pins are reference counts per generation (the reader count of each `Retained` entry in `TransactionManager.history`). Every `Snapshot.close()` and transaction end decrements them, and the maintenance thread retries `reclaim` on every pass (`maintenanceInterval`, 500 ms by default). A retired segment therefore disappears within one interval of its last reader closing.
 
 ### History retention and pins
 

docs/transactions/transactions.md

@@ -88,23 +88,24 @@ If a transaction that spilled is aborted (explicitly, or by `close()` without co
 ```mermaid
 flowchart LR
     subgraph TM[TransactionManager]
-        H[("history: generation id to Generation")]
-        P[("pins: generation id to pin count")]
+        H[("history: generation id to Retained (generation, reader count)")]
         C[current]
     end
-    B[begin / snapshot] -->|pin current.id| P
-    S[snapshotAt g] -->|pin g, then look up history| H
-    X[commit / Snapshot.close / abort] -->|unpin| P
+    B[begin / snapshot] -->|pin current| C
+    S[snapshotAt g] -->|look up history, then pin| H
+    X[commit / Snapshot.close / abort] -->|unpin its Retained| H
     PUB[publication of g+1] -->|history.put, trimHistory| H
 ```
 
-* `snapshot(branch)` and `begin(options)` call `pinCurrent()`: read `current`, increment `pins[id]`, then check that `history` still holds that generation. If a trim removed it in between, it unpins and tries again with the new `current`. Publication puts a generation into `history` before it makes it `current`, so this only loops when a reader races a trim.
-* `snapshotAt(g, branch)` pins `g` first and then looks it up in `history`, so a concurrent trim cannot remove it between lookup and pin. If `g` is gone it unpins and throws `INVALID_SCHEMA` (`"generation g is no longer retained"`).
-* `trimHistory()` (run during publication) computes `excess = history.size() - historyLimit` and walks the history oldest first, removing every generation that is neither pinned nor `current` until `excess` reaches 0. After removing one it checks the pins again and puts the generation back if a reader pinned it in the meantime. Pinned generations are **skipped**, not barriers: a long-running snapshot on an old generation keeps exactly that generation, while newer unpinned ones are still trimmed, so the history size stays bounded by `historyLimit + number of distinct pinned generations`.
-* `oldestPinned()` exposes the oldest pinned generation to maintenance so that compaction never deletes a segment a pinned snapshot may still read. `oldestReadable()`, the smaller of that and the oldest retained generation, tells a checkpoint which dead pages it may release ([maintenance](../storage/maintenance.md#releasing-pages)).
+Each entry of `history` is a `Retained`: the `Generation`, an `AtomicInteger` count of the readers and writers pinned on it, and a `dropped` flag. `current` points at the `Retained` of the current generation, and a `Snapshot` or `Transaction` keeps the `Retained` it pinned, so pinning and unpinning touch only that counter and allocate nothing.
+
+* `snapshot(branch)` and `begin(options)` call `pinCurrent()`: read `current`, increment its count, then check its `dropped` flag. If a trim dropped it in between, it decrements and tries again with the new `current`. Publication puts a generation into `history` before it makes it `current`, so this only loops when a reader races a trim.
+* `snapshotAt(g, branch)` looks `g` up in `history` and pins it the same way. If `g` is gone or was dropped, it throws `INVALID_SCHEMA` (`"generation g is no longer retained"`).
+* `trimHistory()` (run during publication) computes `excess = history.size() - historyLimit` and walks the history oldest first, removing every generation that is neither pinned nor `current` until `excess` reaches 0. To remove one it sets `dropped` and then reads the count again; if a reader pinned it in the meantime it clears `dropped` and skips it. A reader increments before it reads `dropped` and the trim sets `dropped` before it reads the count, so at least one of them sees the other. Pinned generations are **skipped**, not barriers: a long-running snapshot on an old generation keeps exactly that generation, while newer unpinned ones are still trimmed, so the history size stays bounded by `historyLimit + number of distinct pinned generations`.
+* `oldestPinned()` walks the history oldest first and returns the first generation with readers, so that compaction never deletes a segment a pinned snapshot may still read. `oldestReadable()`, the smaller of that and the oldest retained generation, tells a checkpoint which dead pages it may release ([maintenance](../storage/maintenance.md#releasing-pages)).
 * `generationAsOf(wallTime)` returns the newest retained generation with `wallTime <= t` (used by `AS OF` time travel).
 
-A `Snapshot` is a `View` whose trees are rooted at the pinned branch's root vector; `Snapshot.close()` unpins once (`AtomicBoolean` guard). Retention defaults: `EngineOptions.historyLimit = 64`. History is kept in memory only; after a restart it starts again from the generations recovery rebuilt ([catalog and generations](../storage/catalog-and-generations.md#history-retention-and-time-travel)).
+A `Snapshot` is a `View` whose trees are rooted at the pinned branch's root vector; `Snapshot.close()` unpins once (a `closed` flag set by compare-and-set). Retention defaults: `EngineOptions.historyLimit = 64`. History is kept in memory only; after a restart it starts again from the generations recovery rebuilt ([catalog and generations](../storage/catalog-and-generations.md#history-retention-and-time-travel)).
 
 ## 5. Commit
 

engine/src/main/java/io/hstore/engine/tree/Tree.java

@@ -142,12 +142,26 @@ private void summarize(Ref ref, long low, long high, Summary.Builder into) {
     }
 
     public Optional<Entry<V>> first() {
-        TreeCursor<V> cursor = cursor();
-        return cursor.hasNext() ? Optional.of(cursor.next()) : Optional.empty();
+        return edge(false);
     }
 
     public Optional<Entry<V>> last() {
-        TreeCursor<V> cursor = new TreeCursor<>(this, _ -> true, true);
+        return edge(true);
+    }
+
+    private Optional<Entry<V>> edge(boolean last) {
+        if (isEmpty()) {
+            return Optional.empty();
+        }
+        Node node = node(root);
+        while (node instanceof Branch branch && branch.size() > 0) {
+            node = node(branch.child(last ? branch.size() - 1 : 0));
+        }
+        if (node instanceof Leaf leaf && leaf.size() > 0) {
+            int position = last ? leaf.size() - 1 : 0;
+            return Optional.of(new Entry<>(leaf.key(position), schema.cast(leaf.value(position))));
+        }
+        TreeCursor<V> cursor = new TreeCursor<>(this, _ -> true, last);
         return cursor.hasNext() ? Optional.of(cursor.next()) : Optional.empty();
     }
 

engine/src/main/java/io/hstore/engine/txn/Retained.java

@@ -0,0 +1,56 @@
+// SPDX-FileCopyrightText: Metacog Labs
+// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0
+
+package io.hstore.engine.txn;
+
+import io.hstore.engine.catalog.Generation;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+final class Retained {
+
+    private final AtomicInteger readers = new AtomicInteger();
+    private volatile Generation generation;
+    private volatile boolean dropped;
+
+    Retained(Generation generation) {
+        this.generation = generation;
+    }
+
+    Generation generation() {
+        return generation;
+    }
+
+    void rebind(Generation rebound) {
+        generation = rebound;
+    }
+
+    boolean pin() {
+        readers.incrementAndGet();
+        if (!dropped) {
+            return true;
+        }
+        readers.decrementAndGet();
+        return false;
+    }
+
+    void unpin() {
+        readers.decrementAndGet();
+    }
+
+    boolean pinned() {
+        return readers.get() > 0;
+    }
+
+    boolean drop() {
+        if (pinned()) {
+            return false;
+        }
+        dropped = true;
+        if (pinned()) {
+            dropped = false;
+            return false;
+        }
+        return true;
+    }
+}

engine/src/main/java/io/hstore/engine/txn/Snapshot.java

@@ -10,17 +10,30 @@
 import io.hstore.engine.tree.NodeSource;
 import io.hstore.engine.tree.Tree;
 
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
 
 public final class Snapshot extends View implements AutoCloseable {
 
+    private static final VarHandle CLOSED;
+
+    static {
+        try {
+            CLOSED = MethodHandles.lookup().findVarHandle(Snapshot.class, "closed", boolean.class);
+        } catch (ReflectiveOperationException e) {
+            throw new ExceptionInInitializerError(e);
+        }
+    }
+
     private final TransactionManager manager;
+    private final Retained retained;
     private final Generation generation;
     private final Branch branch;
-    private final AtomicBoolean open = new AtomicBoolean(true);
+    private volatile boolean closed;
 
-    Snapshot(TransactionManager manager, Generation generation, Branch branch) {
+    Snapshot(TransactionManager manager, Retained retained, Generation generation, Branch branch) {
         this.manager = manager;
+        this.retained = retained;
         this.generation = generation;
         this.branch = branch;
     }
@@ -59,8 +72,8 @@ public long wallTime() {
 
     @Override
     public void close() {
-        if (open.compareAndSet(true, false)) {
-            manager.unpin(generation.id());
+        if (CLOSED.compareAndSet(this, false, true)) {
+            retained.unpin();
         }
     }
 }