diff --git a/database/pom.xml b/database/pom.xml index 2d6dee46..40cb67a1 100644 --- a/database/pom.xml +++ b/database/pom.xml @@ -31,9 +31,9 @@ --add-modules - jdk.httpserver + jdk.httpserver,jdk.management --add-reads - io.hstore.db=jdk.httpserver + io.hstore.db=jdk.httpserver,java.management,jdk.management @@ -43,7 +43,7 @@ org.apache.maven.plugins maven-surefire-plugin - --add-modules jdk.httpserver --add-reads io.hstore.db=jdk.httpserver + --add-modules jdk.httpserver,jdk.management --add-reads io.hstore.db=jdk.httpserver,java.management,jdk.management diff --git a/database/src/test/java/io/hstore/db/ReadAllocationTest.java b/database/src/test/java/io/hstore/db/ReadAllocationTest.java new file mode 100644 index 00000000..f26fc940 --- /dev/null +++ b/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"); + } + } +} diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index 6df4852e..6c874b02 100644 --- a/docs/architecture/overview.md +++ b/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: diff --git a/docs/storage/catalog-and-generations.md b/docs/storage/catalog-and-generations.md index ea8d3a74..15f3a927 100644 --- a/docs/storage/catalog-and-generations.md +++ b/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` 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`. diff --git a/docs/storage/maintenance.md b/docs/storage/maintenance.md index f59e7338..a4a29f67 100644 --- a/docs/storage/maintenance.md +++ b/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 diff --git a/docs/transactions/transactions.md b/docs/transactions/transactions.md index 14023b5e..0957f230 100644 --- a/docs/transactions/transactions.md +++ b/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 diff --git a/engine/src/main/java/io/hstore/engine/tree/Tree.java b/engine/src/main/java/io/hstore/engine/tree/Tree.java index fb3492da..d28abffc 100644 --- a/engine/src/main/java/io/hstore/engine/tree/Tree.java +++ b/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> first() { - TreeCursor cursor = cursor(); - return cursor.hasNext() ? Optional.of(cursor.next()) : Optional.empty(); + return edge(false); } public Optional> last() { - TreeCursor cursor = new TreeCursor<>(this, _ -> true, true); + return edge(true); + } + + private Optional> 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 cursor = new TreeCursor<>(this, _ -> true, last); return cursor.hasNext() ? Optional.of(cursor.next()) : Optional.empty(); } diff --git a/engine/src/main/java/io/hstore/engine/txn/Retained.java b/engine/src/main/java/io/hstore/engine/txn/Retained.java new file mode 100644 index 00000000..5cd6174f --- /dev/null +++ b/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; + } +} diff --git a/engine/src/main/java/io/hstore/engine/txn/Snapshot.java b/engine/src/main/java/io/hstore/engine/txn/Snapshot.java index 6e0e515e..b9fd2629 100644 --- a/engine/src/main/java/io/hstore/engine/txn/Snapshot.java +++ b/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(); } } } diff --git a/engine/src/main/java/io/hstore/engine/txn/Transaction.java b/engine/src/main/java/io/hstore/engine/txn/Transaction.java index 8f4224c7..ce26edb0 100644 --- a/engine/src/main/java/io/hstore/engine/txn/Transaction.java +++ b/engine/src/main/java/io/hstore/engine/txn/Transaction.java @@ -51,6 +51,7 @@ private record KeyRead(Slot slot, long key) { private final TransactionManager manager; private final long id; + private final Retained retained; private final Generation snapshot; private final Branch branch; private final TxnOptions options; @@ -64,9 +65,10 @@ private record KeyRead(Slot slot, long key) { private long[] spilled = new long[0]; private int spilledPages; - Transaction(TransactionManager manager, long id, Generation snapshot, TxnOptions options) { + Transaction(TransactionManager manager, long id, Retained retained, Generation snapshot, TxnOptions options) { this.manager = manager; this.id = id; + this.retained = retained; this.snapshot = snapshot; this.branch = snapshot.branch(options.branch()); this.options = options; @@ -277,7 +279,7 @@ public CommitResult commit() { } throw failure; } finally { - manager.unpin(snapshot.id()); + retained.unpin(); } } @@ -288,7 +290,7 @@ public void abort() { manager.logAbort(id); manager.discard(spilled()); } - manager.unpin(snapshot.id()); + retained.unpin(); } } diff --git a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java index d307b775..9766af5f 100644 --- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java +++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java @@ -77,11 +77,10 @@ private record PageWrite(long pageId, MemorySegment image) { private final Storage storage; private final List derivations; - private final AtomicReference current; + private final AtomicReference current; private final AtomicReference head; private final GroupCommitter committer; - private final ConcurrentSkipListMap history = new ConcurrentSkipListMap<>(); - private final ConcurrentSkipListMap pins = new ConcurrentSkipListMap<>(); + private final ConcurrentSkipListMap history = new ConcurrentSkipListMap<>(); private final ReentrantLock commitLock = new ReentrantLock(); private final ConcurrentSkipListMap commitEnds = new ConcurrentSkipListMap<>(); private final AtomicLong unwritten = new AtomicLong(); @@ -103,13 +102,12 @@ private record PageWrite(long pageId, MemorySegment image) { public TransactionManager(Storage storage, Generation recovered, Collection retained) { this.storage = storage; this.derivations = Stream.concat(EngineDerivations.ALL.stream(), storage.derivations().stream()).toList(); - this.current = new AtomicReference<>(recovered); this.head = new AtomicReference<>(recovered); this.committer = new GroupCommitter(this::makeDurable); this.atoms = new AtomicLong(recovered.nextAtom()); this.txns = new AtomicLong(recovered.nextTxn()); - retained.forEach(generation -> history.put(generation.id(), generation)); - history.put(recovered.id(), recovered); + retained.forEach(this::retain); + this.current = new AtomicReference<>(retain(recovered)); commitEnds.put(recovered.id(), storage.wal().end()); } @@ -122,11 +120,22 @@ public SlotRegistry slots() { } public Generation current() { - return current.get(); + return current.get().generation(); } public Collection history() { - return List.copyOf(history.values()); + return history.values().stream().map(Retained::generation).toList(); + } + + private Retained retain(Generation generation) { + Retained retained = history.get(generation.id()); + if (retained == null) { + retained = new Retained(generation); + history.put(generation.id(), retained); + } else { + retained.rebind(generation); + } + return retained; } public long allocateAtom() { @@ -138,66 +147,58 @@ Workspace workspace(WriteScope scope, RootVector base, Workspace.EdgeGuard guard } public Snapshot snapshot(int branch) { - Generation generation = pinCurrent(); - try { - return new Snapshot(this, generation, generation.branch(branch)); - } catch (RuntimeException e) { - unpin(generation.id()); - throw e; - } + return snapshot(pinCurrent(), branch); } public Snapshot snapshotAt(long generationId, int branch) { - pin(generationId); - Generation generation = history.get(generationId); - if (generation == null) { - unpin(generationId); + Retained retained = history.get(generationId); + if (retained == null || !retained.pin()) { throw HStoreException.invalid("generation " + generationId + " is no longer retained"); } + return snapshot(retained, branch); + } + + private Snapshot snapshot(Retained retained, int branch) { try { - return new Snapshot(this, generation, generation.branch(branch)); + Generation generation = retained.generation(); + return new Snapshot(this, retained, generation, generation.branch(branch)); } catch (RuntimeException e) { - unpin(generationId); + retained.unpin(); throw e; } } public Optional generationAsOf(long wallTime) { - return history.descendingMap().values().stream().filter(generation -> generation.wallTime() <= wallTime).findFirst(); + return history.descendingMap().values().stream().map(Retained::generation) + .filter(generation -> generation.wallTime() <= wallTime).findFirst(); } public Transaction begin(TxnOptions options) { - Generation generation = pinCurrent(); + Retained retained = pinCurrent(); try { - return new Transaction(this, txns.getAndIncrement(), generation, options); + return new Transaction(this, txns.getAndIncrement(), retained, retained.generation(), options); } catch (RuntimeException e) { - unpin(generation.id()); + retained.unpin(); throw e; } } - private Generation pinCurrent() { + private Retained pinCurrent() { while (true) { - Generation generation = current.get(); - pin(generation.id()); - if (history.containsKey(generation.id())) { - return generation; + Retained retained = current.get(); + if (retained.pin()) { + return retained; } - unpin(generation.id()); } } - void pin(long generation) { - pins.merge(generation, 1, Integer::sum); - } - - void unpin(long generation) { - pins.computeIfPresent(generation, (_, count) -> count > 1 ? count - 1 : null); - } - public OptionalLong oldestPinned() { - Map.Entry first = pins.firstEntry(); - return first == null ? OptionalLong.empty() : OptionalLong.of(first.getKey()); + for (Map.Entry entry : history.entrySet()) { + if (entry.getValue().pinned()) { + return OptionalLong.of(entry.getKey()); + } + } + return OptionalLong.empty(); } public long oldestReadable() { @@ -453,8 +454,7 @@ private Appended append(Generation latest, Branch changed, long txnId, List { - history.put(next.id(), next); - current.set(next); + current.set(retain(next)); storage.references().provisional(next.id(), provisional); if (!event.isEmpty()) { storage.feed().append(event); @@ -554,8 +554,7 @@ public void checkpointed(Generation materialized, Epoch epoch) { exclusive(() -> { Generation rebound = rebound(head.get(), rebinding); head.set(rebound); - current.set(rebound); - history.put(rebound.id(), rebound); + current.set(retain(rebound)); return null; }); recountUnwritten(); @@ -583,16 +582,11 @@ private void makeDurable(int batchSize, boolean wrotePages) { private void trimHistory() { long excess = history.size() - storage.historyLimit(); - long latest = current.get().id(); - for (Iterator generations = history.keySet().iterator(); excess > 0 && generations.hasNext(); ) { - long generation = generations.next(); - if (generation == latest || pins.containsKey(generation)) { - continue; - } - Generation trimmed = history.remove(generation); - if (pins.containsKey(generation)) { - history.put(generation, trimmed); - } else { + Retained latest = current.get(); + for (Iterator generations = history.values().iterator(); excess > 0 && generations.hasNext(); ) { + Retained generation = generations.next(); + if (generation != latest && generation.drop()) { + generations.remove(); excess--; } } diff --git a/engine/src/test/java/io/hstore/engine/EngineTest.java b/engine/src/test/java/io/hstore/engine/EngineTest.java index e951cb60..1c16740c 100644 --- a/engine/src/test/java/io/hstore/engine/EngineTest.java +++ b/engine/src/test/java/io/hstore/engine/EngineTest.java @@ -43,6 +43,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.OptionalLong; import java.util.Set; import java.util.TreeSet; import java.util.concurrent.CopyOnWriteArrayList; @@ -525,6 +526,35 @@ void giantBulkLoadSpillsAndSupportsAlgebra() { } } + @Test + void aPinnedGenerationOutlivesTheHistoryLimitUntilItsReadersClose() { + try (StorageEngine engine = StorageEngine.open(directory, small().withHistoryLimit(2))) { + engine.write(txn -> txn.createNode(NODE, "first")); + long pinned = engine.transactions().current().id(); + Snapshot held = engine.snapshot(); + Transaction open = engine.transactions().begin(TxnOptions.defaults()); + for (int i = 0; i < 5; i++) { + int index = i; + engine.write(txn -> txn.createNode(NODE, "later-" + index)); + } + assertEquals(OptionalLong.of(pinned), engine.transactions().oldestPinned()); + try (Snapshot again = engine.snapshotAt(pinned, Branch.MAIN)) { + assertEquals(pinned, again.generation()); + } + held.close(); + held.close(); + assertEquals(OptionalLong.of(pinned), engine.transactions().oldestPinned()); + open.abort(); + assertEquals(OptionalLong.empty(), engine.transactions().oldestPinned()); + engine.write(txn -> txn.createNode(NODE, "trims")); + assertThrows(HStoreException.class, () -> engine.snapshotAt(pinned, Branch.MAIN)); + assertEquals(2, engine.transactions().history().size()); + try (Snapshot current = engine.snapshot()) { + assertEquals(OptionalLong.of(current.generation()), engine.transactions().oldestPinned()); + } + } + } + @Test void compactionReclaimsChurnedSegments() { EngineOptions options = small().withPagesPerSegment(64).withHistoryLimit(2).withMaintenanceInterval(Duration.ofHours(1)); diff --git a/engine/src/test/java/io/hstore/engine/tree/TreeTest.java b/engine/src/test/java/io/hstore/engine/tree/TreeTest.java index 52dbcfcf..c18d6b8e 100644 --- a/engine/src/test/java/io/hstore/engine/tree/TreeTest.java +++ b/engine/src/test/java/io/hstore/engine/tree/TreeTest.java @@ -230,6 +230,34 @@ void diffReportsExactDeltasAndSkipsSharedPages() { assertTrue(pages.loads < 40, "diff visited " + pages.loads + " pages"); } + @Test + void firstAndLastFollowEditsAtBothEnds() { + MemoryPages pages = new MemoryPages(1024); + assertEquals(Optional.empty(), Tree.empty(LONGS, pages).first()); + assertEquals(Optional.empty(), Tree.empty(LONGS, pages).last()); + TreeMap oracle = asMap(new TreeSet<>(LongStream.range(0, 20_000).map(x -> x * 3).boxed().toList())); + Tree tree = pages.persist(build(pages, new TreeSet<>(oracle.keySet()))); + assertTrue(tree.height() > 1); + WriteScope scope = new WriteScope(); + for (int round = 0; round < 300; round++) { + assertEquals(oracle.firstKey(), tree.first().orElseThrow().key()); + assertEquals(oracle.firstEntry().getValue(), tree.first().orElseThrow().value()); + assertEquals(oracle.lastKey(), tree.last().orElseThrow().key()); + assertEquals(oracle.lastEntry().getValue(), tree.last().orElseThrow().value()); + long low = oracle.firstKey(); + long high = oracle.lastKey(); + tree = tree.remove(scope, low).remove(scope, high); + oracle.remove(low); + oracle.remove(high); + if (round % 3 == 0) { + tree = tree.put(scope, high + 1, (long) round); + oracle.put(high + 1, (long) round); + } + } + assertEquals(oracle.firstKey(), pages.persist(tree).first().orElseThrow().key()); + assertEquals(oracle.lastKey(), pages.persist(tree).last().orElseThrow().key()); + } + @Test void keysetPaginationIsStable() { MemoryPages pages = new MemoryPages(1024);