diff --git a/benchmarks/src/main/java/io/hstore/bench/Comparison.java b/benchmarks/src/main/java/io/hstore/bench/Comparison.java index 32f8dbd..d6b8db8 100644 --- a/benchmarks/src/main/java/io/hstore/bench/Comparison.java +++ b/benchmarks/src/main/java/io/hstore/bench/Comparison.java @@ -70,6 +70,8 @@ private interface Slice { } private static final int BATCH = 1_000; + private static final int SCAN_PASSES = 400; + private static final int SCAN_WARMUP = 40; private static final int COMMITS = 5_000; private static final int ID_BYTES = 8; private static final int UPDATE_BYTES = ID_BYTES + 24; @@ -167,10 +169,13 @@ private static void run(Map options) throws Exception { store.largeEdge(largeSize); return largeSize; }))); - store.scanLargeEdge(); - results.add(reading(store, () -> measure("large.scan", "enumerate all members of the large hyperedge (x20)", 20L * largeSize, () -> { + for (int i = 0; i < SCAN_WARMUP; i++) { + store.scanLargeEdge(); + } + results.add(reading(store, () -> measureReading("large.scan", "enumerate all members of the large hyperedge (x%d, after %d warm-up passes)" + .formatted(SCAN_PASSES, SCAN_WARMUP), (long) SCAN_PASSES * largeSize, () -> { long total = 0; - for (int i = 0; i < 20; i++) { + for (int i = 0; i < SCAN_PASSES; i++) { total += store.scanLargeEdge(); } return total; @@ -295,6 +300,13 @@ private static Result measure(String workload, String description, long operatio return new Result(workload, description, operations, (System.nanoTime() - started) / 1e6, checksum).using(JvmUsage.now().since(usage)); } + private static Result measureReading(String workload, String description, long operations, LongSupplier work) { + JvmUsage usage = JvmUsage.onThisThread(); + long started = System.nanoTime(); + long checksum = work.getAsLong(); + return new Result(workload, description, operations, (System.nanoTime() - started) / 1e6, checksum).using(JvmUsage.onThisThread().since(usage)); + } + private static long batched(int length, Slice slice) { long total = 0; for (int from = 0; from < length; from += BATCH) { @@ -307,7 +319,7 @@ private static Result read(Store store, String workload, String description, int if (warm) { batched(Math.max(BATCH, operations / 10), slice); } - return reading(store, () -> measure(workload, description, operations, () -> batched(operations, slice))); + return reading(store, () -> measureReading(workload, description, operations, () -> batched(operations, slice))); } private static Result writing(Store store, long logicalBytes, Supplier workload) { diff --git a/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java b/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java index 99d442b..f0dafb9 100644 --- a/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java +++ b/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java @@ -20,6 +20,7 @@ import java.util.List; import java.util.Map; import java.util.function.IntConsumer; +import java.util.function.LongConsumer; final class HStoreStore implements Store { @@ -140,11 +141,11 @@ public long incidence(int[] probes, int from, int to) { @Override public long members(int[] probes, int from, int to) { return database.read(reader -> { - long total = 0; + MemberCount count = new MemberCount(); for (int i = from; i < to; i++) { - total += reader.memberIds(edges[probes[i]]).filter(atom -> atom != 0).count(); + reader.forEachMemberId(edges[probes[i]], count); } - return total; + return count.total; }); } @@ -198,7 +199,11 @@ public void largeEdge(int count) { @Override public long scanLargeEdge() { - return database.read(reader -> reader.memberIds(large).filter(atom -> atom != 0).count()); + return database.read(reader -> { + MemberCount count = new MemberCount(); + reader.forEachMemberId(large, count); + return count.total; + }); } @Override @@ -266,4 +271,15 @@ public void reopen(Runnable whileClosed) { public void close() { database.close(); } + + private static final class MemberCount implements LongConsumer { + long total; + + @Override + public void accept(long atom) { + if (atom != 0) { + total++; + } + } + } } diff --git a/benchmarks/src/main/java/io/hstore/bench/JvmUsage.java b/benchmarks/src/main/java/io/hstore/bench/JvmUsage.java index 3a64faf..b68dc9e 100644 --- a/benchmarks/src/main/java/io/hstore/bench/JvmUsage.java +++ b/benchmarks/src/main/java/io/hstore/bench/JvmUsage.java @@ -9,13 +9,22 @@ record JvmUsage(long collections, long pauseMillis, long allocated) { static JvmUsage now() { + return now(true); + } + + static JvmUsage onThisThread() { + return now(false); + } + + private static JvmUsage now(boolean allThreads) { long collections = 0; long pauseMillis = 0; for (GarbageCollectorMXBean collector : ManagementFactory.getGarbageCollectorMXBeans()) { collections += Math.max(0, collector.getCollectionCount()); pauseMillis += Math.max(0, collector.getCollectionTime()); } - long allocated = ((com.sun.management.ThreadMXBean) ManagementFactory.getThreadMXBean()).getTotalThreadAllocatedBytes(); + com.sun.management.ThreadMXBean threads = (com.sun.management.ThreadMXBean) ManagementFactory.getThreadMXBean(); + long allocated = allThreads ? threads.getTotalThreadAllocatedBytes() : threads.getCurrentThreadAllocatedBytes(); return new JvmUsage(collections, pauseMillis, allocated); } diff --git a/database/src/main/java/io/hstore/db/Reader.java b/database/src/main/java/io/hstore/db/Reader.java index caaa43b..9aafec3 100644 --- a/database/src/main/java/io/hstore/db/Reader.java +++ b/database/src/main/java/io/hstore/db/Reader.java @@ -32,6 +32,7 @@ import java.util.OptionalInt; import java.util.OptionalLong; import java.util.Set; +import java.util.function.LongConsumer; import java.util.function.Predicate; import java.util.stream.LongStream; import java.util.stream.Stream; @@ -203,6 +204,10 @@ public LongStream memberIds(long edge) { return edge(edge).memberIds(); } + public void forEachMemberId(long edge, LongConsumer action) { + edge(edge).forEachMemberId(action); + } + public Stream members(long edge, long validAt) { return edge(edge).validAt(validAt).map(incidence -> member(incidence, -1)); } diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index 90ad61e..63deeda 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -154,7 +154,7 @@ sequenceDiagram Notes: -* Nodes are decoded once and then shared by every reader through `NodeCache`. It has 16 shards, each a `ConcurrentHashMap` plus a CLOCK (second-chance) ring, with a total byte budget of `node_cache_mb`, each node counted at an estimate of its heap size (`Node.heapBytes`). Lookups are lock-free; only admission takes a per-shard lock. Decoded nodes are immutable (`owner == null`), so sharing them needs no copying. See [pages.md](../storage/pages.md#caching-and-io-accounting). +* Nodes are decoded once and then shared by every reader through `NodeCache`. It has 16 shards, each an open-addressing table keyed by page number plus a CLOCK (second-chance) ring, with a total byte budget of `node_cache_mb`, each node counted at an estimate of its heap size (`Node.heapBytes`). Lookups are lock-free; only admission takes a per-shard lock. Decoded nodes are immutable (`owner == null`), so sharing them needs no copying. See [pages.md](../storage/pages.md#caching-and-io-accounting). * `IoTrace` is bound with `ScopedValue.where(...)` around a query. `recordRead` throws `HStoreException.limit` once page reads plus cache hits exceed the query's page budget (`query_page_budget`). * Historical reads use `TransactionManager.snapshotAt(generation, branch)`. It looks the generation up in the retained `history` map; see [catalog-and-generations.md](../storage/catalog-and-generations.md#history-retention-and-time-travel). diff --git a/docs/benchmarks.md b/docs/benchmarks.md index cd5cbe2..1d69b7f 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -43,7 +43,7 @@ taken from a common hyperedge (so co-membership counts are non-zero), and 20,000 | `ingest.nodes` | insert nodes, 1,000 per transaction | `Writer.node` | `graph.add(String)` in `transact` | | `ingest.edges` | insert hyperedges, 1,000 per transaction | `Writer.edge` + `Writer.load` | `graph.add(new HGPlainLink(…))` | | `read.incidence` | enumerate a node's incidence set, 1,000 probes per read transaction | `Reader.incident` | `getIncidenceSet(h).getSearchResult()` | -| `read.members` | enumerate a hyperedge's members | `Reader.members` | `HGLink.getTargetAt` over `graph.get(h)` | +| `read.members` | enumerate a hyperedge's members | `Reader.forEachMemberId` | `HGLink.getTargetAt` over `graph.get(h)` | | `read.comembership` | count hyperedges containing both nodes of a pair | `TreeAlgebra.intersectKeys` (leapfrog) | `hg.count(hg.and(hg.incident(a), hg.incident(b)))` | | `read.twohop` | count the distinct nodes that share a hyperedge with a node, 10,000 probes | `Reader.incident` + `Reader.memberIds` | `getIncidenceSet` + the link's targets | | `read.incidence.parallel` | `read.incidence` from one thread per core | same | same | @@ -53,7 +53,7 @@ taken from a common hyperedge (so co-membership counts are non-zero), and 20,000 | `mixed.read` | single incidence lookups from `threads − 1` threads for 10 s while `mixed.write` runs | `Reader.incident` | `getIncidenceSet(h).getSearchResult()` | | `mixed.write` | one thread committing 10 updates at a time on a fixed schedule of 1,000 updates/s | `Writer.set` | `graph.replace` | | `large.ingest` | one hyperedge containing every node | `Writer.load` | one `HGPlainLink` | -| `large.scan` | enumerate the large hyperedge twenty times | `Reader.members` | `getTargetAt` loop | +| `large.scan` | enumerate the large hyperedge 400 times, after 40 warm-up passes | `Reader.forEachMemberId` | `getTargetAt` loop | | `large.probe` | test whether a node belongs to the large hyperedge | `View.incidence(edge, node)` (membership-tree lookup) | `getIncidenceSet(node).contains(edge)` | | `reopen` | close and reopen, including recovery | `HypergraphDatabase.open` | `HGEnvironment.get` | | `read.incidence.cold` | `read.incidence` immediately after reopening | | | @@ -144,8 +144,9 @@ then the rest of the page image if it's bigger. JE reports sequential plus rando cold reads and with a small `CACHE_MB`. The logical sizes are rough estimates; compare the two stores against each other rather than reading the ratios as absolute numbers. -**Memory.** Every workload also records, from the JVM's own MXBeans, the bytes the whole process allocated -(`getTotalThreadAllocatedBytes`) and the GC pause time (`GarbageCollectorMXBean.getCollectionTime`, which is +**Memory.** Every workload also records, from the JVM's own MXBeans, the bytes allocated +(`getTotalThreadAllocatedBytes` for the whole process; for the single-threaded read workloads, only the reading +thread's `getCurrentThreadAllocatedBytes`, so background maintenance doesn't count against a read) and the GC pause time (`GarbageCollectorMXBean.getCollectionTime`, which is stop-the-world time with ParallelGC). Rows where both stores paused for 0 ms are left out. After ingest, the run forces a full GC and records how much heap the loaded store keeps. A baseline taken before the store opens is subtracted, so the shared dataset arrays don't count. This number mostly reflects how each engine's cache is @@ -244,8 +245,9 @@ not what drives HStore's footprint: atom cache, which holds a Java array of target handles, so a scan is an array loop. HStore walks a B+tree cursor, decodes incidence columns, resolves role sets and builds a `Member` record for each member. That record carries roles, weight, validity, position and qualifier. The cost buys per-member metadata and - snapshot-consistent reads, which a plain link cannot express. This is the clearest optimisation target - (a raw `LongStream` member path that skips record materialisation). + snapshot-consistent reads, which a plain link cannot express. The benchmark now reads ids through + `Reader.forEachMemberId`, which builds neither `Member` records nor tree entries; the numbers below predate that. + What remains is that every hyperedge keeps its members in a tree of its own, one more page to read. * **Edge ingest at small batch sizes (about 0.6–0.8× in async mode).** Each commit rewrites the incident-tree paths of every touched member, which is copy-on-write write amplification. JE appends log records and defers the work to its cleaner. In `sync` mode group commit amortises the fsync and HStore leads (1.35×). diff --git a/docs/database/data-model.md b/docs/database/data-model.md index 1150811..bfd0356 100644 --- a/docs/database/data-model.md +++ b/docs/database/data-model.md @@ -461,6 +461,7 @@ Optional.empty | `property(atom, name[, validAt])`, `properties`, `text(value)`, `document` | property access; `text` dereferences payloads | | `edge`, `members`, `members(edge, validAt)`, `membersWithRole`, `membersWeighted` | membership access with summary pruning | | `memberIds(edge)` | member ids only, as a `LongStream` in membership order, without building `Member` records | +| `forEachMemberId(edge, action)` | the same ids handed to a `LongConsumer`, with nothing allocated per member | | `cardinality`, `degree`, `incident`, `position` | topology statistics and reverse lookup | | `lookup`, `range` | property and JSON index access with exact re-check | | `similar(text, k, consistency)`, `similar(model, vector, k, consistency)`, `embedding` | [semantic](semantic.md) | diff --git a/docs/storage/pages.md b/docs/storage/pages.md index 64925c6..95f5a27 100644 --- a/docs/storage/pages.md +++ b/docs/storage/pages.md @@ -313,11 +313,11 @@ read-only memory, so nothing can write into a stored image through a read. The cache stores **decoded, frozen** `Node` objects, not bytes. A hit therefore costs no decoding and no allocation. The cache has 16 shards; a node goes to shard `mix(pageId) >>> 60`. Each shard has: -* a `ConcurrentHashMap` from page id to `Slot(pageId, node, volatile referenced)`; +* an open-addressing table of `Slot(pageId, node, volatile referenced)`, probed linearly from `mix(pageId)`, so a lookup needs no boxed key. It is kept at most half full, counting removed slots, and is rebuilt into a new array when it would pass that; * a byte budget of `max(64 KiB, node_cache_mb / 16)` and a circular, doubly linked ring of slots with a CLOCK hand. Each slot weighs `Node.heapBytes()`, an estimate of the decoded node's heap size: 144 bytes for the node and its summary, its key and value or child arrays, then 96 bytes plus the encoded value per leaf entry, or 128 bytes per child reference in a branch. Those constants come from measuring retained heap against cache contents at two dataset sizes; * a `ReentrantLock` taken only for admission. -**Reads are lock-free.** `get` is a `ConcurrentHashMap.get`, plus setting `referenced = true` if it was clear. The check avoids a redundant volatile write on hot entries. +**Reads are lock-free.** `get` probes the table, reading each slot with acquire ordering, plus setting `referenced = true` if it was clear. Writers change the table only under the shard lock, publishing each slot with release ordering, and a rebuild publishes the new array through a volatile field. A reader still probing the old array can find a node that was just evicted; that is harmless, because a page number's contents never change. The check avoids a redundant volatile write on hot entries. **Admission (`put`) uses CLOCK second-chance eviction under the shard lock:** diff --git a/docs/storage/persistent-tree.md b/docs/storage/persistent-tree.md index 8bc3c89..a9b08ae 100644 --- a/docs/storage/persistent-tree.md +++ b/docs/storage/persistent-tree.md @@ -35,7 +35,7 @@ Leaf Branch Summary summary (volatile, recomputed lazily after mutation) ``` -**Routing.** `Branch.route(key)` binary-searches `separators[1..size-1]` for the last separator `<= key` and returns index 0 when there is none. `separators[0]` therefore never takes part in routing. A freshly grown root stores `Long.MIN_VALUE` there. When a branch is frozen (`Branch.frozenWith`), `separators[0]` is rewritten to the first child's `summary().min()`, so the encoded page carries its real lower bound and delta-encodes compactly. +**Routing.** `Branch.route(key)` searches `separators[1..size-1]` for the last separator `<= key` and returns index 0 when there is none. In a branch with 16 or more children it first guesses the position from where `key` falls between the first and last separator, checks a window of 8 around the guess, and only then falls back to a binary search over what is left. Page numbers and atom ids are handed out in order, so the guess is usually right. `Leaf.search` does the same over a leaf's keys and returns exactly what `Arrays.binarySearch` would. `separators[0]` therefore never takes part in routing. A freshly grown root stores `Long.MIN_VALUE` there. When a branch is frozen (`Branch.frozenWith`), `separators[0]` is rewritten to the first child's `summary().min()`, so the encoded page carries its real lower bound and delta-encodes compactly. **Page budget.** `Layout.of(pageSize)` computes: @@ -180,7 +180,9 @@ A `WriteScope` built with a `PageSink` (every `Transaction` has one) can spill c ### Cursors with summary pruning -A `TreeCursor` (`TreeCursor.java`) keeps an explicit stack of up to 48 `(branch, slot)` pairs and is built with an `admit: Predicate`. A child subtree is entered only if `admit.test(child.summary())`. Because the summary sits in the parent, a rejected subtree is skipped without I/O. Queries built on this: +A `TreeCursor` (`TreeCursor.java`) keeps an explicit stack of `(branch, slot)` pairs, with room for four levels to start with and growing up to 48, and is built with an `admit: Predicate`. A child subtree is entered only if `admit.test(child.summary())`. Because the summary sits in the parent, a rejected subtree is skipped without I/O. Queries built on this: + +`Tree.keys()`, `Tree.forEachKey` and `TreeCursor.forEachRemainingKey` walk the keys straight out of each leaf's `long[]` without building an `Entry` per key, and `Tree.forEachValue` does the same for values. Use them when only keys or only values are needed. | Query | Predicate | |---|---| diff --git a/engine/src/main/java/io/hstore/engine/topology/Hyperedge.java b/engine/src/main/java/io/hstore/engine/topology/Hyperedge.java index b3d1e7f..fa4e5d0 100644 --- a/engine/src/main/java/io/hstore/engine/topology/Hyperedge.java +++ b/engine/src/main/java/io/hstore/engine/topology/Hyperedge.java @@ -23,6 +23,7 @@ import java.util.Set; import java.util.function.BinaryOperator; import java.util.function.Consumer; +import java.util.function.LongConsumer; import java.util.function.UnaryOperator; import java.util.stream.LongStream; import java.util.stream.Stream; @@ -127,6 +128,13 @@ public LongStream memberIds() { }; } + public void forEachMemberId(LongConsumer action) { + switch (kind) { + case SET -> members.forEachKey(action); + case ORDERED -> members.forEachValue(incidence -> action.accept(incidence.member())); + } + } + public Stream validAt(long instant) { return members.scan(summary -> summary.mayBeValidAt(instant)) .map(Entry::value) diff --git a/engine/src/main/java/io/hstore/engine/tree/Branch.java b/engine/src/main/java/io/hstore/engine/tree/Branch.java index dd424b7..1f39dcf 100644 --- a/engine/src/main/java/io/hstore/engine/tree/Branch.java +++ b/engine/src/main/java/io/hstore/engine/tree/Branch.java @@ -47,6 +47,23 @@ int route(long key) { int low = 1; int high = size - 1; int result = 0; + if (size >= Trees.INTERPOLATE_FROM && key > separators[1] && key < separators[size - 1]) { + long first = separators[1]; + int guess = Math.clamp(1 + (long) (((double) key - first) / ((double) separators[size - 1] - first) * (size - 2)), 1, size - 1); + if (separators[guess] <= key) { + int end = Math.min(size - 1, guess + Trees.INTERPOLATE_WINDOW); + result = separators[end] <= key ? end : guess; + low = result + 1; + high = separators[end] <= key ? high : end - 1; + } else { + int start = Math.max(1, guess - Trees.INTERPOLATE_WINDOW); + if (separators[start] <= key) { + result = start; + low = start + 1; + } + high = separators[start] <= key ? guess - 1 : start - 1; + } + } while (low <= high) { int mid = (low + high) >>> 1; if (separators[mid] <= key) { diff --git a/engine/src/main/java/io/hstore/engine/tree/Leaf.java b/engine/src/main/java/io/hstore/engine/tree/Leaf.java index ccd124a..79dae89 100644 --- a/engine/src/main/java/io/hstore/engine/tree/Leaf.java +++ b/engine/src/main/java/io/hstore/engine/tree/Leaf.java @@ -65,6 +65,23 @@ int heapBytes() { } int search(long key) { + if (size >= Trees.INTERPOLATE_FROM) { + long low = keys[0]; + long high = keys[size - 1]; + if (key > low && key < high) { + int guess = Math.clamp((long) (((double) key - low) / ((double) high - low) * (size - 1)), 0, size - 1); + long found = keys[guess]; + if (found == key) { + return guess; + } + if (found < key) { + int end = Math.min(size, guess + Trees.INTERPOLATE_WINDOW + 1); + return keys[end - 1] >= key ? Arrays.binarySearch(keys, guess + 1, end, key) : Arrays.binarySearch(keys, end, size, key); + } + int start = Math.max(0, guess - Trees.INTERPOLATE_WINDOW); + return keys[start] <= key ? Arrays.binarySearch(keys, start, guess, key) : Arrays.binarySearch(keys, 0, start, key); + } + } return Arrays.binarySearch(keys, 0, size, key); } diff --git a/engine/src/main/java/io/hstore/engine/tree/NodeCache.java b/engine/src/main/java/io/hstore/engine/tree/NodeCache.java index 9eeb06a..b9997fb 100644 --- a/engine/src/main/java/io/hstore/engine/tree/NodeCache.java +++ b/engine/src/main/java/io/hstore/engine/tree/NodeCache.java @@ -3,7 +3,8 @@ package io.hstore.engine.tree; -import java.util.concurrent.ConcurrentHashMap; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; import java.util.concurrent.atomic.LongAdder; import java.util.concurrent.locks.ReentrantLock; @@ -81,18 +82,88 @@ private static final class Slot { } private static final class Shard { - private final ConcurrentHashMap entries = new ConcurrentHashMap<>(); + private static final int MIN_TABLE = 64; + private static final Slot GONE = new Slot(0, null, 0); + private static final VarHandle SLOTS = MethodHandles.arrayElementVarHandle(Slot[].class); + private final ReentrantLock admission = new ReentrantLock(); private final long budget; + private volatile Slot[] table = new Slot[MIN_TABLE]; private volatile long bytes; + private int live; + private int gone; private Slot hand; Shard(long budget) { this.budget = budget; } + private static int home(long pageId, int mask) { + return (int) Hashing.mix(pageId) & mask; + } + + private Slot find(long pageId) { + Slot[] slots = table; + int mask = slots.length - 1; + for (int i = home(pageId, mask); ; i = (i + 1) & mask) { + Slot slot = (Slot) SLOTS.getAcquire(slots, i); + if (slot == null) { + return null; + } + if (slot != GONE && slot.pageId == pageId) { + return slot; + } + } + } + + private void insert(Slot slot) { + if ((live + gone + 1) * 2 > table.length) { + rebuild(Math.max(MIN_TABLE, Integer.highestOneBit(Math.max(1, live + 1) * 4))); + } + Slot[] slots = table; + int mask = slots.length - 1; + int i = home(slot.pageId, mask); + while (slots[i] != null && slots[i] != GONE) { + i = (i + 1) & mask; + } + if (slots[i] == GONE) { + gone--; + } + SLOTS.setRelease(slots, i, slot); + live++; + } + + private void remove(Slot slot) { + Slot[] slots = table; + int mask = slots.length - 1; + for (int i = home(slot.pageId, mask); slots[i] != null; i = (i + 1) & mask) { + if (slots[i] == slot) { + SLOTS.setRelease(slots, i, GONE); + live--; + gone++; + return; + } + } + } + + private void rebuild(int capacity) { + Slot[] fresh = new Slot[capacity]; + int mask = capacity - 1; + for (Slot slot : table) { + if (slot != null && slot != GONE) { + int i = home(slot.pageId, mask); + while (fresh[i] != null) { + i = (i + 1) & mask; + } + fresh[i] = slot; + } + } + gone = 0; + table = fresh; + } + Node get(long pageId) { - Slot slot = entries.get(pageId); + Slot slot = find(pageId); if (slot == null) { return null; } @@ -105,13 +176,13 @@ Node get(long pageId) { void put(long pageId, Node node, int weight) { admission.lock(); try { - Slot existing = entries.get(pageId); + Slot existing = find(pageId); if (existing != null && existing.node == node) { return; } if (existing != null) { unlink(existing); - entries.remove(pageId, existing); + remove(existing); } if (weight > budget) { return; @@ -121,7 +192,7 @@ void put(long pageId, Node node, int weight) { } Slot slot = new Slot(pageId, node, weight); link(slot); - entries.put(pageId, slot); + insert(slot); } finally { admission.unlock(); } @@ -134,7 +205,7 @@ private void evict() { } Slot victim = hand; unlink(victim); - entries.remove(victim.pageId, victim); + remove(victim); } private void link(Slot slot) { @@ -169,7 +240,9 @@ private void unlink(Slot slot) { void clear() { admission.lock(); try { - entries.clear(); + table = new Slot[MIN_TABLE]; + live = 0; + gone = 0; hand = null; bytes = 0; } finally { 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 52063db..e85d173 100644 --- a/engine/src/main/java/io/hstore/engine/tree/Tree.java +++ b/engine/src/main/java/io/hstore/engine/tree/Tree.java @@ -4,12 +4,15 @@ package io.hstore.engine.tree; import java.util.ArrayList; +import java.util.Comparator; import java.util.Iterator; import java.util.List; import java.util.Objects; import java.util.Optional; import java.util.Spliterator; import java.util.Spliterators; +import java.util.function.Consumer; +import java.util.function.LongConsumer; import java.util.function.Predicate; import java.util.function.UnaryOperator; import java.util.stream.LongStream; @@ -190,7 +193,39 @@ public Stream> range(long low, long high) { } public LongStream keys() { - return stream().mapToLong(Entry::key); + TreeCursor cursor = cursor(); + return StreamSupport.longStream(new Spliterators.AbstractLongSpliterator(Long.MAX_VALUE, + Spliterator.ORDERED | Spliterator.DISTINCT | Spliterator.SORTED | Spliterator.NONNULL) { + @Override + public boolean tryAdvance(LongConsumer action) { + if (!cursor.hasNext()) { + return false; + } + action.accept(cursor.key()); + cursor.advance(); + return true; + } + + @Override + public void forEachRemaining(LongConsumer action) { + cursor.forEachRemainingKey(action); + } + + @Override + public Comparator getComparator() { + return null; + } + }, false); + } + + public void forEachKey(LongConsumer action) { + cursor().forEachRemainingKey(action); + } + + public void forEachValue(Consumer action) { + for (TreeCursor cursor = cursor(); cursor.hasNext(); cursor.advance()) { + action.accept(cursor.value()); + } } public Stream values() { diff --git a/engine/src/main/java/io/hstore/engine/tree/TreeCursor.java b/engine/src/main/java/io/hstore/engine/tree/TreeCursor.java index c5d5c1b..c74429a 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeCursor.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeCursor.java @@ -3,20 +3,23 @@ package io.hstore.engine.tree; +import java.util.Arrays; import java.util.Iterator; import java.util.NoSuchElementException; +import java.util.function.LongConsumer; import java.util.function.Predicate; public final class TreeCursor implements Iterator> { private static final int MAX_DEPTH = 48; + private static final int INITIAL_DEPTH = 4; private final Tree tree; private final Predicate admit; private final boolean descending; private final int step; - private final Branch[] path = new Branch[MAX_DEPTH]; - private final int[] slots = new int[MAX_DEPTH]; + private Branch[] path = new Branch[INITIAL_DEPTH]; + private int[] slots = new int[INITIAL_DEPTH]; private int depth = -1; private Leaf leaf; private int position; @@ -104,6 +107,22 @@ public Entry next() { return entry; } + public void forEachRemainingKey(LongConsumer action) { + while (leaf != null) { + Leaf current = leaf; + if (descending) { + for (int i = position; i >= 0; i--) { + action.accept(current.key(i)); + } + } else { + for (int i = position, size = current.size(); i < size; i++) { + action.accept(current.key(i)); + } + } + advanceLeaf(); + } + } + public void advance() { position += step; if (!inLeaf()) { @@ -182,6 +201,11 @@ private int lastChildUpTo(Branch branch, int from, long key) { private void push(Branch branch, int slot) { depth++; + if (depth == path.length) { + int grown = Math.min(MAX_DEPTH, path.length * 2); + path = Arrays.copyOf(path, grown); + slots = Arrays.copyOf(slots, grown); + } path[depth] = branch; slots[depth] = slot; } diff --git a/engine/src/main/java/io/hstore/engine/tree/Trees.java b/engine/src/main/java/io/hstore/engine/tree/Trees.java index cd0bdfc..cf7b270 100644 --- a/engine/src/main/java/io/hstore/engine/tree/Trees.java +++ b/engine/src/main/java/io/hstore/engine/tree/Trees.java @@ -5,6 +5,9 @@ final class Trees { + static final int INTERPOLATE_FROM = 16; + static final int INTERPOLATE_WINDOW = 8; + private Trees() { } diff --git a/engine/src/test/java/io/hstore/engine/tree/NodeCacheTest.java b/engine/src/test/java/io/hstore/engine/tree/NodeCacheTest.java index 987731d..673dbe9 100644 --- a/engine/src/test/java/io/hstore/engine/tree/NodeCacheTest.java +++ b/engine/src/test/java/io/hstore/engine/tree/NodeCacheTest.java @@ -5,6 +5,13 @@ import org.junit.jupiter.api.Test; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.random.RandomGenerator; import java.util.random.RandomGeneratorFactory; import java.util.stream.LongStream; @@ -91,4 +98,62 @@ void largeBudgetsAreSplitIntoShards() { assertEquals(1, NodeCache.shardsFor(1 << 20, PAGE)); assertEquals(1, NodeCache.shardsFor(1, PAGE)); } + + @Test + void everyPageCachedLongEnoughIsFoundAndNothingElseIs() { + NodeCache cache = new NodeCache(64L << 20, PAGE); + for (long page = 1; page <= 50_000; page++) { + cache.put(page, node(page), 100); + if (page % 3 == 0) { + cache.put(page, node(page), 100); + } + } + for (long page = 1; page <= 50_000; page++) { + assertEquals(page, ((Leaf) cache.get(page)).firstKey()); + } + for (long page = 50_001; page <= 60_000; page++) { + assertNull(cache.get(page)); + } + cache.clear(); + assertNull(cache.get(1)); + assertEquals(0, cache.residentBytes()); + } + + @Test + void readersSeeTheRightNodeWhileWritersEvictAndGrowTheTable() throws Exception { + NodeCache cache = new NodeCache(256 * 1024, PAGE); + AtomicBoolean done = new AtomicBoolean(); + AtomicLong wrong = new AtomicLong(); + AtomicLong found = new AtomicLong(); + try (ExecutorService readers = Executors.newFixedThreadPool(4)) { + List> running = new ArrayList<>(); + for (int t = 0; t < 4; t++) { + RandomGenerator random = RandomGeneratorFactory.of("L64X128MixRandom").create(t); + running.add(readers.submit(() -> { + while (!done.get()) { + long page = 1 + random.nextLong(20_000); + Node node = cache.get(page); + if (node != null) { + found.incrementAndGet(); + if (((Leaf) node).firstKey() != page) { + wrong.incrementAndGet(); + } + } + } + })); + } + RandomGenerator random = RandomGeneratorFactory.of("L64X128MixRandom").create(99); + for (int round = 0; round < 300_000; round++) { + long page = 1 + random.nextLong(20_000); + cache.put(page, node(page), 64 + random.nextInt(512)); + } + done.set(true); + for (Future reader : running) { + reader.get(); + } + } + assertEquals(0, wrong.get()); + assertTrue(found.get() > 0); + assertTrue(cache.residentBytes() <= 256 * 1024); + } } diff --git a/engine/src/test/java/io/hstore/engine/tree/NodeSearchTest.java b/engine/src/test/java/io/hstore/engine/tree/NodeSearchTest.java new file mode 100644 index 0000000..7f02c4c --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/tree/NodeSearchTest.java @@ -0,0 +1,64 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.tree; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Arrays; +import java.util.random.RandomGenerator; +import java.util.random.RandomGeneratorFactory; +import java.util.stream.LongStream; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class NodeSearchTest { + + private static long[] keys(RandomGenerator random, int size, int shape) { + return switch (shape) { + case 0 -> LongStream.range(0, size).map(i -> 1_000 + i).toArray(); + case 1 -> random.longs(size, -1_000_000, 1_000_000).distinct().sorted().toArray(); + case 2 -> LongStream.range(0, size).map(i -> i * i * i).toArray(); + default -> LongStream.concat(LongStream.range(0, size / 2), LongStream.range(0, size - size / 2).map(i -> Long.MAX_VALUE / 2 + i)).toArray(); + }; + } + + private static long probe(RandomGenerator random, long[] keys) { + long span = keys[keys.length - 1] - keys[0]; + return switch (random.nextInt(3)) { + case 0 -> keys[random.nextInt(keys.length)]; + case 1 -> keys[random.nextInt(keys.length)] + (random.nextBoolean() ? 1 : -1); + default -> keys[0] - 5 + (long) (random.nextDouble() * (span + 10)); + }; + } + + private static int route(long[] separators, int size, long key) { + int result = 0; + for (int i = 1; i < size; i++) { + if (separators[i] <= key) { + result = i; + } + } + return result; + } + + @ParameterizedTest + @ValueSource(ints = {0, 1, 2, 3}) + void interpolationFindsWhatABinarySearchFinds(int shape) { + RandomGenerator random = RandomGeneratorFactory.of("L64X128MixRandom").create(shape); + for (int round = 0; round < 400; round++) { + long[] keys = keys(random, 1 + random.nextInt(600), shape); + int size = keys.length; + Leaf leaf = Leaf.owned(TreeTest.LONGS, null, keys, Arrays.stream(keys).boxed().toArray(), size); + Ref[] children = new Ref[size]; + Arrays.fill(children, Ref.EMPTY); + Branch branch = Branch.owned(TreeTest.LONGS, null, keys.clone(), children, size, 1); + for (int i = 0; i < 200; i++) { + long key = probe(random, keys); + assertEquals(Arrays.binarySearch(keys, 0, size, key), leaf.search(key), "leaf search for " + key); + assertEquals(route(keys, size, key), branch.route(key), "branch route for " + key); + } + } + } +} 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 eebdc77..f3d8eb1 100644 --- a/engine/src/test/java/io/hstore/engine/tree/TreeTest.java +++ b/engine/src/test/java/io/hstore/engine/tree/TreeTest.java @@ -16,6 +16,7 @@ import java.lang.foreign.MemorySegment; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; @@ -27,6 +28,7 @@ import java.util.random.RandomGeneratorFactory; import java.util.stream.LongStream; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -68,6 +70,48 @@ void randomMutationsMatchOracle(long seed) { assertMatches(oracle, pages.persist(tree)); } + @Test + void keysAndCursorsWalkADeepTreeInOrder() { + RandomGenerator random = random(9); + MemoryPages pages = new MemoryPages(1024); + WriteScope scope = new WriteScope(); + Tree tree = Tree.empty(LONGS, pages); + TreeMap oracle = new TreeMap<>(); + for (int step = 0; step < 400_000; step++) { + long key = random.nextLong(100_000_000); + tree = tree.put(scope, key, key % 97); + oracle.put(key, key % 97); + if (step % 49_999 == 0) { + tree = pages.persist(tree); + scope.freeze(); + } + } + tree = pages.persist(tree); + assertTrue(tree.height() > 5, "height " + tree.height()); + long[] expected = oracle.keySet().stream().mapToLong(Long::longValue).toArray(); + assertArrayEquals(expected, tree.keys().toArray()); + LongStream.Builder visited = LongStream.builder(); + tree.forEachKey(visited::add); + assertArrayEquals(expected, visited.build().toArray()); + assertArrayEquals(Arrays.copyOf(expected, 25), tree.keys().limit(25).toArray()); + List values = new ArrayList<>(); + tree.forEachValue(values::add); + assertEquals(List.copyOf(oracle.values()), values); + long[] reversed = LongStream.range(0, expected.length).map(i -> expected[expected.length - 1 - (int) i]).toArray(); + assertArrayEquals(reversed, tree.descending().mapToLong(Entry::key).toArray()); + for (int probe = 0; probe < 2_000; probe++) { + long key = random.nextLong(100_000_100) - 50; + assertEquals(Optional.ofNullable(oracle.ceilingKey(key)), tree.ceiling(key).map(Entry::key), "ceiling " + key); + assertEquals(Optional.ofNullable(oracle.floorKey(key)), tree.floor(key).map(Entry::key), "floor " + key); + TreeCursor cursor = tree.cursor(); + if (probe < 20 && cursor.seek(key)) { + LongStream.Builder rest = LongStream.builder(); + cursor.forEachRemainingKey(rest::add); + assertArrayEquals(oracle.tailMap(key, true).keySet().stream().mapToLong(Long::longValue).toArray(), rest.build().toArray()); + } + } + } + @Test void permutationsYieldEqualFingerprints() { List keys = new ArrayList<>(LongStream.range(0, 3000).boxed().toList());