feat(hg_analytics): the scale thesis, measured — parallel + distributed + out-of-core#26
Merged
Merged
Conversation
…thesis, measured) Cashes "Rust + our architecture outscales them" with numbers, not assertion: - pagerank_parallel: pull-based (each node's rank computed from its in-neighbours → no write contention), O(E) work parallel, O(n) reductions serial. Memory-bandwidth-bound (random rank[] gather) so it scales modestly — HONEST ceiling. - betweenness_parallel: the source loop is embarrassingly parallel + COMPUTE-bound → near-linear. Deterministic: sources split into a FIXED chunk count (independent of thread count) and partials summed in chunk order, so the result is bit-identical on any core count. Measured (8-core M-series, examples/scale_bench.rs, release): PageRank 2M nodes / 20M edges / 20 iters: 1.78x, 147M edges·iter/s, Δ vs serial 8e-22 Betweenness 6k nodes / 48k edges: 4.86x (61% of linear), matches serial (Δ 9e-11) Both bit-deterministic 1-thread == 8-thread. 12 tests (incl. parallel==serial + determinism). The betweenness number is the real proof: the compute-bound analytics scale with cores, in Rust, deterministically. Next legs (spec'd, not built): out-of-core CSR storage (kill the RAM ceiling) + distributed query over the federation (federation partition = shard — the move Neptune can't make).
…s — the federation wedge
Leg 2 of the scale thesis, and the one the centralized incumbents structurally can't copy: the
graph is range-partitioned into shards (a sovereign Autobase log = one Shard), each superstep every
shard computes its OWNED nodes locally in parallel from a globally-exchanged rank halo, then the
disjoint owned ranges are gathered. Only the O(n) halo crosses shard boundaries — the O(E) edges
stay sovereign per participant. That's Pregel/BSP, but native to our federation.
- Shard { lo, hi, in_adj } + partition_edges() + distributed_pagerank()
- Deterministic (disjoint owned ranges, fixed source order); matches single-graph PageRank EXACTLY.
Measured (8-core, 2M nodes / 20M edges / 20 iters, examples/scale_bench.rs):
distributed 8 shards: 1.35s == single-graph (max|Δ| ~0) 16 MB halo/superstep vs 20M edges local
(vs 2.77s monolithic parallel — partition locality also helps single-node)
Neptune can't do this: their data is central. Ours is partitioned by construction, so the query
pushes down and only ranks are exchanged. 13 tests (incl. sharded==single-graph at k∈{1,2,3,5}).
mdheller
changed the title
Leg 3 (the #1 real gap vs centralized in-memory engines): the O(E) edge structure lives in a memory-mapped CSR file (paged by the OS), so only the O(n) rank vectors are heap-resident. PageRank runs DIRECTLY over the mapping — edges are read from disk-backed pages, never materialized into a heap Vec — so a graph LARGER than RAM is processable. - ooc.rs: write_csr (in-edge CSR: offsets u64 / in_neighbors u32 / out_deg u32, aligned) + MmapCsr (memmap2, zero-copy bytemuck views) + pagerank_mmap (rayon over the mmap, deterministic). Measured (2M nodes / 20M edges / 20 iters): CSR on disk: 104 MB (mmap'd) heap resident: ~32 MB (only O(n) rank vectors) pagerank_mmap: 1.086s — FASTER than the 2.77s in-heap parallel (CSR is cache-friendly vs Vec<Vec>) == in-memory PageRank exactly. 14 tests (incl. out_of_core == in_memory). This is single-machine; the file is the "disk", the OS page cache is the buffer pool. The last mile is a bounded-RAM streaming build (the writer still uses O(n+m) temp heap) — but the QUERY path is now RAM-ceiling-free.
mdheller
changed the title
…arger than RAM Completes the out-of-core leg's INGEST side. write_csr_streaming builds the mmap CSR from a re-iterable edge STREAM holding only O(n) heap (in_deg/offsets/out_deg/cursor) — it never materializes the O(E) edges. Two streaming passes: (1) count degrees → offsets; (2) random-write in-neighbours directly into the disk-backed (MmapMut) neighbours region. So you can now BUILD a >RAM graph, not just query one. Verified byte-identical to the batch write_csr, and pagerank_mmap over the streamed CSR == in-memory. 15 tests. Both ends of the RAM ceiling — build and query — are now bounded to O(n) heap.
examples/ooc_scale.rs streams 100M edges from a generator straight into the mmap CSR (never materializes the edge list) and runs PageRank over the disk-backed mapping. Measured (10M nodes / 100M edges, 8-core): in-heap edge Vec would be ~1600 MB; instead heap resident ~240 MB (O(n) only) CSR on disk: 520 MB stream-build 2.72s pagerank_mmap(10) 5.80s 172M edges·iter/s Σrank = 1.000000 (mass conserved) — correct at scale
…O ingest of >RAM graphs
write_csr_bucketed removes the random-write page-thrash of write_csr_streaming: destination nodes
are split into contiguous buckets and the neighbours region is emitted one bucket at a time in
order, so ALL output I/O is sequential. Peak heap is O(n) + O(m/num_buckets) — the per-bucket
buffer — independent of total edge count, so a truly larger-than-RAM neighbours array builds
cleanly (stream the edges once per bucket). Byte-identical to the batch write_csr at every bucket
count (b∈{1,2,3,5,8}). 16 tests.
examples/dist_socket.rs — the distributed model over an ACTUAL network transport, not shared memory. The coordinator partitions the graph, writes one shard file per participant, and spawns N worker PROCESSES. Each BSP superstep it broadcasts the rank halo over TCP; each worker computes its owned nodes from its LOCAL shard file (edges never leave the process) and returns its slice; the coordinator gathers. Only the O(n) halo crosses the wire. Measured (200k nodes / 2M edges / 4 worker processes, 20 supersteps): 68 ms over TCP 160 MB halo over the wire edges NEVER left their worker == single-graph PageRank: max|Δ| 8.47e-21 (EXACT) This closes the "single-machine simulation" asterisk on leg 2: distinct processes, real sockets, real serialization, sovereign per-process edges — and the answer is exact. The Hypercore/Autobase layer is the production transport; this proves the compute+exchange model over a real socket.
…el generalizes
connected_components (single-graph min-label propagation) + distributed_connected_components (BSP
over CcShards): each superstep every shard recomputes its owned labels in parallel from the
exchanged O(n) label halo; only labels cross shard boundaries, edges stay sovereign. Deterministic;
matches single-graph at any shard count (k∈{1,2,3,6}).
Proves the distributed wedge is NOT PageRank-specific — the same partition-native, only-the-halo-
crosses model works for the whole vertex-centric class. 17 tests.
examples/warm_scale.rs: after a +50-edge delta on a 20M-edge graph, recompute from the prior fixed point vs cold. Result is EXACT (max|Δ| 1.3e-15, same fixed point) but the speedup is a modest 1.3x — reported straight, NOT dressed up. A well-mixed random graph converges fast even cold, so warm-start helps only ~half the iterations here. The real incremental win is on slow-converging / large-diameter graphs and tighter tolerances; this benchmark is the honest floor, not the ceiling.
… edges examples/billions.rs uses the bucketed (sequential-I/O) builder so it never thrashes. Measured on a laptop: 50M nodes / 500M edges built + queried out-of-core with 1.2 GB heap (vs ~8 GB in-heap), 2.60 GB CSR on disk, 126M edges·iter/s, Σrank=1.0 (correct). Streaming (random-write) thrashes past ~100M edges once neighbours exceed page cache; bucketed completes (slow build, no thrash). 1B on this box times out — the ceiling is the machine's RAM/disk, not the code. Cluster is the next instrument; this proves the single-node out-of-core path to a half-billion edges.
Deterministic seeded splitmix64 RMAT stream (A=0.57/B=0.19/C=0.19/D=0.05), re-iterable for the two-pass CSR builders. scale -> 2^scale vertices, edgefactor edges/vertex. The standard, no-download dataset for the weekend cluster run.
The scaling unlock. Plain distributed_pagerank broadcasts the full O(n) rank vector to every shard each superstep (k*n). Boundary-only halo exchanges only the ranks of the remote vertices a shard's edges reference (its ghosts), so the recurring per-superstep cost is Sum|ghosts| << k*n. Bit-identical to serial pagerank (max|delta| < 1e-12), deterministic (sorted ghost order), and total_halo_bytes measures the real recurring network cost.
…adcast RMAT scale-16 (worst case, range partition): halo = 17% of full broadcast (5.9x less), max|delta| vs serial 8e-17. Ring 1M (perfect locality): 1 ghost/shard, ~1e6x less, exact. This is the honest floor the edge-cut partitioner improves on.
Deterministic streaming partitioners that place each vertex on the shard where its neighbours already live, minimizing edge cut and thus the boundary halo. partition_edges_boundary_at + relabel_contiguous let a smart partition drive the same boundary-halo PageRank on a non-uniform layout, bit-identical to serial. Measured on RMAT scale-16, k=16 (halo_bench): range : 85% cut, halo 1.43 MB (5.9x < full), balanced 1.0x fennel: 20% cut, halo 640 KB (13.1x < full), 3.0x imbalance ldg : 76% cut, halo 1.16 MB (7.2x < full), 1.1x balanced Fennel = 4x fewer crossing edges, 2.2x less recurring network vs range.
…zing Graph500 -> Fennel -> boundary-halo PageRank across scale 14..20, reporting the numbers that size the cluster before the spend: fattest-shard RAM, per-node halo bytes/superstep, total network/superstep, edge cut %, throughput. Measured (k=16, RMAT ef=16): scale 20 (1M nodes / 16.8M edges): cut 15.1%, fattest shard 132 MB, halo/node 3.1 MB, 0.76s/20 iters (442M e.it/s), Sigma-rank 1.000 exact. Edge cut IMPROVES with scale (23.3% -> 15.1%). Extrapolation: 125.8 bytes/edge -> 1B edges = 8 nodes @ 16GB / 4 @ 32GB; 10B = 20-79 nodes; 100B = 197-787 nodes. The Saturday plan, measured not guessed.
Mirrors the boundary-halo PageRank path for CC: only ghost LABELS (u32) cross shard boundaries, not the full O(n) label vector. Bit-identical to single-graph connected_components on a multi-component graph. Proves the boundary-halo model is not PageRank-specific but covers the vertex-centric class (PageRank + CC).
The actual cluster artifact, proven locally over real TCP first. 8 worker PROCESSES, Fennel-partitioned + relabelled graph. Workers KEEP their owned ranks across supersteps; only the boundary crosses the wire both ways (ghost halos down, boundary-owned values + dangling scalar up). NO O(n) message per step; one final O(n) gather collects the answer. Measured (scale-18, 4.2M edges, 8 procs, 25 iters): 122ms, |B|=49% of n, wire 64.5 MB total vs full-broadcast BSP 419 MB (6.5x less), max|delta| 8.6e-16 vs single-graph PageRank (EXACT). Edges never leave their worker file.
One image is every pod (HG_ROLE=coordinator|worker). dist_boundary refactored to ship shards over the socket (no shared filesystem) and take roles/graph size from env (HG_ROLE/HG_COORD/HG_ORDINAL, HG_SCALE/HG_EDGEFACTOR/HG_ITERS/HG_SHARDS); worker retries the coordinator connect so k8s pod-start order doesn't matter. Local self-contained run unchanged (still 8.6e-16 exact). deploy/bench: Dockerfile (Rust 1.96 multi-stage -> slim), k8s configmap + headless coordinator Service/Job + Indexed worker Job (JOB_COMPLETION_INDEX = ordinal), run.sh (build/push/apply/stream/teardown, keeps fan-out == HG_SHARDS), README runbook. Default = scale 26 / 8 shards (~1B edges, the MVP proof). Spin up -> work -> TEAR DOWN discipline; cluster create/delete stays explicit.
Same Graph500 graph, same machine, matched damping/tol. Rust emits the shared graph + our result; the Python script runs BOTH pure-Python networkx AND scipy's optimized sparse power iteration (no strawman) and reports speedup + top-100 agreement. Measured (8-core laptop): scale 15 (524K edges): rust 9.4ms | scipy 32ms (3.4x, 100% agree) | nx 271ms (29x) scale 17 (2.1M edges): rust 44ms | scipy 122ms (2.8x, 100% agree) | nx 1.28s (29x) ~3x faster than BLAS-optimized sparse single-node, 100% same ranking — AND we distribute (boundary-halo), which neither baseline can.
…tor relay) Removes the coordinator from the hot path: workers form a mesh and exchange ghost values DIRECTLY (worker c sends d exactly the owned values that are d's ghosts). The coordinator keeps only setup + a per-step SCALAR dangling all-reduce (k floats up, k down — O(k), a barrier) + the one-time final gather. Mesh is built deterministically (connect-higher / accept-lower, announce id), reader thread per peer drains into a channel to avoid TCP deadlock. Measured (scale-18, 4.2M edges): k=8 → 99.99% of recurring bytes peer-to-peer, coordinator only 128 B/step; k=16 mesh (120 conns) same. max|delta| 8.6e-16 EXACT, deterministic across runs. This is the unlock past ~tens of nodes: coordinator traffic is O(k), independent of graph size. Fixed en route: ghost-scatter wrote to local_rank[ghost_pos] instead of [owned+ghost_pos], freezing the halo at the uniform init (had shown 3.9e-3); now bit-exact.
distributed_bfs_boundary — hop distance from a source via Bellman-Ford-style relaxation over the undirected boundary CC shards; only ghost distances (u32) cross. This is the TRAVERSAL shape (frontier expansion, not fixpoint smoothing), so the boundary-halo model now covers the LDBC traversal class too, alongside PageRank + connected-components. Converges in O(diameter) supersteps, bit-exact vs serial BFS. Weighted SSSP is the same loop with +w(u,v) instead of +1. Distributed boundary-halo suite: PageRank, connected-components, BFS. 27 tests.
…card Runs the whole boundary-halo suite (PageRank / WCC / BFS = 3 LDBC Graphalytics kernels, 3 computational shapes) on ONE Fennel-partitioned Graph500 graph and prints a scorecard: per-kernel time, recurring halo bytes, and bit-exact verification vs single-graph. The Saturday deliverable in one command. Measured scale-20 (1M nodes / 16.8M edges, 16 shards): partition 0.86s; PR 1.04s (halo 7.6MB, max|Δ| 6e-17), WCC 0.24s, BFS 0.20s — ALL EXACT. BFS from 0 reached 645803/1048576 in max 5 hops.
…cker) saturday.sh: full ephemeral lifecycle — GKE create (spot) -> Cloud Build image -> run -> TEAR DOWN cluster on exit. --preflight validates auth/APIs/AR-repo/ manifests and prints the plan while spending nothing (caught the known gcloud re-auth blocker cleanly in a dry-run). run.sh: BUILDER=docker|cloudbuild (cloudbuild needs no local docker daemon). cloudbuild.yaml: server-side build honouring the non-root Dockerfile. Dockerfile now builds BOTH runtimes (dist_boundary relay = MVP default, dist_p2p mesh = scale path; p2p needs POD_IP->HG_ADVERTISE wiring). README updated with the one-command path + preflight.
…le capstone) Documents every measured number with the exact command to reproduce it: single- node vs networkx/scipy (~3x faster than sparse-BLAS, 100% agreement), boundary- halo scaling (Fennel 13x less wire), real multi-process TCP (relay 6.5x, P2P 99.99% peer-to-peer), the 3-kernel LDBC suite (all bit-exact), out-of-core 500M edges, and the ~126 B/edge cluster sizing. Honest-limits section included. Turns the scattered examples into one verifiable benchmark story.
distributed_sssp_boundary — weighted single-source shortest path via BSP Bellman-Ford relaxation over BoundaryWShard (adjacency carries edge weights); only ghost DISTANCES (f64) cross. Shortest-path distances are unique so the min-fixpoint is order-independent and EXACT vs serial Dijkstra (max|Δ| 0). BFS is the w≡1 special case. Wired into ldbc_suite → now 4/6 LDBC Graphalytics kernels (BFS, PR, WCC, SSSP), all bit-exact. Scale-20/16sh: SSSP 0.49s, halo 10.8MB, max|Δ| 0. BENCHMARKS.md updated. 28 tests.
…lo distributed distributed_cdlp_boundary — LDBC community detection by SYNCHRONOUS label propagation for a fixed iteration count: each round a vertex adopts the most frequent neighbour label, ties -> smallest id. Only ghost labels (u32) cross. The label-VOTING shape (distinct from WCC's min-propagation). Deterministic and bit-exact vs single-graph CDLP. Wired into ldbc_suite -> 5/6 LDBC Graphalytics kernels (BFS, PR, WCC, CDLP, SSSP), all bit-exact. Scale-20/16sh: CDLP 3.90s (per-node label histogram x 10 rounds), rest sub-second. Only LCC (needs 2-hop halo) remains. 29 tests.
…mplete distributed_lcc_boundary — local clustering coefficient with a 2-HOP boundary halo: ghosts carry their adjacency (not a scalar) so each owned vertex can count edges among its neighbours. Single-pass, so the heavier halo is exchanged once. LCC(v) = Sum|N(a) cap N(v)| / (deg*(deg-1)). Bit-exact vs single-graph (max|D| 0). Wired into ldbc_suite -> the FULL six-kernel LDBC Graphalytics suite (BFS, PR, WCC, CDLP, LCC, SSSP), boundary-halo distributed, ALL verified bit-exact. Scale-18/16sh: 5 kernels sub-second-to-~1s; LCC 21s (triangle counting on RMAT power-law hubs is adversarial — inherent, single-pass, exact). 30 tests.
vs_kuzu.py — the comparison that usually 'needs a cluster', run in-process: KuzuDB (embedded graph DB, Cypher + native page_rank) on the SAME RMAT graph, same machine, compute-to-compute. Measured: scale-17 (2.1M edges) hg 46ms vs Kuzu 559ms = 12.2x faster; scale-18 (4.2M edges) hg 109ms vs Kuzu 1122ms = 10.3x faster; top-100 ranking 100% identical both. Kuzu also pays a ~0.5s bulk-load we don't. Honest caveats: Kuzu is single-machine embedded, default damping/iters may differ (but ranking agrees 100%). Closes the biggest open gap — a real graph-DB head-to-head, not just linear-algebra libraries. BENCHMARKS.md updated.
The Kuzu head-to-head exposed a real gap: our single-machine WCC used label propagation (chosen for the distributed BSP path) and LOST to Kuzu's union-find (117ms vs 98ms). Added connected_components_uf (union-find, path halving + union by size, near-linear) — induces the same partition as the label-prop reference, verified by test. Cut our WCC to 13ms. Result: hg_analytics now beats KuzuDB (a real embedded graph DB) on BOTH kernels at scale-18/4.2M edges: PageRank 10.1x (ranking 100% agree), WCC 7.6x (same 88104 components). vs_baseline emits WCC timing; vs_kuzu compares both. 31 tests.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
"Rust + our architecture outscale the incumbents" — turned into measured Rust, not assertion. Reproduce:
cargo run --release --example {scale_bench,ooc_scale,dist_socket}. 16 tests, deterministic, all match the serial reference.1 — Parallel analytics (language + cores)
pagerank_parallel(pull, rayon) ·betweenness_parallel(deterministic via fixed-chunk reduction).2 — Distributed PageRank — the wedge Neptune can't copy
Shard/partition_edges/distributed_pagerank(BSP): partition = participant, each superstep only the O(n) rank halo crosses shard boundaries; edges stay sovereign.dist_socket.rs): 4 worker processes over actual TCP — 200k/2M, 20 supersteps in 68ms, 160 MB halo over the wire, edges never left their process, answer EXACT (Δ 8e-21). No longer a simulation.3 — Out-of-core (kill the RAM ceiling), both ends
MmapCsr/pagerank_mmap— O(E) edges mmap'd from disk, only O(n) heap.write_csr_streaming(O(n) heap) +write_csr_bucketed(fully sequential I/O, no page-thrash).ooc_scale.rs): 100M edges — heap ~240 MB (vs ~1.6 GB in-heap), 520 MB CSR on disk, 172M edges·iter/s, mass conserved.Honest limits (straight)
dist_socketis loopback TCP on one box — real serialization + processes, but not a WAN; the Hypercore/Autobase layer is the production transport.🤖 Generated with Claude Code