From 5b4d35773248211ff4f1fbab8d41e4d4f2518f34 Mon Sep 17 00:00:00 2001 From: Alexy Khrabrov Date: Sun, 20 Sep 2026 01:26:15 +0000 Subject: [PATCH 1/8] grust-procedures: a work meter that batches, and a concurrency setting charge_work admits through a compare-exchange on one counter, which is right single-threaded and is the shape that degrades when N workers retry against the same cache line once per visited entry. WorkMeter admits a block at a time and spends it locally; when a block does not fit it admits exactly what was asked for, so a budget still fails at the unit that exceeds it, and unspent units return on drop so final usage counts work performed. Cancellation is still observed per charge; the deadline samples per block, the same cadence per worker as before. reserve_for_workers admits per-worker scratch once, before a parallel region, keeping the memory mutex off the per-element path. Concurrency is set on the context rather than added to ExecutionLimits: the limits struct is built by literal in two dozen places across the workspace, and a setter that refuses to run once the context is shared gives the same per-execution semantics with a default of one thread. Parallel counterparts of the budget contracts cover a budget that exactly fits, one a unit short, cancellation reaching every worker and an expired deadline, at sixteen threads. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Sak2FQcdgs5pkL2ruKrSUZ --- crates/grust-procedures/src/lib.rs | 2 +- crates/grust-procedures/src/resources.rs | 169 +++++++++++++++ crates/grust-procedures/tests/contracts.rs | 2 + .../tests/contracts/parallel.rs | 199 ++++++++++++++++++ 4 files changed, 371 insertions(+), 1 deletion(-) create mode 100644 crates/grust-procedures/tests/contracts/parallel.rs diff --git a/crates/grust-procedures/src/lib.rs b/crates/grust-procedures/src/lib.rs index 2cc9e355..9c0c1724 100644 --- a/crates/grust-procedures/src/lib.rs +++ b/crates/grust-procedures/src/lib.rs @@ -25,7 +25,7 @@ pub use registry::{ }; pub use resources::{ Cancellation, ExecutionContext, ExecutionLimits, MemoryAccount, MemoryReservation, - ResourceUsage, + ResourceUsage, WORK_BLOCK_UNITS, WorkMeter, }; pub use signature::{ Argument, Correlation, Determinism, Field, GraphRequirement, OptionField, ProcedureDefinition, diff --git a/crates/grust-procedures/src/resources.rs b/crates/grust-procedures/src/resources.rs index d3470923..28862f90 100644 --- a/crates/grust-procedures/src/resources.rs +++ b/crates/grust-procedures/src/resources.rs @@ -53,12 +53,20 @@ const DEADLINE_SAMPLE_UNITS: usize = 1024; #[derive(Debug)] struct Shared { limits: ExecutionLimits, + concurrency: usize, work_units: AtomicUsize, cancelled: AtomicBool, charges_since_deadline_read: AtomicUsize, state: Mutex, } +/// Units a [`WorkMeter`] admits in one shared-counter operation while the budget +/// has room for a whole block. Kernels charge once per visited entry, so at N +/// workers the shared counter would otherwise be a contended cache line rather +/// than an occasional one. Near the limit the meter drops to exact admission, +/// so a budget still fails at exactly the unit that exceeds it. +pub const WORK_BLOCK_UNITS: usize = 1024; + /// Whether a state check must read the clock or may rely on the sampled read. #[derive(Clone, Copy)] enum DeadlineCheck { @@ -86,6 +94,7 @@ impl ExecutionContext { } Ok(Self(Arc::new(Shared { limits, + concurrency: 1, work_units: AtomicUsize::new(0), cancelled: AtomicBool::new(false), charges_since_deadline_read: AtomicUsize::new(0), @@ -93,6 +102,36 @@ impl ExecutionContext { }))) } + /// Permit kernels to use up to `workers` threads for this execution. + /// + /// Concurrency is a property of one execution, not of the process: an + /// embedder that runs kernels inside a server with its own runtime and + /// thread pool decides here how many threads a query may add, and the + /// default of one keeps existing callers single-threaded. Set it before + /// the context is shared, so no worker can observe it changing. + /// + /// # Errors + /// Rejects zero workers, and a context that is already shared. + pub fn with_concurrency(mut self, workers: usize) -> Result { + if workers == 0 { + return Err(ProcedureError::InvalidArguments( + "concurrency must be at least one".into(), + )); + } + let shared = Arc::get_mut(&mut self.0).ok_or_else(|| { + ProcedureError::InvalidArguments( + "concurrency must be set before the context is shared".into(), + ) + })?; + shared.concurrency = workers; + Ok(self) + } + + /// Threads kernels may use for this execution; one unless set. + pub fn concurrency(&self) -> usize { + self.0.concurrency + } + /// Read immutable limits. pub fn limits(&self) -> ExecutionLimits { self.0.limits @@ -122,9 +161,33 @@ impl ExecutionContext { self.check_state(DeadlineCheck::Exact) } + /// A batched work meter for one worker inside a parallel region. + /// + /// The meter admits [`WORK_BLOCK_UNITS`] at a time from the shared counter + /// and spends them locally, so N workers do not contend on one cache line + /// per visited entry. Admission stays exact: when a block does not fit, the + /// meter admits precisely the units asked for, so the budget fails at the + /// same unit it would fail at single-threaded. Unspent units are returned + /// when the meter drops, so final usage counts work actually performed; + /// [`Self::usage`] read while meters are live can exceed it by less than + /// one block per worker. + /// + /// Cancellation is still observed per charge. The deadline is sampled per + /// block, which is the same cadence per worker as a single-threaded charge. + pub fn work_meter(&self) -> WorkMeter { + WorkMeter { + context: self.clone(), + granted: 0, + } + } + /// Charge work before performing it. Counter overflow is a budget failure. pub fn charge_work(&self, units: usize) -> Result<()> { self.check_state(DeadlineCheck::Sampled)?; + self.admit_work(units) + } + + fn admit_work(&self, units: usize) -> Result<()> { // Admit exactly, never overshooting the budget: the compare-exchange // recomputes admission against the value it actually replaces, so a // concurrent charge cannot slip past the limit between load and store. @@ -149,6 +212,21 @@ impl ExecutionContext { } } + /// Return admitted units that were never spent. Only a meter calls this, + /// and only with units it admitted, so the counter cannot underflow. + fn refund_work(&self, units: usize) { + if units != 0 { + self.0.work_units.fetch_sub(units, Ordering::Relaxed); + } + } + + fn check_cancelled(&self) -> Result<()> { + if self.0.cancelled.load(Ordering::Acquire) { + return Err(ProcedureError::Cancelled); + } + Ok(()) + } + fn check_state(&self, deadline: DeadlineCheck) -> Result<()> { if self.0.cancelled.load(Ordering::Acquire) { return Err(ProcedureError::Cancelled); @@ -186,6 +264,30 @@ impl ExecutionContext { }))) } + /// Reserve one worker's scratch times the number of workers, before the + /// parallel region starts. + /// + /// Memory admission takes a lock, which single-threaded is a constant + /// overhead and under threads would be a serialization point. Kernels that + /// need per-worker scratch admit all of it here, once, so admission is + /// exact and does not depend on how work happens to be scheduled. + /// + /// # Errors + /// Reports a memory budget the total would exceed, including on overflow. + pub fn reserve_for_workers( + &self, + workers: usize, + bytes_each: usize, + ) -> Result { + let bytes = workers + .checked_mul(bytes_each) + .ok_or(ProcedureError::BudgetExceeded { + resource: "memory", + limit: self.0.limits.memory_bytes, + })?; + self.reserve(bytes) + } + /// Check admission without reserving or changing measured peak usage. /// The actual owner must still reserve before allocating. pub fn check_memory_available(&self, bytes: usize) -> Result<()> { @@ -263,6 +365,73 @@ impl ExecutionContext { } } +/// One worker's share of the work budget, admitted in blocks. +/// +/// Created by [`ExecutionContext::work_meter`], used by one thread, and dropped +/// at the end of that thread's chunk. Charge before performing the work, as with +/// [`ExecutionContext::charge_work`]. +pub struct WorkMeter { + context: ExecutionContext, + granted: usize, +} + +impl WorkMeter { + /// Charge work this worker is about to perform. + /// + /// # Errors + /// Reports cancellation, an expired deadline, and a work budget that the + /// charge would exceed. + pub fn charge(&mut self, units: usize) -> Result<()> { + self.context.check_cancelled()?; + if let Some(left) = self.granted.checked_sub(units) { + self.granted = left; + return Ok(()); + } + self.admit(units) + } + + /// Admit a block, or exactly what was asked for when a block does not fit. + #[cold] + fn admit(&mut self, units: usize) -> Result<()> { + self.context.check_state(DeadlineCheck::Sampled)?; + let wanted = units.saturating_add(WORK_BLOCK_UNITS); + if wanted > units && self.context.admit_work(wanted - self.granted).is_ok() { + self.granted = wanted - units; + return Ok(()); + } + // Near the limit, spend the remaining budget one charge at a time so + // that the failure happens at the same unit as it would at one thread. + self.context.admit_work(units - self.granted)?; + self.granted = 0; + Ok(()) + } + + /// Shared execution state, for an exact poll or a nested reservation. + pub fn execution(&self) -> &ExecutionContext { + &self.context + } + + /// Poll cancellation and the deadline exactly, as [`ExecutionContext::checkpoint`]. + /// + /// # Errors + /// Reports cancellation and an expired deadline. + pub fn checkpoint(&self) -> Result<()> { + self.context.checkpoint() + } + + /// Return unspent units now rather than at drop. + pub fn finish(self) { + drop(self); + } +} + +impl Drop for WorkMeter { + fn drop(&mut self) { + self.context.refund_work(self.granted); + self.granted = 0; + } +} + /// An exclusive owner of temporary allocation admission. Unlike a reservation, /// it can grow without allocating a token per scalar or intermediate row. pub struct MemoryAccount { diff --git a/crates/grust-procedures/tests/contracts.rs b/crates/grust-procedures/tests/contracts.rs index 36d8da0a..0f999e87 100644 --- a/crates/grust-procedures/tests/contracts.rs +++ b/crates/grust-procedures/tests/contracts.rs @@ -12,6 +12,8 @@ mod builtins; mod cache; #[path = "contracts/cancellation.rs"] mod cancellation; +#[path = "contracts/parallel.rs"] +mod parallel; #[test] fn temporary_memory_accounts_release_only_their_own_admission() { diff --git a/crates/grust-procedures/tests/contracts/parallel.rs b/crates/grust-procedures/tests/contracts/parallel.rs new file mode 100644 index 00000000..b63bb3d1 --- /dev/null +++ b/crates/grust-procedures/tests/contracts/parallel.rs @@ -0,0 +1,199 @@ +//! The budget contracts at more than one thread. The single-threaded versions +//! live in `failures.rs`; these assert that batching the work counter across +//! workers keeps them: a budget that fits succeeds, a budget one unit short +//! fails, cancellation reaches every worker, and an expired deadline stops the +//! region. Each test uses plain threads, so the accounting layer stays free of +//! a runtime or a thread pool. + +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; + +use super::*; + +const WORKERS: usize = 16; +const PER_WORKER: usize = 10_000; + +fn limits(work_units: usize, deadline: Option) -> ExecutionLimits { + ExecutionLimits { + memory_bytes: 1 << 20, + work_units, + batch_rows: 8, + deadline, + } +} + +/// Charge `per_worker` units on each of `WORKERS` threads, returning how many +/// workers failed and the usage the context reports afterwards. +fn charge_everywhere(execution: &ExecutionContext, per_worker: usize) -> (usize, ResourceUsage) { + let failures = Arc::new(AtomicUsize::new(0)); + std::thread::scope(|scope| { + for _ in 0..WORKERS { + let execution = execution.clone(); + let failures = Arc::clone(&failures); + scope.spawn(move || { + let mut meter = execution.work_meter(); + for _ in 0..per_worker { + if meter.charge(1).is_err() { + failures.fetch_add(1, Ordering::Relaxed); + return; + } + } + }); + } + }); + ( + failures.load(Ordering::Relaxed), + execution.usage().expect("usage"), + ) +} + +#[test] +fn a_budget_that_exactly_fits_succeeds_and_counts_the_work_performed() { + let total = WORKERS * PER_WORKER; + let execution = ExecutionContext::new(limits(total, None)).expect("valid limits"); + let (failures, usage) = charge_everywhere(&execution, PER_WORKER); + assert_eq!(failures, 0, "no worker may fail inside its budget"); + // Unspent blocks are returned when each meter drops, so the final count is + // the work actually charged, not the blocks admitted to charge it. + assert_eq!(usage.work_units, total); +} + +#[test] +fn a_budget_one_unit_short_fails_at_one_worker() { + let total = WORKERS * PER_WORKER; + let execution = ExecutionContext::new(limits(total - 1, None)).expect("valid limits"); + let (failures, usage) = charge_everywhere(&execution, PER_WORKER); + assert_eq!(failures, 1, "exactly the worker that ran out should fail"); + assert!( + usage.work_units <= total - 1, + "spent {} of a {} unit budget", + usage.work_units, + total - 1 + ); +} + +#[test] +fn every_worker_observes_cancellation() { + let execution = ExecutionContext::new(limits(usize::MAX, None)).expect("valid limits"); + let started = Arc::new(AtomicUsize::new(0)); + let cancelled = Arc::new(AtomicUsize::new(0)); + std::thread::scope(|scope| { + for _ in 0..WORKERS { + let execution = execution.clone(); + let started = Arc::clone(&started); + let cancelled = Arc::clone(&cancelled); + scope.spawn(move || { + let mut meter = execution.work_meter(); + started.fetch_add(1, Ordering::Relaxed); + loop { + match meter.charge(1) { + Ok(()) => continue, + Err(ProcedureError::Cancelled) => { + cancelled.fetch_add(1, Ordering::Relaxed); + return; + } + Err(other) => panic!("unexpected failure: {other}"), + } + } + }); + } + while started.load(Ordering::Relaxed) < WORKERS { + std::hint::spin_loop(); + } + execution.cancel().expect("cancel"); + }); + assert_eq!(cancelled.load(Ordering::Relaxed), WORKERS); +} + +#[test] +fn an_expired_deadline_stops_every_worker() { + let execution = ExecutionContext::new(limits( + usize::MAX, + Some(Instant::now() + Duration::from_millis(50)), + )) + .expect("valid limits"); + let expired = Arc::new(AtomicUsize::new(0)); + std::thread::scope(|scope| { + for _ in 0..WORKERS { + let execution = execution.clone(); + let expired = Arc::clone(&expired); + scope.spawn(move || { + let mut meter = execution.work_meter(); + loop { + match meter.charge(1) { + Ok(()) => continue, + Err(ProcedureError::DeadlineExceeded) => { + expired.fetch_add(1, Ordering::Relaxed); + return; + } + Err(other) => panic!("unexpected failure: {other}"), + } + } + }); + } + }); + assert_eq!(expired.load(Ordering::Relaxed), WORKERS); +} + +#[test] +fn a_dropped_meter_returns_what_it_did_not_spend() { + let execution = + ExecutionContext::new(limits(WORK_BLOCK_UNITS * 4, None)).expect("valid limits"); + { + let mut meter = execution.work_meter(); + meter.charge(1).expect("first charge admits a block"); + assert!( + execution.usage().expect("usage").work_units > 1, + "a live meter holds more than it has spent" + ); + } + assert_eq!( + execution.usage().expect("usage").work_units, + 1, + "dropping the meter leaves only the work performed" + ); + // The whole budget remains spendable afterwards, one meter at a time. + for _ in 0..4 { + let mut meter = execution.work_meter(); + for _ in 0..WORK_BLOCK_UNITS / 2 { + meter.charge(1).expect("within budget"); + } + } + assert_eq!( + execution.usage().expect("usage").work_units, + 1 + 4 * (WORK_BLOCK_UNITS / 2) + ); +} + +#[test] +fn per_worker_scratch_is_admitted_before_the_region() { + let execution = ExecutionContext::new(limits(64, None)).expect("valid limits"); + let scratch = execution + .reserve_for_workers(WORKERS, 1024) + .expect("fits the memory budget"); + assert_eq!(scratch.bytes(), WORKERS * 1024); + assert!( + execution.reserve_for_workers(WORKERS, usize::MAX).is_err(), + "an overflowing total is a budget failure, not a wrap" + ); + assert!( + execution.reserve_for_workers(1 << 20, 1 << 20).is_err(), + "a total beyond the memory budget is refused" + ); +} + +#[test] +fn concurrency_is_one_unless_set_and_cannot_change_once_shared() { + let execution = ExecutionContext::new(limits(8, None)).expect("valid limits"); + assert_eq!(execution.concurrency(), 1); + let execution = execution.with_concurrency(8).expect("not yet shared"); + assert_eq!(execution.concurrency(), 8); + assert!(execution.clone().with_concurrency(2).is_err()); + assert!( + ExecutionContext::new(limits(8, None)) + .expect("valid limits") + .with_concurrency(0) + .is_err() + ); +} From 1929f6040d831669f4285047b5e77ffab76531eb Mon Sep 17 00:00:00 2001 From: Alexy Khrabrov Date: Sun, 20 Sep 2026 01:38:48 +0000 Subject: [PATCH 2/8] grust-algorithms: rayon in degree, PageRank, breadth-first search and components The pool is bounded by the execution, never by the machine: an execution that asks for threads gets them from a pool sized to that request, and one that asks for nothing runs exactly the code it ran before. Nothing reads the CPU count, because these kernels also run inside a server that owns its own runtime and pool. Results do not move with the thread count. Reductions combine per-chunk partials in chunk order rather than completion order, so sums are bit-for-bit equal at one worker and sixteen. Degree gives every node its own output slot. PageRank pulls into each target over the reverse topology instead of pushing out of each source, which is what removes the write conflicts; the weighted projection keeps the push kernel, which needs the source's weight total per arc. Breadth-first search claims each node with one compare-exchange per level, so a node enters the next frontier once and its distance is the level that claimed it. Components link the larger root to the smaller, so the surviving root is the component's minimum row: the same canonical label the sequential kernel produces. Work accounting goes through per-worker meters, and degree's charges are unchanged unit for unit, so work-unit comparisons against earlier runs still mean something. tests/parallel.rs is the differential oracle on a 120k-node graph: the same answers as the sequential paths, exactly for integers and identifiers and within a stated tolerance where the parallel path sums floats in a different order, and identical to itself at two and sixteen threads. examples/scaling.rs times the kernels on a real edge list, separating the sequential baseline from the parallel implementation at one worker, so a change of algorithm cannot be reported as a speedup from threads. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Sak2FQcdgs5pkL2ruKrSUZ --- crates/grust-algorithms/Cargo.toml | 4 +- crates/grust-algorithms/examples/scaling.rs | 188 ++++++++++++++++ crates/grust-algorithms/src/buffer.rs | 30 +++ crates/grust-algorithms/src/degree.rs | 86 +++++--- crates/grust-algorithms/src/pagerank.rs | 129 ++++++++++- crates/grust-algorithms/src/parallel.rs | 229 +++++++++++++++++++- crates/grust-algorithms/src/traversal.rs | 205 +++++++++++++++++- crates/grust-algorithms/tests/parallel.rs | 155 +++++++++++++ crates/grust-procedures/src/resources.rs | 20 +- 9 files changed, 990 insertions(+), 56 deletions(-) create mode 100644 crates/grust-algorithms/examples/scaling.rs create mode 100644 crates/grust-algorithms/tests/parallel.rs diff --git a/crates/grust-algorithms/Cargo.toml b/crates/grust-algorithms/Cargo.toml index f0875d64..7a538f67 100644 --- a/crates/grust-algorithms/Cargo.toml +++ b/crates/grust-algorithms/Cargo.toml @@ -12,7 +12,9 @@ categories.workspace = true [features] default = ["parallel"] arrow = ["dep:arrow-array", "dep:grust-arrow"] -# Run kernels on the caller's rayon pool. Results are identical at any width. +# Threads for the kernels, bounded by what each execution asks for. Results are +# identical at any width. Off compiles only the sequential code, which is also +# the differential oracle the parallel paths are tested against. parallel = ["dep:rayon"] [dependencies] diff --git a/crates/grust-algorithms/examples/scaling.rs b/crates/grust-algorithms/examples/scaling.rs new file mode 100644 index 00000000..a720fae2 --- /dev/null +++ b/crates/grust-algorithms/examples/scaling.rs @@ -0,0 +1,188 @@ +//! Time the parallel kernels against their sequential paths on a real graph. +//! +//! scaling [--workers 1,2,4,8,16] [--iterations 10] +//! [--undirected] [--kernels degree,pagerank,bfs,wcc] +//! +//! The edge list is one `source target` pair per line, `#` comments ignored, +//! as SNAP and the strain harness datasets ship them. The same projection is +//! built once per worker count, because concurrency belongs to the execution +//! that owns the projection; build time is reported separately from kernel +//! time, so a kernel's scaling is not diluted by a sequential load. +//! +//! Every row prints the result the kernel produced (a component count, a +//! reachable count, a score checksum) so that a speedup cannot come from doing +//! less work: the values must match across worker counts. +use std::collections::HashMap; +use std::time::Instant; + +use grust_algorithms::{ + ExecutionContext, ExecutionLimits, GraphProjection, Orientation, PageRankOptions, + ProjectionEdge, SnapshotIdentity, bfs, degree, pagerank, weakly_connected_components, +}; + +fn main() -> Result<(), Box> { + let mut args = std::env::args().skip(1); + let path = args.next().ok_or("usage: scaling [options]")?; + // Zero means the sequential kernels: an execution that never asks for + // threads, which is what an embedder gets by default. + let mut workers = vec![0usize, 1, 2, 4, 8, 16]; + let mut iterations = 10usize; + let mut orientation = Orientation::Outgoing; + let mut kernels = vec![ + "degree".to_string(), + "pagerank".to_string(), + "bfs".to_string(), + "wcc".to_string(), + ]; + while let Some(flag) = args.next() { + let mut value = || args.next().ok_or_else(|| format!("{flag} needs a value")); + match flag.as_str() { + "--workers" => { + workers = value()? + .split(',') + .map(|part| part.parse::()) + .collect::>()?; + } + "--iterations" => iterations = value()?.parse()?, + "--undirected" => orientation = Orientation::Undirected, + "--kernels" => kernels = value()?.split(',').map(String::from).collect(), + other => return Err(format!("unknown argument `{other}`").into()), + } + } + + let started = Instant::now(); + let text = std::fs::read_to_string(&path)?; + // Rows are numbered on first appearance, as the strain harness does, so the + // node table is the graph's own order rather than a sorted one. + let mut rows: HashMap<&str, usize> = HashMap::new(); + let mut names: Vec<&str> = Vec::new(); + let mut pairs: Vec<(usize, usize)> = Vec::new(); + for line in text.lines() { + if line.starts_with('#') || line.trim().is_empty() { + continue; + } + let mut fields = line.split_whitespace(); + let (Some(source), Some(target)) = (fields.next(), fields.next()) else { + continue; + }; + let source = match rows.get(source) { + Some(&row) => row, + None => { + let row = names.len(); + names.push(source); + rows.insert(source, row); + row + } + }; + let target = match rows.get(target) { + Some(&row) => row, + None => { + let row = names.len(); + names.push(target); + rows.insert(target, row); + row + } + }; + pairs.push((source, target)); + } + println!( + "{path}: {} nodes, {} edges, parsed in {:.1}s", + names.len(), + pairs.len(), + started.elapsed().as_secs_f64() + ); + + println!( + "{:<10} {:>10} {:>10} {:>10} {:>9} result", + "kernel", "workers", "build s", "kernel s", "vs seq" + ); + let mut baseline: HashMap = HashMap::new(); + for &threads in &workers { + let context = ExecutionContext::new(ExecutionLimits { + memory_bytes: 18 << 30, + work_units: usize::MAX, + batch_rows: 8192, + deadline: None, + })?; + let context = if threads == 0 { + context + } else { + context.with_concurrency(threads)? + }; + let build = Instant::now(); + let graph = GraphProjection::from_topology( + SnapshotIdentity::new("scaling".into(), "r1".into(), "bench".into())?, + names.iter().map(|name| (*name).into()).collect(), + pairs + .iter() + .enumerate() + .map(|(ordinal, &(source, target))| ProjectionEdge { + source, + target, + ordinal, + id: None, + }) + .collect(), + None, + orientation, + &context, + )?; + let build = build.elapsed().as_secs_f64(); + for kernel in &kernels { + let started = Instant::now(); + let result = match kernel.as_str() { + "degree" => { + let degrees = degree(&graph)?; + format!("arcs {}", degrees.counts().iter().sum::()) + } + "pagerank" => { + let ranks = pagerank( + &graph, + PageRankOptions { + damping: 0.85, + tolerance: 0.0, + max_iterations: iterations, + personalization: None, + }, + )?; + format!( + "iterations {} top {:.9}", + ranks.iterations(), + ranks.values().iter().copied().fold(0.0f64, f64::max) + ) + } + "bfs" => { + let distances = bfs(&graph, names[0])?; + let reached = distances.values().iter().filter(|d| d.is_finite()).count(); + let sum: f64 = distances.values().iter().filter(|d| d.is_finite()).sum(); + format!("reached {reached} sum {sum}") + } + "wcc" => { + let components = weakly_connected_components(&graph)?; + let mut labels: Vec = components.values().to_vec(); + labels.sort_unstable(); + labels.dedup(); + format!("components {}", labels.len()) + } + other => return Err(format!("unknown kernel `{other}`").into()), + }; + let seconds = started.elapsed().as_secs_f64(); + let speedup = match baseline.get(kernel) { + Some(&first) => format!("{:.2}x", first / seconds), + None => { + baseline.insert(kernel.clone(), seconds); + "-".to_string() + } + }; + let label = if threads == 0 { + "sequential".to_string() + } else { + threads.to_string() + }; + println!( + "{kernel:<10} {label:>10} {build:>10.1} {seconds:>10.3} {speedup:>9} {result}" + ); + } + } + Ok(()) +} diff --git a/crates/grust-algorithms/src/buffer.rs b/crates/grust-algorithms/src/buffer.rs index a46a719a..9391fed1 100644 --- a/crates/grust-algorithms/src/buffer.rs +++ b/crates/grust-algorithms/src/buffer.rs @@ -39,6 +39,36 @@ impl Buffer { }) } + /// A buffer the caller will fill by index, initialized without charging + /// work, for kernels that charge the same elements in their own pass. + /// + /// [`Self::filled`] charges initialization because its caller does not + /// otherwise visit the elements. A kernel that writes every element and + /// charges as it goes would charge twice, which would change what a work + /// budget means for no change in the work performed. + pub(crate) fn indexed(count: usize, value: T, context: &ExecutionContext) -> Result + where + T: Clone, + { + let mut buffer = Self::capacity(count, context)?; + context.checkpoint()?; + buffer.values.resize(count, value); + Ok(buffer) + } + + /// As [`Self::indexed`], for elements that cannot be cloned, such as the + /// atomics a parallel kernel claims entries in. + pub(crate) fn indexed_with( + count: usize, + make: impl FnMut() -> T, + context: &ExecutionContext, + ) -> Result { + let mut buffer = Self::capacity(count, context)?; + context.checkpoint()?; + buffer.values.resize_with(count, make); + Ok(buffer) + } + pub(crate) fn filled(count: usize, value: T, context: &ExecutionContext) -> Result where T: Clone, diff --git a/crates/grust-algorithms/src/degree.rs b/crates/grust-algorithms/src/degree.rs index 55b45b09..901f7d00 100644 --- a/crates/grust-algorithms/src/degree.rs +++ b/crates/grust-algorithms/src/degree.rs @@ -36,51 +36,69 @@ impl Degrees { /// execution takes O(V + A) work and sums weights in stable CSR order. Projection /// weights must already be finite and nonnegative; a nonfinite sum is a numerical /// error. No reverse index, graph copy, or implicit normalization is built. +/// Each node's count and strength depend only on its own arc range, so workers +/// write disjoint slices of the output and need no synchronization. Weighted +/// sums stay in CSR order within a node, which is why the result does not depend +/// on how the nodes were divided. pub fn degree(graph: &GraphProjection) -> Result { let context = graph.execution(); context.checkpoint()?; let adjacency = graph.outgoing(); - let mut counts = Buffer::capacity(graph.node_count(), context)?; + let n = graph.node_count(); + let mut counts = Buffer::indexed(n, 0usize, context)?; let mut strengths = adjacency .weights .as_ref() - .map(|_| Buffer::capacity(graph.node_count(), context)) + .map(|_| Buffer::indexed(n, 0.0f64, context)) .transpose()?; - // Precharge bounded chunks, as Buffer initialization does, instead of - // locking the execution context for every scalar. Arc charges span node - // boundaries so low-degree graphs also amortize the accounting lock. - let mut arcs_charged_until = 0; - for first in (0..graph.node_count()).step_by(1024) { - let last = first.saturating_add(1024).min(graph.node_count()); - context.charge_work(last - first)?; - for node in first..last { - let start = adjacency.offsets.values[node]; - let end = adjacency.offsets.values[node + 1]; - counts.values.push(end - start); - if let (Some(weights), Some(strengths)) = (&adjacency.weights, &mut strengths) { - let mut sum = 0.0; - let mut position = start; - while position < end { - if position == arcs_charged_until { - arcs_charged_until = - position.saturating_add(1024).min(weights.values.len()); - context.charge_work(arcs_charged_until - position)?; - } - let stop = end.min(arcs_charged_until); - for &weight in &weights.values[position..stop] { + let arcs = adjacency.targets.values.len(); + let workers = crate::parallel::workers(context, n.saturating_add(arcs)).unwrap_or(1); + let offsets = &adjacency.offsets.values; + let weights = adjacency.weights.as_ref().map(|weights| &weights.values); + + // Counts first: O(V) from the CSR offsets alone. + crate::parallel::for_each_chunk( + context, + workers, + &mut counts.values, + |first, slice, meter| { + meter.charge(slice.len())?; + for (index, count) in slice.iter_mut().enumerate() { + let node = first + index; + *count = offsets[node + 1] - offsets[node]; + } + Ok(()) + }, + )?; + + if let (Some(weights), Some(strengths)) = (weights, &mut strengths) { + crate::parallel::for_each_chunk( + context, + workers, + &mut strengths.values, + |first, slice, meter| { + for (index, strength) in slice.iter_mut().enumerate() { + let node = first + index; + let (start, end) = (offsets[node], offsets[node + 1]); + // The counts pass already charged this node; charge its arcs + // before summing them, so a high-degree node cannot run + // unmetered and cancellation stays prompt. + meter.charge(end - start)?; + let mut sum = 0.0; + for &weight in &weights[start..end] { sum += weight; } - position = stop; - } - // Nonnegative finite inputs cannot recover after overflowing. - if !sum.is_finite() { - return Err(AlgorithmError::Numerical( - "weighted degree exceeds finite Float64 domain".into(), - )); + // Nonnegative finite inputs cannot recover after overflowing. + if !sum.is_finite() { + return Err(AlgorithmError::Numerical( + "weighted degree exceeds finite Float64 domain".into(), + )); + } + *strength = sum; } - strengths.values.push(sum); - } - } + Ok(()) + }, + )?; } context.checkpoint()?; Ok(Degrees { diff --git a/crates/grust-algorithms/src/pagerank.rs b/crates/grust-algorithms/src/pagerank.rs index 169f5a6d..59567441 100644 --- a/crates/grust-algorithms/src/pagerank.rs +++ b/crates/grust-algorithms/src/pagerank.rs @@ -122,8 +122,25 @@ pub fn pagerank(graph: &GraphProjection, options: PageRankOptions<'_>) -> Result converged: true, }); } - let mut next = Buffer::filled(n, 0.0, context)?; let adjacency = graph.outgoing(); + // An unweighted projection has a probability of 1/out-degree on every arc, + // so the iteration can be computed by pulling into each target instead of + // pushing out of each source. Pulling is what makes it parallel: targets own + // their own sums, so no worker writes where another might read, and each sum + // stays in reverse-CSR order, so the result does not depend on the division + // of work. A weighted projection needs the source's weight total per arc and + // keeps the sequential push below. + let workers = crate::parallel::workers( + context, + n.saturating_add(adjacency.targets.values.len()) + .saturating_mul(2), + ); + if let Some(workers) = workers + && !graph.is_weighted() + { + return pull(graph, options, &teleport.values, scores, workers); + } + let mut next = Buffer::filled(n, 0.0, context)?; // Scale each row by its largest edge weight before summing, preventing // overflow for valid finite weights such as two f64::MAX parallel edges. let mut scales = Buffer::filled(n, 0.0f64, context)?; @@ -197,3 +214,113 @@ pub fn pagerank(graph: &GraphProjection, options: PageRankOptions<'_>) -> Result converged: false, }) } + +/// The parallel iteration for an unweighted projection. +/// +/// Each iteration computes, for every target `v`, +/// `next[v] = base * teleport[v] + damping * Σ score[u] / out_degree(u)` +/// over the in-arcs of `v`, which is the same distribution the push loop above +/// produces, summed in a different but equally fixed order. Every reduction over +/// nodes, the dangling mass and the residual, combines per-chunk partial sums in +/// chunk order, so the sequence of iterations and the final scores are identical +/// at one worker and at sixteen. +fn pull( + graph: &GraphProjection, + options: PageRankOptions<'_>, + teleport: &[f64], + mut scores: Buffer, + workers: usize, +) -> Result { + let context = graph.execution(); + let n = graph.node_count(); + let adjacency = graph.outgoing(); + let offsets = &adjacency.offsets.values; + let reverse = graph.reverse()?; + let mut next = Buffer::indexed(n, 0.0f64, context)?; + let mut residual = f64::INFINITY; + for iteration in 1..=options.max_iterations { + // Dangling mass: scores of nodes with no outgoing arc. + let dangling = crate::parallel::map_chunks( + context, + workers, + &scores.values, + |first, slice, meter| { + meter.charge(slice.len())?; + let mut sum = 0.0; + for (index, &score) in slice.iter().enumerate() { + let node = first + index; + if offsets[node + 1] == offsets[node] { + sum += score; + } + } + Ok(sum) + }, + )?; + let dangling = crate::parallel::reduce_in_order(&dangling, 0.0, |total, part| total + part); + let base = (1.0 - options.damping) + options.damping * dangling; + let scores_now = &scores.values; + let nonfinite = std::sync::atomic::AtomicBool::new(false); + crate::parallel::for_each_chunk( + context, + workers, + &mut next.values, + |first, slice, meter| { + for (index, value) in slice.iter_mut().enumerate() { + let node = first + index; + let arcs = reverse.range(node); + meter.charge(1 + arcs.len())?; + let mut sum = 0.0; + for arc in arcs { + let source = reverse.targets.values[arc]; + let out_degree = offsets[source + 1] - offsets[source]; + // A node with no outgoing arc contributes through the + // dangling mass in `base`, never through an arc. + sum += scores_now[source] / out_degree as f64; + } + let updated = base * teleport[node] + options.damping * sum; + if !updated.is_finite() { + nonfinite.store(true, std::sync::atomic::Ordering::Relaxed); + } + *value = updated; + } + Ok(()) + }, + )?; + if nonfinite.load(std::sync::atomic::Ordering::Relaxed) { + return Err(AlgorithmError::Numerical( + "PageRank produced a nonfinite score".into(), + )); + } + let parts = crate::parallel::map_chunks( + context, + workers, + &scores.values, + |first, slice, meter| { + meter.charge(slice.len())?; + let mut sum = 0.0; + for (index, &before) in slice.iter().enumerate() { + sum += (next.values[first + index] - before).abs(); + } + Ok(sum) + }, + )?; + residual = crate::parallel::reduce_in_order(&parts, 0.0, |total, part| total + part); + std::mem::swap(&mut scores, &mut next); + if residual <= options.tolerance { + return Ok(PageRank { + graph: graph.clone(), + scores, + iterations: iteration, + residual, + converged: true, + }); + } + } + Ok(PageRank { + graph: graph.clone(), + scores, + iterations: options.max_iterations, + residual, + converged: false, + }) +} diff --git a/crates/grust-algorithms/src/parallel.rs b/crates/grust-algorithms/src/parallel.rs index ee61f18c..aa3120c7 100644 --- a/crates/grust-algorithms/src/parallel.rs +++ b/crates/grust-algorithms/src/parallel.rs @@ -1,11 +1,224 @@ -//! Deterministic fan-out. A kernel splits its nodes into contiguous blocks of -//! roughly equal work, runs one task per block on the caller's rayon pool, and -//! combines the block results **in block order**. The partition depends only on -//! the graph and the pool width, and every combination used with it is exact -//! (integer sums, per-node writes to disjoint slots), so the answer is the same -//! at any width. Without the `parallel` feature the blocks run in a plain loop. - -use grust_procedures::Result; +//! Where kernels get their threads, and the rules they keep when they use them. +//! +//! Four properties hold for every parallel kernel in this crate. +//! +//! **The threads are the execution's, never the machine's.** How many threads a +//! query may add is [`ExecutionContext::concurrency`], which an embedder sets +//! per execution and which defaults to none at all; [`workers`] reports it, and +//! an execution that asked for nothing runs the sequential code. Nothing here +//! reads the CPU count, because these kernels also run inside a server that +//! owns its own runtime and thread pool. +//! +//! **Partitions follow the work for disjoint writes, and are fixed for +//! reductions.** A pass whose tasks write their own output slots, or merge into +//! shared atomics, may cut its work by the worker count, and should: see +//! [`balanced_ranges`] and [`chunk_len`]. A pass whose results are summed may +//! not, because regrouping the terms of a float sum changes its low bits. +//! Those use a fixed chunk, through [`map_chunks`], [`for_chunks`] or +//! [`ordered_blocks`]. +//! +//! **Results are combined in index order, never in completion order**, so a +//! kernel's answer is bit-for-bit the same at one worker and at sixteen, and +//! identifiers and orderings a kernel emits come from node rows rather than +//! from worker identity. +//! +//! **The sequential path stays.** Below a measured floor a kernel runs its +//! sequential code, because a pool costs more than it saves on small inputs, +//! and without the `parallel` feature only that code is compiled. It is also +//! the differential oracle the parallel paths are tested against. +//! +//! The two halves of this module came from two branches that grew in parallel: +//! the range and block helpers from the analytics catalog, the execution-bound +//! worker count, pool and work meters from the rayon branch. [`width`] is the +//! catalog's original source of the worker count, kept only until its kernels +//! move onto [`workers`]; it reads the ambient rayon pool, which is exactly what +//! an embedded server must not do. + +use grust_procedures::{ExecutionContext, Result}; + + +/// Work units below which kernels stay sequential. +/// +/// Provisional: this is a starting value, to be replaced by the crossover the +/// scaling benchmark measures on a machine with real cores. Until that number +/// exists, the constant is deliberately high, so small inputs keep today's +/// behavior rather than paying for a pool that may not earn it. +pub(crate) const SEQUENTIAL_BELOW_UNITS: usize = 1 << 18; + +/// Worker count for this execution, or `None` when the kernel should run its +/// sequential path: too little work, no concurrency asked for, or a build +/// without the `parallel` feature. +/// +/// An execution that explicitly asks for one worker gets `Some(1)`, which runs +/// the parallel implementation on one thread. That distinction is what lets the +/// scaling benchmark separate a change of algorithm from the cost of threads. +pub(crate) fn workers(context: &ExecutionContext, units: usize) -> Option { + if units < SEQUENTIAL_BELOW_UNITS { + return None; + } + #[cfg(feature = "parallel")] + { + context.concurrency_requested() + } + #[cfg(not(feature = "parallel"))] + { + let _ = context; + None + } +} + +/// Number of items per chunk so that `workers` threads each get a few chunks, +/// which keeps a skewed workload balanced without making chunks so small that +/// the per-chunk work meter dominates. +pub(crate) fn chunk_len(items: usize, workers: usize) -> usize { + if workers <= 1 { + return items.max(1); + } + items.div_ceil(workers.saturating_mul(4)).max(1024) +} + +/// Combine per-chunk results in chunk order, so the combination is independent +/// of the order in which the chunks finished. +pub(crate) fn reduce_in_order( + parts: &[T], + initial: R, + mut fold: impl FnMut(R, T) -> R, +) -> R { + let mut accumulated = initial; + for part in parts { + accumulated = fold(accumulated, *part); + } + accumulated +} + +/// Run `body` over `items` split into chunks, in parallel when `workers > 1`. +/// +/// `body` receives the chunk's starting index, its own mutable output slice and +/// a work meter of its own. Output slices are disjoint, so no kernel needs a +/// lock or an atomic to write its results. +#[cfg(feature = "parallel")] +pub(crate) fn for_each_chunk( + context: &ExecutionContext, + workers: usize, + output: &mut [T], + body: impl Fn(usize, &mut [T], &mut grust_procedures::WorkMeter) -> crate::Result<()> + Send + Sync, +) -> crate::Result<()> { + let chunk = chunk_len(output.len(), workers); + if workers <= 1 { + let mut meter = context.work_meter(); + for (index, slice) in output.chunks_mut(chunk).enumerate() { + body(index * chunk, slice, &mut meter)?; + } + return Ok(()); + } + use rayon::prelude::*; + pool(workers)?.install(|| { + output + .par_chunks_mut(chunk) + .enumerate() + .try_for_each(|(index, slice)| { + let mut meter = context.work_meter(); + body(index * chunk, slice, &mut meter) + }) + }) +} + +/// The sequential form, compiled when the `parallel` feature is off. +#[cfg(not(feature = "parallel"))] +pub(crate) fn for_each_chunk( + context: &ExecutionContext, + workers: usize, + output: &mut [T], + body: impl Fn(usize, &mut [T], &mut grust_procedures::WorkMeter) -> crate::Result<()>, +) -> crate::Result<()> { + let _ = workers; + let chunk = chunk_len(output.len(), 1); + let mut meter = context.work_meter(); + for (index, slice) in output.chunks_mut(chunk).enumerate() { + body(index * chunk, slice, &mut meter)?; + } + Ok(()) +} + +/// Map each chunk of `items` to one value, in parallel when `workers > 1`, and +/// return the values in chunk order for an ordered reduction. +#[cfg(feature = "parallel")] +pub(crate) fn map_chunks( + context: &ExecutionContext, + workers: usize, + items: &[T], + body: impl Fn(usize, &[T], &mut grust_procedures::WorkMeter) -> crate::Result + Send + Sync, +) -> crate::Result> { + let chunk = chunk_len(items.len(), workers); + if workers <= 1 { + let mut meter = context.work_meter(); + return items + .chunks(chunk) + .enumerate() + .map(|(index, slice)| body(index * chunk, slice, &mut meter)) + .collect(); + } + use rayon::prelude::*; + pool(workers)?.install(|| { + items + .par_chunks(chunk) + .enumerate() + .map(|(index, slice)| { + let mut meter = context.work_meter(); + body(index * chunk, slice, &mut meter) + }) + .collect() + }) +} + +#[cfg(not(feature = "parallel"))] +pub(crate) fn map_chunks( + context: &ExecutionContext, + workers: usize, + items: &[T], + body: impl Fn(usize, &[T], &mut grust_procedures::WorkMeter) -> crate::Result, +) -> crate::Result> { + let _ = workers; + let chunk = chunk_len(items.len(), 1); + let mut meter = context.work_meter(); + items + .chunks(chunk) + .enumerate() + .map(|(index, slice)| body(index * chunk, slice, &mut meter)) + .collect() +} + +/// One pool per worker count, built on first use and reused afterwards. +/// +/// A pool per kernel call would pay thread creation on every query; a pool +/// sized to the machine would ignore the execution's limit. Keying the cache by +/// worker count gives each distinct limit its own bounded pool, shared by every +/// query that asks for that many threads. +#[cfg(feature = "parallel")] +fn pool(workers: usize) -> crate::Result> { + use std::collections::HashMap; + use std::sync::{Arc, Mutex, OnceLock}; + + use grust_procedures::ProcedureError; + + static POOLS: OnceLock>>> = OnceLock::new(); + let pools = POOLS.get_or_init(|| Mutex::new(HashMap::new())); + let mut pools = pools + .lock() + .map_err(|_| ProcedureError::ResourceStatePoisoned)?; + if let Some(pool) = pools.get(&workers) { + return Ok(Arc::clone(pool)); + } + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(workers) + .thread_name(|index| format!("grust-algorithms-{index}")) + .build() + .map_err(|error| ProcedureError::Provider(Box::new(error)))?; + let pool = Arc::new(pool); + pools.insert(workers, Arc::clone(&pool)); + Ok(pool) +} + /// Tasks the caller's pool can run at once; 1 without the `parallel` feature. pub(crate) fn width() -> usize { diff --git a/crates/grust-algorithms/src/traversal.rs b/crates/grust-algorithms/src/traversal.rs index 60d549d7..c83cbc17 100644 --- a/crates/grust-algorithms/src/traversal.rs +++ b/crates/grust-algorithms/src/traversal.rs @@ -67,17 +67,34 @@ fn bfs_sources<'a>( ) -> Result { let context = graph.execution(); context.checkpoint()?; + let adjacency = graph.outgoing(); + let workers = crate::parallel::workers( + context, + graph + .node_count() + .saturating_add(adjacency.targets.values.len()), + ); + let mut roots = Vec::new(); + { + let mut seen = Buffer::filled(graph.node_count(), false, context)?; + for source in sources { + context.charge_work(1)?; + let source = graph.source(source)?; + if !seen.values[source] { + seen.values[source] = true; + roots.push(source); + } + } + } + if let Some(workers) = workers { + return levels(graph, &roots, workers); + } let mut distances = Buffer::filled(graph.node_count(), f64::INFINITY, context)?; let mut queue = Buffer::capacity(graph.node_count(), context)?; - for source in sources { - context.charge_work(1)?; - let source = graph.source(source)?; - if distances.values[source].is_infinite() { - distances.values[source] = 0.0; - queue.values.push(source); - } + for &source in &roots { + distances.values[source] = 0.0; + queue.values.push(source); } - let adjacency = graph.outgoing(); let mut head = 0; while head < queue.values.len() { context.charge_work(1)?; @@ -95,11 +112,101 @@ fn bfs_sources<'a>( Ok(Distances::new(graph, distances)) } +/// Level-synchronous breadth-first search. +/// +/// Each level expands the whole frontier in parallel. A node is claimed by the +/// one worker whose compare-exchange on its level succeeds, so it enters the +/// next frontier exactly once, and its distance is the level that claimed it. +/// Distances therefore do not depend on the order arcs were examined, which is +/// what makes the parallel result identical to the sequential queue's. The next +/// frontier is assembled from per-chunk lists in chunk order, so even the +/// frontier itself is reproducible. +fn levels(graph: &GraphProjection, roots: &[usize], workers: usize) -> Result { + use std::sync::atomic::{AtomicU32, Ordering}; + + const UNVISITED: u32 = u32::MAX; + let context = graph.execution(); + let n = graph.node_count(); + let adjacency = graph.outgoing(); + let offsets = &adjacency.offsets.values; + let targets = &adjacency.targets.values; + let level_of = Buffer::indexed_with(n, || AtomicU32::new(UNVISITED), context)?; + // Every node is claimed at most once, so the frontiers together hold at most + // one entry per node. Admit that once rather than per level or per worker. + let _frontiers = context.reserve(n.saturating_mul(2 * size_of::()))?; + let mut frontier: Vec = Vec::new(); + frontier.try_reserve(roots.len())?; + for &root in roots { + level_of.values[root].store(0, Ordering::Relaxed); + frontier.push(root); + } + let mut level: u32 = 0; + while !frontier.is_empty() { + context.checkpoint()?; + let next_level = level.saturating_add(1); + let lists = crate::parallel::map_chunks(context, workers, &frontier, |_, slice, meter| { + let mut discovered: Vec = Vec::new(); + for &node in slice { + let arcs = offsets[node]..offsets[node + 1]; + meter.charge(1 + arcs.len())?; + for arc in arcs { + let next = targets[arc]; + if level_of.values[next] + .compare_exchange( + UNVISITED, + next_level, + Ordering::Relaxed, + Ordering::Relaxed, + ) + .is_ok() + { + discovered.try_reserve(1)?; + discovered.push(next); + } + } + } + Ok(discovered) + })?; + frontier.clear(); + for list in &lists { + frontier.try_reserve(list.len())?; + frontier.extend_from_slice(list); + } + level = next_level; + } + let mut distances = Buffer::indexed(n, f64::INFINITY, context)?; + crate::parallel::for_each_chunk( + context, + workers, + &mut distances.values, + |first, slice, meter| { + meter.charge(slice.len())?; + for (index, distance) in slice.iter_mut().enumerate() { + let claimed = level_of.values[first + index].load(Ordering::Relaxed); + *distance = if claimed == UNVISITED { + f64::INFINITY + } else { + claimed as f64 + }; + } + Ok(()) + }, + )?; + Ok(Distances::new(graph, distances)) +} + /// Connected components after ignoring edge direction. Union by size and path /// halving use O(V) scratch and never require reverse adjacency. pub fn weakly_connected_components(graph: &GraphProjection) -> Result { let context = graph.execution(); let n = graph.node_count(); + let workers = crate::parallel::workers( + context, + n.saturating_add(graph.edge_count().saturating_mul(2)), + ); + if let Some(workers) = workers { + return union_find(graph, workers); + } let mut parents = Buffer::capacity(n, context)?; for node in 0..n { context.charge_work(1)?; @@ -150,6 +257,88 @@ fn root( Ok(node) } +/// Components by concurrent union-find. +/// +/// Workers union the endpoints of disjoint edge ranges into a shared parent +/// array. A link always points from the larger root to the smaller one, so the +/// root that survives in each component is its minimum node row: the same +/// canonical label the sequential kernel produces, arrived at without depending +/// on the order edges were merged. A losing compare-exchange re-reads the roots +/// and retries, which is why one pass over the edges is enough. +fn union_find(graph: &GraphProjection, workers: usize) -> Result { + use std::sync::atomic::{AtomicUsize, Ordering}; + + fn root(parents: &[AtomicUsize], mut node: usize) -> usize { + loop { + let parent = parents[node].load(Ordering::Relaxed); + if parent == node { + return node; + } + node = parent; + } + } + + let context = graph.execution(); + let n = graph.node_count(); + let mut row = 0; + let parents = Buffer::indexed_with( + n, + || { + let node = row; + row += 1; + AtomicUsize::new(node) + }, + context, + )?; + let parents = &parents.values; + crate::parallel::map_chunks(context, workers, graph.edges(), |_, slice, meter| { + for edge in slice { + meter.charge(1)?; + let mut left = root(parents, edge.source); + let mut right = root(parents, edge.target); + while left != right { + meter.charge(1)?; + let (larger, smaller) = if left > right { + (left, right) + } else { + (right, left) + }; + match parents[larger].compare_exchange( + larger, + smaller, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + // Another worker linked this root first; follow where it went. + Err(observed) => { + left = root(parents, observed); + right = root(parents, smaller); + } + } + } + } + Ok(()) + })?; + let mut labels = Buffer::indexed(n, 0usize, context)?; + crate::parallel::for_each_chunk( + context, + workers, + &mut labels.values, + |first, slice, meter| { + meter.charge(slice.len())?; + for (index, label) in slice.iter_mut().enumerate() { + *label = root(parents, first + index); + } + Ok(()) + }, + )?; + Ok(Components { + graph: graph.clone(), + labels, + }) +} + /// Strong components via two iterative depth-first passes. Stack usage is O(V) /// in admitted heap buffers, including on chains with millions of vertices. pub fn strongly_connected_components(graph: &GraphProjection) -> Result { diff --git a/crates/grust-algorithms/tests/parallel.rs b/crates/grust-algorithms/tests/parallel.rs new file mode 100644 index 00000000..fd6ba2fe --- /dev/null +++ b/crates/grust-algorithms/tests/parallel.rs @@ -0,0 +1,155 @@ +//! The parallel paths against the sequential ones, on a graph large enough to +//! cross the threshold where kernels use threads at all. +//! +//! Two properties per kernel. Against one thread: the same answer, exactly for +//! integers and identifiers, within a stated tolerance where the parallel path +//! sums floats in a different order. Against itself at two and sixteen threads: +//! bit-for-bit identical, because a result that moves with the thread count is +//! a result nobody can reproduce. + +use grust_algorithms::{ + ExecutionContext, ExecutionLimits, GraphProjection, Orientation, PageRankOptions, + ProjectionEdge, SnapshotIdentity, bfs, degree, multi_source_bfs, pagerank, + weakly_connected_components, +}; + +const NODES: usize = 120_000; +const EDGES: usize = 600_000; + +fn context(workers: usize) -> ExecutionContext { + ExecutionContext::new(ExecutionLimits { + memory_bytes: 1 << 30, + work_units: usize::MAX, + batch_rows: 8192, + deadline: None, + }) + .expect("valid limits") + .with_concurrency(workers) + .expect("not yet shared") +} + +/// A deterministic graph: a permutation cycle so every node has a successor, +/// plus pseudo-random chords from a fixed linear congruential sequence, so the +/// degree distribution is skewed enough that a naive split would be unbalanced. +fn graph(workers: usize, weighted: bool) -> GraphProjection { + let context = context(workers); + let nodes = (0..NODES).map(|id| id.to_string().into()).collect(); + let mut edges = Vec::with_capacity(EDGES); + let mut state = 0x2545_F491_4F6C_DD1Du64; + let mut random = move || { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + state + }; + for ordinal in 0..EDGES { + let (source, target) = if ordinal < NODES { + (ordinal, (ordinal + 1) % NODES) + } else { + // Bias sources towards low rows, giving a few very high degrees. + let source = (random() % (NODES as u64 / 16)) as usize; + (source, (random() % NODES as u64) as usize) + }; + edges.push(ProjectionEdge { + source, + target, + ordinal, + id: None, + }); + } + let weights = weighted.then(|| (0..EDGES).map(|index| 1.0 + (index % 7) as f64).collect()); + GraphProjection::from_topology( + SnapshotIdentity::new("parallel".into(), "r1".into(), "tests".into()).expect("identity"), + nodes, + edges, + weights, + Orientation::Outgoing, + &context, + ) + .expect("projection") +} + +#[test] +fn degree_is_identical_at_every_thread_count() { + let sequential = degree(&graph(1, true)).expect("sequential degree"); + for workers in [2, 16] { + let parallel = degree(&graph(workers, true)).expect("parallel degree"); + assert_eq!(parallel.counts(), sequential.counts(), "{workers} threads"); + // Each node sums its own arcs in CSR order on both paths, so weighted + // strengths are bit-for-bit equal, not merely close. + assert_eq!( + parallel.strengths(), + sequential.strengths(), + "{workers} threads" + ); + } +} + +#[test] +fn pagerank_agrees_with_the_sequential_push_and_does_not_move_with_threads() { + let options = PageRankOptions { + damping: 0.85, + tolerance: 1e-10, + max_iterations: 100, + personalization: None, + }; + let sequential = pagerank(&graph(1, false), options).expect("sequential pagerank"); + let two = pagerank(&graph(2, false), options).expect("pagerank at two threads"); + let sixteen = pagerank(&graph(16, false), options).expect("pagerank at sixteen threads"); + + // The parallel path pulls into each target instead of pushing out of each + // source, so the same distribution is summed in a different order. Scores + // agree to far better than any reported precision; the mass still sums to one. + for (node, (&pushed, &pulled)) in sequential.values().iter().zip(sixteen.values()).enumerate() { + assert!( + (pushed - pulled).abs() <= 1e-12 + 1e-9 * pushed.abs(), + "node {node}: push {pushed} against pull {pulled}" + ); + } + let total: f64 = sixteen.values().iter().sum(); + assert!((total - 1.0).abs() < 1e-9, "mass {total}"); + assert_eq!(sequential.converged(), sixteen.converged()); + + // Thread count changes nothing at all. + assert_eq!(two.values(), sixteen.values()); + assert_eq!(two.iterations(), sixteen.iterations()); + assert_eq!(two.residual(), sixteen.residual()); +} + +#[test] +fn pagerank_keeps_the_sequential_path_for_weighted_projections() { + let options = PageRankOptions::default(); + let sequential = pagerank(&graph(1, true), options).expect("sequential pagerank"); + let parallel = pagerank(&graph(16, true), options).expect("pagerank at sixteen threads"); + // Weighted projections take the push kernel on both paths, so the numbers + // are identical rather than merely close. + assert_eq!(parallel.values(), sequential.values()); + assert_eq!(parallel.iterations(), sequential.iterations()); +} + +#[test] +fn breadth_first_distances_are_identical_at_every_thread_count() { + let sequential = bfs(&graph(1, false), "0").expect("sequential bfs"); + let sources: Vec = ["0", "7", "4242"].iter().map(|id| id.to_string()).collect(); + let sequential_multi = multi_source_bfs(&graph(1, false), &sources).expect("sequential"); + for workers in [2, 16] { + let parallel = bfs(&graph(workers, false), "0").expect("parallel bfs"); + assert_eq!(parallel.values(), sequential.values(), "{workers} threads"); + let parallel_multi = + multi_source_bfs(&graph(workers, false), &sources).expect("parallel multi-source"); + assert_eq!( + parallel_multi.values(), + sequential_multi.values(), + "{workers} threads" + ); + } +} + +#[test] +fn components_are_identical_at_every_thread_count() { + let sequential = weakly_connected_components(&graph(1, false)).expect("sequential wcc"); + for workers in [2, 16] { + let parallel = weakly_connected_components(&graph(workers, false)).expect("parallel wcc"); + assert_eq!(parallel.values(), sequential.values(), "{workers} threads"); + } +} diff --git a/crates/grust-procedures/src/resources.rs b/crates/grust-procedures/src/resources.rs index 28862f90..a9856a67 100644 --- a/crates/grust-procedures/src/resources.rs +++ b/crates/grust-procedures/src/resources.rs @@ -53,7 +53,7 @@ const DEADLINE_SAMPLE_UNITS: usize = 1024; #[derive(Debug)] struct Shared { limits: ExecutionLimits, - concurrency: usize, + concurrency: Option, work_units: AtomicUsize, cancelled: AtomicBool, charges_since_deadline_read: AtomicUsize, @@ -94,7 +94,7 @@ impl ExecutionContext { } Ok(Self(Arc::new(Shared { limits, - concurrency: 1, + concurrency: None, work_units: AtomicUsize::new(0), cancelled: AtomicBool::new(false), charges_since_deadline_read: AtomicUsize::new(0), @@ -102,7 +102,8 @@ impl ExecutionContext { }))) } - /// Permit kernels to use up to `workers` threads for this execution. + /// Permit kernels to use up to `workers` threads for this execution, and + /// with it the parallel implementation of each kernel, even at one worker. /// /// Concurrency is a property of one execution, not of the process: an /// embedder that runs kernels inside a server with its own runtime and @@ -123,12 +124,23 @@ impl ExecutionContext { "concurrency must be set before the context is shared".into(), ) })?; - shared.concurrency = workers; + shared.concurrency = Some(workers); Ok(self) } /// Threads kernels may use for this execution; one unless set. pub fn concurrency(&self) -> usize { + self.0.concurrency.unwrap_or(1) + } + + /// The concurrency the caller asked for, or `None` when it never asked. + /// + /// Kernels distinguish the two: an execution that says nothing about + /// threads runs exactly the code it ran before parallel paths existed, + /// while an execution that explicitly asks for one worker runs the parallel + /// implementation on one thread. Keeping those apart is what lets a + /// benchmark separate the cost of threading from the change of algorithm. + pub fn concurrency_requested(&self) -> Option { self.0.concurrency } From 076ace4b185b1968854521f42e3d98f7973abb7d Mon Sep 17 00:00:00 2001 From: Alexy Khrabrov Date: Sun, 20 Sep 2026 02:09:37 +0000 Subject: [PATCH 3/8] grust-algorithms: measured thresholds, per-level frontiers, parallel statistics The sequential thresholds are now measurements rather than taste. On quegee, on roadNet-CA prefixes: PageRank and components are already faster in parallel at 20,000 edges, so the shared threshold sits just under that size. Breadth-first search is different, and needed two changes. Its own threshold is higher, because claiming a node with a compare-exchange costs more than testing a distance when nothing contends: it ran at 0.78x of sequential on a 20,000-edge graph and only reached 1.37x at 200,000. And it now decides level by level on the frontier it is about to expand, so a road network whose levels hold hundreds of nodes expands them in place while com-Orkut, whose frontiers reach millions, still spreads them. Projection statistics counts self-loops in parallel, the partials added in chunk order. Warm numbers at sixteen workers against the sequential kernels, twenty PageRank iterations, second call so a one-off reverse-topology build is not counted: roadNet-CA PageRank 34.4x, components 16.5x, breadth-first 2.7x, degree 2.5x; com-Orkut PageRank 22.5x, components 37.3x, breadth-first 7.8x. The results printed beside each timing are identical at every worker count, so none of it comes from doing less work. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Sak2FQcdgs5pkL2ruKrSUZ --- crates/grust-algorithms/examples/scaling.rs | 119 +++++++++++--------- crates/grust-algorithms/src/parallel.rs | 50 +++++++- crates/grust-algorithms/src/statistics.rs | 36 +++++- crates/grust-algorithms/src/traversal.rs | 54 +++++---- 4 files changed, 172 insertions(+), 87 deletions(-) diff --git a/crates/grust-algorithms/examples/scaling.rs b/crates/grust-algorithms/examples/scaling.rs index a720fae2..f9748aca 100644 --- a/crates/grust-algorithms/examples/scaling.rs +++ b/crates/grust-algorithms/examples/scaling.rs @@ -27,6 +27,10 @@ fn main() -> Result<(), Box> { // threads, which is what an embedder gets by default. let mut workers = vec![0usize, 1, 2, 4, 8, 16]; let mut iterations = 10usize; + // A kernel that builds a cached structure on first use (PageRank builds the + // reverse topology) pays for it once. Timing a second call separates that + // one-off cost from the iteration it amortises over. + let mut repeat = 1usize; let mut orientation = Orientation::Outgoing; let mut kernels = vec![ "degree".to_string(), @@ -44,6 +48,7 @@ fn main() -> Result<(), Box> { .collect::>()?; } "--iterations" => iterations = value()?.parse()?, + "--repeat" => repeat = value()?.parse()?, "--undirected" => orientation = Orientation::Undirected, "--kernels" => kernels = value()?.split(',').map(String::from).collect(), other => return Err(format!("unknown argument `{other}`").into()), @@ -129,59 +134,67 @@ fn main() -> Result<(), Box> { )?; let build = build.elapsed().as_secs_f64(); for kernel in &kernels { - let started = Instant::now(); - let result = match kernel.as_str() { - "degree" => { - let degrees = degree(&graph)?; - format!("arcs {}", degrees.counts().iter().sum::()) - } - "pagerank" => { - let ranks = pagerank( - &graph, - PageRankOptions { - damping: 0.85, - tolerance: 0.0, - max_iterations: iterations, - personalization: None, - }, - )?; - format!( - "iterations {} top {:.9}", - ranks.iterations(), - ranks.values().iter().copied().fold(0.0f64, f64::max) - ) - } - "bfs" => { - let distances = bfs(&graph, names[0])?; - let reached = distances.values().iter().filter(|d| d.is_finite()).count(); - let sum: f64 = distances.values().iter().filter(|d| d.is_finite()).sum(); - format!("reached {reached} sum {sum}") - } - "wcc" => { - let components = weakly_connected_components(&graph)?; - let mut labels: Vec = components.values().to_vec(); - labels.sort_unstable(); - labels.dedup(); - format!("components {}", labels.len()) - } - other => return Err(format!("unknown kernel `{other}`").into()), - }; - let seconds = started.elapsed().as_secs_f64(); - let speedup = match baseline.get(kernel) { - Some(&first) => format!("{:.2}x", first / seconds), - None => { - baseline.insert(kernel.clone(), seconds); - "-".to_string() - } - }; - let label = if threads == 0 { - "sequential".to_string() - } else { - threads.to_string() - }; - println!( - "{kernel:<10} {label:>10} {build:>10.1} {seconds:>10.3} {speedup:>9} {result}" - ); + for pass in 1..=repeat { + let started = Instant::now(); + let result = match kernel.as_str() { + "degree" => { + let degrees = degree(&graph)?; + format!("arcs {}", degrees.counts().iter().sum::()) + } + "pagerank" => { + let ranks = pagerank( + &graph, + PageRankOptions { + damping: 0.85, + tolerance: 0.0, + max_iterations: iterations, + personalization: None, + }, + )?; + format!( + "iterations {} top {:.9}", + ranks.iterations(), + ranks.values().iter().copied().fold(0.0f64, f64::max) + ) + } + "bfs" => { + let distances = bfs(&graph, names[0])?; + let reached = distances.values().iter().filter(|d| d.is_finite()).count(); + let sum: f64 = distances.values().iter().filter(|d| d.is_finite()).sum(); + format!("reached {reached} sum {sum}") + } + "wcc" => { + let components = weakly_connected_components(&graph)?; + let mut labels: Vec = components.values().to_vec(); + labels.sort_unstable(); + labels.dedup(); + format!("components {}", labels.len()) + } + other => return Err(format!("unknown kernel `{other}`").into()), + }; + let seconds = started.elapsed().as_secs_f64(); + let key = format!("{kernel}/{pass}"); + let speedup = match baseline.get(&key) { + Some(&first) => format!("{:.2}x", first / seconds), + None => { + baseline.insert(key, seconds); + "-".to_string() + } + }; + let label = if threads == 0 { + "sequential".to_string() + } else { + threads.to_string() + }; + let name = if repeat > 1 { + format!("{kernel}#{pass}") + } else { + kernel.clone() + }; + println!( + "{name:<10} {label:>10} {build:>10.1} {seconds:>10.3} {speedup:>9} {result}" + ); + } } } Ok(()) diff --git a/crates/grust-algorithms/src/parallel.rs b/crates/grust-algorithms/src/parallel.rs index aa3120c7..de320b7d 100644 --- a/crates/grust-algorithms/src/parallel.rs +++ b/crates/grust-algorithms/src/parallel.rs @@ -39,11 +39,39 @@ use grust_procedures::{ExecutionContext, Result}; /// Work units below which kernels stay sequential. /// -/// Provisional: this is a starting value, to be replaced by the crossover the -/// scaling benchmark measures on a machine with real cores. Until that number -/// exists, the constant is deliberately high, so small inputs keep today's -/// behavior rather than paying for a pool that may not earn it. -pub(crate) const SEQUENTIAL_BELOW_UNITS: usize = 1 << 18; +/// Measured on quegee (Xeon Platinum 8124M, 8 physical cores and 16 threads) +/// with `examples/scaling`, on prefixes of roadNet-CA at 20,000 to 500,000 +/// edges. PageRank and components were already faster in parallel at the +/// smallest size measured (4.4x and 4.3x at 20,000 edges), because their +/// parallel forms are better algorithms as well as threaded ones, so the +/// crossover for them is below the smallest graph worth measuring. This +/// constant sits just under that size: small enough not to withhold a win, +/// large enough that a graph of a few thousand edges never pays for a pool. +/// +/// Breadth-first search is the exception and does not use this constant alone: +/// on a high-diameter graph its levels are tiny, and at these sizes the +/// parallel expansion was 0.16x to 0.94x of sequential. It decides level by +/// level instead, on the size of the frontier it is about to expand. +pub(crate) const SEQUENTIAL_BELOW_UNITS: usize = 1 << 14; + +/// Work units below which breadth-first search stays sequential. +/// +/// Measured on roadNet-CA prefixes: at 20,000 edges the parallel expansion ran +/// at 0.78x of the sequential queue even with every level expanded in place, +/// because claiming a node with a compare-exchange costs more than testing a +/// distance when nothing contends. It reached 1.37x at 200,000 edges and 2.7x on +/// the whole graph, so the crossover is a few hundred thousand units; this sits +/// at the conservative end of that range. +pub(crate) const BREADTH_FIRST_SEQUENTIAL_BELOW_UNITS: usize = 1 << 18; + +/// Frontier size below which one breadth-first level expands sequentially. +/// +/// Measured on the same runs: on roadNet-CA, whose levels hold hundreds of +/// nodes, whole-kernel parallelism lost until the graph reached millions of +/// edges, while on com-Orkut, whose frontiers reach millions, it gained 7.8x. +/// Deciding per level keeps both: a small frontier costs nothing to expand in +/// place, and a large one is worth spreading. +pub(crate) const SEQUENTIAL_FRONTIER_BELOW: usize = 2048; /// Worker count for this execution, or `None` when the kernel should run its /// sequential path: too little work, no concurrency asked for, or a build @@ -53,7 +81,17 @@ pub(crate) const SEQUENTIAL_BELOW_UNITS: usize = 1 << 18; /// the parallel implementation on one thread. That distinction is what lets the /// scaling benchmark separate a change of algorithm from the cost of threads. pub(crate) fn workers(context: &ExecutionContext, units: usize) -> Option { - if units < SEQUENTIAL_BELOW_UNITS { + workers_above(context, units, SEQUENTIAL_BELOW_UNITS) +} + +/// As [`workers`], for a kernel whose own crossover is higher than the shared +/// one. Breadth-first search is the one such kernel today. +pub(crate) fn workers_above( + context: &ExecutionContext, + units: usize, + minimum: usize, +) -> Option { + if units < minimum { return None; } #[cfg(feature = "parallel")] diff --git a/crates/grust-algorithms/src/statistics.rs b/crates/grust-algorithms/src/statistics.rs index f05970e4..e8799aab 100644 --- a/crates/grust-algorithms/src/statistics.rs +++ b/crates/grust-algorithms/src/statistics.rs @@ -66,12 +66,36 @@ impl GraphProjection { /// Inspect this selected projection. Scanning loops is O(E), cancellable and /// charged as work; this does not execute an analytics kernel or build reverse CSR. pub fn statistics(&self) -> Result { - let mut self_loops = 0; - for edge in self.edges() { - self.execution().charge_work(1)?; - self_loops += usize::from(edge.source == edge.target); - } - self.execution().checkpoint()?; + let context = self.execution(); + // Counting loops is a sum over edges, so workers count their own ranges + // and the partials are added in chunk order. Integer addition is + // associative, so the total is the same however the edges were divided. + let self_loops = match crate::parallel::workers(context, self.edge_count()) { + Some(workers) => { + let parts = crate::parallel::map_chunks( + context, + workers, + self.edges(), + |_, slice, meter| { + meter.charge(slice.len())?; + Ok(slice + .iter() + .filter(|edge| edge.source == edge.target) + .count()) + }, + )?; + crate::parallel::reduce_in_order(&parts, 0usize, |total, part| total + part) + } + None => { + let mut self_loops = 0; + for edge in self.edges() { + context.charge_work(1)?; + self_loops += usize::from(edge.source == edge.target); + } + self_loops + } + }; + context.checkpoint()?; let arcs = self.outgoing().targets.values.len(); Ok(ProjectionStatistics { nodes: self.node_count(), diff --git a/crates/grust-algorithms/src/traversal.rs b/crates/grust-algorithms/src/traversal.rs index c83cbc17..3e290c36 100644 --- a/crates/grust-algorithms/src/traversal.rs +++ b/crates/grust-algorithms/src/traversal.rs @@ -68,11 +68,12 @@ fn bfs_sources<'a>( let context = graph.execution(); context.checkpoint()?; let adjacency = graph.outgoing(); - let workers = crate::parallel::workers( + let workers = crate::parallel::workers_above( context, graph .node_count() .saturating_add(adjacency.targets.values.len()), + crate::parallel::BREADTH_FIRST_SEQUENTIAL_BELOW_UNITS, ); let mut roots = Vec::new(); { @@ -144,29 +145,38 @@ fn levels(graph: &GraphProjection, roots: &[usize], workers: usize) -> Result = Vec::new(); - for &node in slice { - let arcs = offsets[node]..offsets[node + 1]; - meter.charge(1 + arcs.len())?; - for arc in arcs { - let next = targets[arc]; - if level_of.values[next] - .compare_exchange( - UNVISITED, - next_level, - Ordering::Relaxed, - Ordering::Relaxed, - ) - .is_ok() - { - discovered.try_reserve(1)?; - discovered.push(next); + // A level with a small frontier is expanded in place: spreading a few + // hundred nodes over sixteen workers costs more than it saves, which is + // what the measurements on road networks showed. + let level_workers = if frontier.len() < crate::parallel::SEQUENTIAL_FRONTIER_BELOW { + 1 + } else { + workers + }; + let lists = + crate::parallel::map_chunks(context, level_workers, &frontier, |_, slice, meter| { + let mut discovered: Vec = Vec::new(); + for &node in slice { + let arcs = offsets[node]..offsets[node + 1]; + meter.charge(1 + arcs.len())?; + for arc in arcs { + let next = targets[arc]; + if level_of.values[next] + .compare_exchange( + UNVISITED, + next_level, + Ordering::Relaxed, + Ordering::Relaxed, + ) + .is_ok() + { + discovered.try_reserve(1)?; + discovered.push(next); + } } } - } - Ok(discovered) - })?; + Ok(discovered) + })?; frontier.clear(); for list in &lists { frontier.try_reserve(list.len())?; From 70b70038b6c49b0f5e9506eaaa5e78727a44a886 Mon Sep 17 00:00:00 2001 From: Alexy Khrabrov Date: Sat, 19 Sep 2026 22:21:08 -0700 Subject: [PATCH 4/8] Review reproductions: three tests that fail on work/rayon-kernels at 8d7fb95 Two for WorkMeter refusing work that fits a budget because another meter holds an unspent block; one for PageRank scores changing bits with the worker count because chunk length depends on it. See codex-to-codex.md 2026-09-20T07:30Z. Co-Authored-By: Claude Fable 5.1 --- crates/grust-algorithms/tests/parallel.rs | 66 +++++++++++++++++++ .../tests/contracts/parallel.rs | 64 ++++++++++++++++++ 2 files changed, 130 insertions(+) diff --git a/crates/grust-algorithms/tests/parallel.rs b/crates/grust-algorithms/tests/parallel.rs index fd6ba2fe..029d99a0 100644 --- a/crates/grust-algorithms/tests/parallel.rs +++ b/crates/grust-algorithms/tests/parallel.rs @@ -153,3 +153,69 @@ fn components_are_identical_at_every_thread_count() { assert_eq!(parallel.values(), sequential.values(), "{workers} threads"); } } + +// --- Review reproduction (catalog agent, 2026-09-20). Fails at 8d7fb95. --- + +/// Half the nodes have no out-arcs, so the dangling mass is a long float sum. +/// Folding per-chunk sums in order fixes their order, not their grouping: if the +/// chunk length depends on the worker count, so do the low bits of every score. +#[test] +fn pagerank_is_bit_identical_at_every_worker_count_with_dangling_nodes() { + const N: usize = 200_000; + let run = |workers: usize| { + let context = context(workers); + let mut state = 0x9E37_79B9_7F4A_7C15u64; + let mut next = move || { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + state + }; + let edges = (0..3 * N) + .map(|ordinal| { + let source = (next() % (N as u64 / 2)) as usize; + let target = next() % N as u64; + ProjectionEdge { + source, + target: (target * target / N as u64) as usize, + ordinal, + id: None, + } + }) + .collect(); + let graph = GraphProjection::from_topology( + SnapshotIdentity::new("g".into(), "r".into(), "reader".into()).unwrap(), + (0..N).map(|id| id.to_string().into()).collect(), + edges, + None, + Orientation::Outgoing, + &context, + ) + .unwrap(); + let result = pagerank( + &graph, + PageRankOptions { + max_iterations: 40, + tolerance: 1e-12, + ..Default::default() + }, + ) + .unwrap(); + let bits: Vec = result.values().iter().map(|v| v.to_bits()).collect(); + (bits, result.iterations(), result.residual().to_bits()) + }; + let one = run(1); + for workers in [2, 4, 16] { + let other = run(workers); + let differing = one.0.iter().zip(&other.0).filter(|(a, b)| a != b).count(); + assert_eq!( + differing, 0, + "{differing} of {N} scores differ between 1 and {workers} workers" + ); + assert_eq!( + (one.1, one.2), + (other.1, other.2), + "iterations or residual moved at {workers} workers" + ); + } +} diff --git a/crates/grust-procedures/tests/contracts/parallel.rs b/crates/grust-procedures/tests/contracts/parallel.rs index b63bb3d1..a350fa28 100644 --- a/crates/grust-procedures/tests/contracts/parallel.rs +++ b/crates/grust-procedures/tests/contracts/parallel.rs @@ -197,3 +197,67 @@ fn concurrency_is_one_unless_set_and_cannot_change_once_shared() { .is_err() ); } + +// --- Review reproductions (catalog agent, 2026-09-20). Both fail at 8d7fb95. --- + +/// Two meters driven from one thread, so there is no race to blame. One meter +/// charges a single unit and then sits on the rest of its block; the other must +/// still be able to spend what the budget has left. Whether work fits a budget +/// must not depend on how many meters exist. +#[test] +fn an_idle_meters_unspent_block_does_not_refuse_work_that_fits() { + const BUDGET: usize = 1500; + let execution = ExecutionContext::new(limits(BUDGET, None)).expect("valid limits"); + let mut idle = execution.work_meter(); + let mut busy = execution.work_meter(); + idle.charge(1).expect("first unit"); + for unit in 2..=BUDGET { + busy.charge(1).unwrap_or_else(|error| { + panic!( + "refused unit {unit} of a {BUDGET}-unit budget after {} units of work: {error}", + unit - 1 + ) + }); + } + assert!( + busy.charge(1).is_err(), + "unit {} exceeds the budget", + BUDGET + 1 + ); +} + +/// The same at sixteen threads with skewed work: one worker does nine tenths of +/// it while the others finish early but keep their meters alive, as workers do +/// inside a parallel region until it ends. +#[test] +fn skewed_work_that_exactly_fits_succeeds_at_sixteen_threads() { + let light = 100; + let heavy = 9 * (WORKERS - 1) * light; + let total = heavy + (WORKERS - 1) * light; + let execution = ExecutionContext::new(limits(total, None)).expect("valid limits"); + let failures = AtomicUsize::new(0); + let finished = std::sync::Barrier::new(WORKERS); + std::thread::scope(|scope| { + for worker in 0..WORKERS { + let (execution, failures, finished) = (execution.clone(), &failures, &finished); + scope.spawn(move || { + let mut meter = execution.work_meter(); + let units = if worker == 0 { heavy } else { light }; + for _ in 0..units { + if meter.charge(1).is_err() { + failures.fetch_add(1, Ordering::Relaxed); + break; + } + } + // Hold the meter until every worker is done, like a region does. + finished.wait(); + }); + } + }); + assert_eq!( + failures.load(Ordering::Relaxed), + 0, + "the work fits the budget" + ); + assert_eq!(execution.usage().expect("usage").work_units, total); +} From 579c2096b6234585db8888cbbae4abcf0f3b362a Mon Sep 17 00:00:00 2001 From: Alexy Khrabrov Date: Sun, 20 Sep 2026 05:34:46 +0000 Subject: [PATCH 5/8] Fix both defects the review reproduced, and pay for one of them 1. A work meter could refuse work that fits. Exact admission was measured against a counter that still included other meters' unspent grants, and those came back only when their meter dropped, so whether a query succeeded depended on how many meters existed. Grants are now reclaimable: a meter that cannot be admitted returns its own idle remainder, takes back every other live meter's, and retries. The registry is touched when a meter is created, dropped or refused, never on a charge. 2. PageRank's scores moved with the worker count. Folding per-chunk sums in chunk order fixes their order but not their grouping, and the chunk length came from the worker count, so the dangling mass and through it every score changed in the low bits. Reductions now use a fixed chunk of 4,096, matching the catalog kernels; passes that write each item's own slot keep worker-derived chunks, where grouping cannot affect the answer. The cost is real and worth stating: on roadNet-CA at sixteen workers PageRank falls from 34.4x to 28.8x against the sequential kernel, and components from 16.5x to 14.4x. Creating one meter per worker task rather than per chunk recovered part of it; a fixed reduction chunk is simply more chunks. Also from the review: the pool cache is capped at eight distinct worker counts so an embedder passing user-supplied values cannot accumulate idle threads, the doc now says pools are shared per count per process, the discarded admission error is explained, and the changelog records that PageRank's low digits differ from the sequential kernel's once concurrency is requested. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Sak2FQcdgs5pkL2ruKrSUZ --- CHANGELOG.md | 17 +++ crates/grust-algorithms/src/parallel.rs | 88 ++++++++++--- crates/grust-algorithms/src/traversal.rs | 14 +- crates/grust-procedures/src/resources.rs | 123 +++++++++++++++--- .../tests/contracts/parallel.rs | 2 +- 5 files changed, 202 insertions(+), 42 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 84067445..4b004dbc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -184,6 +184,23 @@ reconstructed from Git history, release commits, and the shipped docs. requires every kernel's Arrow columns to equal its declared outputs. `projectionStats` and `estimateCsr` are refused there: they are not kernels over a projection. +### Added + +- `ExecutionContext::with_concurrency` permits a kernel to use threads, and + `ExecutionContext::work_meter` gives one worker a batched share of the work + budget. An execution that does not ask for threads behaves exactly as before. +- `grust-algorithms` runs `degree`, `pagerank`, `bfs`, `multiSourceBfs`, `wcc` + and `projectionStats` in parallel when an execution asks for threads, behind + the default `parallel` feature. Results do not depend on the worker count. + +### Changed + +- PageRank computes an unweighted projection by pulling into each target rather + than pushing out of each source when concurrency is requested. The + distribution is the same and the two paths agree to within a relative 1e-9, + but the low digits of a score differ from the sequential kernel's, at any + worker count including one. Evidence that pins PageRank scores to more than + nine digits should be regenerated rather than compared across this change. ## 0.21.0 — Tadpole — 2026-09-18 ### Cypher language diff --git a/crates/grust-algorithms/src/parallel.rs b/crates/grust-algorithms/src/parallel.rs index de320b7d..67f64058 100644 --- a/crates/grust-algorithms/src/parallel.rs +++ b/crates/grust-algorithms/src/parallel.rs @@ -20,7 +20,9 @@ //! **Results are combined in index order, never in completion order**, so a //! kernel's answer is bit-for-bit the same at one worker and at sixteen, and //! identifiers and orderings a kernel emits come from node rows rather than -//! from worker identity. +//! from worker identity. The fixed reduction chunk is [`REDUCTION_CHUNK_LEN`]; +//! [`map_chunks_sized`] takes its own size, for a pass where grouping cannot +//! change an answer. //! //! **The sequential path stays.** Below a measured floor a kernel runs its //! sequential code, because a pool costs more than it saves on small inputs, @@ -105,9 +107,11 @@ pub(crate) fn workers_above( } } -/// Number of items per chunk so that `workers` threads each get a few chunks, -/// which keeps a skewed workload balanced without making chunks so small that -/// the per-chunk work meter dominates. +/// Items per chunk for a pass that writes each item's own output slot. +/// +/// Sized from the worker count, so threads each get a few chunks and a skewed +/// workload stays balanced. Safe here because the chunking cannot change any +/// result: every item's value depends on that item alone. pub(crate) fn chunk_len(items: usize, workers: usize) -> usize { if workers <= 1 { return items.max(1); @@ -115,6 +119,18 @@ pub(crate) fn chunk_len(items: usize, workers: usize) -> usize { items.div_ceil(workers.saturating_mul(4)).max(1024) } +/// Items per chunk for a pass whose results are reduced. +/// +/// Fixed, and deliberately not a function of the worker count. Folding +/// per-chunk values in chunk order fixes the order of a float sum but not its +/// grouping, and regrouping a float sum changes its low bits. A dangling mass +/// summed in chunks of 4,096 is therefore the same number at one worker and at +/// sixteen, which a chunk length derived from the worker count would not be — +/// the defect `pagerank_is_bit_identical_at_every_worker_count_with_dangling_nodes` +/// reproduces. The constant matches the catalog kernels' `spectral.rs`, whose +/// residuals are bit-identical across widths. +pub(crate) const REDUCTION_CHUNK_LEN: usize = 4096; + /// Combine per-chunk results in chunk order, so the combination is independent /// of the order in which the chunks finished. pub(crate) fn reduce_in_order( @@ -150,14 +166,13 @@ pub(crate) fn for_each_chunk( return Ok(()); } use rayon::prelude::*; + // One meter per worker task rather than per chunk: registering a meter takes + // a lock, and a fixed reduction chunk makes chunks many and small. pool(workers)?.install(|| { - output - .par_chunks_mut(chunk) - .enumerate() - .try_for_each(|(index, slice)| { - let mut meter = context.work_meter(); - body(index * chunk, slice, &mut meter) - }) + output.par_chunks_mut(chunk).enumerate().try_for_each_init( + || context.work_meter(), + |meter, (index, slice)| body(index * chunk, slice, meter), + ) }) } @@ -181,13 +196,14 @@ pub(crate) fn for_each_chunk( /// Map each chunk of `items` to one value, in parallel when `workers > 1`, and /// return the values in chunk order for an ordered reduction. #[cfg(feature = "parallel")] -pub(crate) fn map_chunks( +pub(crate) fn map_chunks_sized( context: &ExecutionContext, workers: usize, items: &[T], + chunk: usize, body: impl Fn(usize, &[T], &mut grust_procedures::WorkMeter) -> crate::Result + Send + Sync, ) -> crate::Result> { - let chunk = chunk_len(items.len(), workers); + let chunk = chunk.max(1); if workers <= 1 { let mut meter = context.work_meter(); return items @@ -201,23 +217,24 @@ pub(crate) fn map_chunks( items .par_chunks(chunk) .enumerate() - .map(|(index, slice)| { - let mut meter = context.work_meter(); - body(index * chunk, slice, &mut meter) - }) + .map_init( + || context.work_meter(), + |meter, (index, slice)| body(index * chunk, slice, meter), + ) .collect() }) } #[cfg(not(feature = "parallel"))] -pub(crate) fn map_chunks( +pub(crate) fn map_chunks_sized( context: &ExecutionContext, workers: usize, items: &[T], + chunk: usize, body: impl Fn(usize, &[T], &mut grust_procedures::WorkMeter) -> crate::Result, ) -> crate::Result> { let _ = workers; - let chunk = chunk_len(items.len(), 1); + let chunk = chunk.max(1); let mut meter = context.work_meter(); items .chunks(chunk) @@ -226,12 +243,38 @@ pub(crate) fn map_chunks( .collect() } +/// Map each chunk of `items` to one value and return the values in chunk order, +/// for a caller that reduces them. The grouping is fixed by +/// [`REDUCTION_CHUNK_LEN`], so a float reduction is identical at every worker +/// count; a pass whose results do not depend on grouping can pick its own chunk +/// size with [`map_chunks_sized`]. +pub(crate) fn map_chunks( + context: &ExecutionContext, + workers: usize, + items: &[T], + body: impl Fn(usize, &[T], &mut grust_procedures::WorkMeter) -> crate::Result + Send + Sync, +) -> crate::Result> { + map_chunks_sized(context, workers, items, REDUCTION_CHUNK_LEN, body) +} + /// One pool per worker count, built on first use and reused afterwards. /// /// A pool per kernel call would pay thread creation on every query; a pool /// sized to the machine would ignore the execution's limit. Keying the cache by -/// worker count gives each distinct limit its own bounded pool, shared by every -/// query that asks for that many threads. +/// worker count gives each distinct limit its own bounded pool. +/// +/// Two things follow, and both are deliberate. Pools are shared: two executions +/// that each ask for eight workers run on the same eight threads, so the bound +/// is per distinct count per process rather than per query — which is what an +/// embedded server wants, since a hundred concurrent queries asking for eight +/// each should not start eight hundred threads. And the cache is capped, so an +/// embedder passing a user-supplied count cannot accumulate a pool per value: +/// past [`MAX_CACHED_POOLS`] distinct counts the cache is emptied and refilled. +/// Emptying it does not stop a pool that is running; its last user drops it. +/// Distinct worker counts whose pools are kept alive between queries. +#[cfg(feature = "parallel")] +const MAX_CACHED_POOLS: usize = 8; + #[cfg(feature = "parallel")] fn pool(workers: usize) -> crate::Result> { use std::collections::HashMap; @@ -253,6 +296,9 @@ fn pool(workers: usize) -> crate::Result> { .build() .map_err(|error| ProcedureError::Provider(Box::new(error)))?; let pool = Arc::new(pool); + if pools.len() >= MAX_CACHED_POOLS { + pools.clear(); + } pools.insert(workers, Arc::clone(&pool)); Ok(pool) } diff --git a/crates/grust-algorithms/src/traversal.rs b/crates/grust-algorithms/src/traversal.rs index 3e290c36..2791f034 100644 --- a/crates/grust-algorithms/src/traversal.rs +++ b/crates/grust-algorithms/src/traversal.rs @@ -153,8 +153,15 @@ fn levels(graph: &GraphProjection, roots: &[usize], workers: usize) -> Result = Vec::new(); for &node in slice { let arcs = offsets[node]..offsets[node + 1]; @@ -176,7 +183,8 @@ fn levels(graph: &GraphProjection, roots: &[usize], workers: usize) -> Result>>, state: Mutex, } @@ -98,6 +102,7 @@ impl ExecutionContext { work_units: AtomicUsize::new(0), cancelled: AtomicBool::new(false), charges_since_deadline_read: AtomicUsize::new(0), + grants: Mutex::new(Vec::new()), state: Mutex::new(State::default()), }))) } @@ -187,10 +192,47 @@ impl ExecutionContext { /// Cancellation is still observed per charge. The deadline is sampled per /// block, which is the same cadence per worker as a single-threaded charge. pub fn work_meter(&self) -> WorkMeter { + let balance = Arc::new(AtomicUsize::new(0)); + // Registration is per meter, not per charge: a kernel creates one per + // chunk of work, so this lock is taken a handful of times per region. + self.0 + .grants + .lock() + .unwrap_or_else(|poison| poison.into_inner()) + .push(Arc::clone(&balance)); WorkMeter { context: self.clone(), - granted: 0, + balance, + } + } + + /// Return every other live meter's unspent grant to the budget, and report + /// how much came back. + fn reclaim_grants(&self, own: &Arc) -> usize { + let grants = self + .0 + .grants + .lock() + .unwrap_or_else(|poison| poison.into_inner()); + let mut reclaimed = 0usize; + for balance in grants.iter() { + if Arc::ptr_eq(balance, own) { + continue; + } + reclaimed = reclaimed.saturating_add(balance.swap(0, Ordering::Relaxed)); } + drop(grants); + self.refund_work(reclaimed); + reclaimed + } + + fn release_meter(&self, own: &Arc) { + self.refund_work(own.swap(0, Ordering::Relaxed)); + self.0 + .grants + .lock() + .unwrap_or_else(|poison| poison.into_inner()) + .retain(|balance| !Arc::ptr_eq(balance, own)); } /// Charge work before performing it. Counter overflow is a budget failure. @@ -384,7 +426,9 @@ impl ExecutionContext { /// [`ExecutionContext::charge_work`]. pub struct WorkMeter { context: ExecutionContext, - granted: usize, + /// Admitted but unspent units. Shared, not local, because a meter that runs + /// out must be able to take back what idle meters are sitting on. + balance: Arc, } impl WorkMeter { @@ -395,27 +439,73 @@ impl WorkMeter { /// charge would exceed. pub fn charge(&mut self, units: usize) -> Result<()> { self.context.check_cancelled()?; - if let Some(left) = self.granted.checked_sub(units) { - self.granted = left; - return Ok(()); + // The balance can be taken by another meter between the load and the + // exchange, so this retries rather than subtracting blindly. + let mut held = self.balance.load(Ordering::Relaxed); + while held >= units { + match self.balance.compare_exchange_weak( + held, + held - units, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => return Ok(()), + Err(observed) => held = observed, + } } self.admit(units) } - /// Admit a block, or exactly what was asked for when a block does not fit. + /// Admit a block; failing that, exactly what was asked for; failing that, + /// take back every idle grant and try once more. + /// + /// The last step is what keeps admission independent of the number of + /// meters. Without it, a block granted to a meter that has stopped charging + /// counts against the budget until that meter drops, so work that fits + /// could be refused because of who else exists — the failure the contract + /// test `an_idle_meters_unspent_block_does_not_refuse_work_that_fits` + /// reproduces. #[cold] fn admit(&mut self, units: usize) -> Result<()> { self.context.check_state(DeadlineCheck::Sampled)?; - let wanted = units.saturating_add(WORK_BLOCK_UNITS); - if wanted > units && self.context.admit_work(wanted - self.granted).is_ok() { - self.granted = wanted - units; - return Ok(()); + // Three attempts: grant, exact, and one more of each after reclaiming. + // A grant in transit inside another meter's own admission cannot be + // reclaimed, so a retry covers that window instead of a lock. + for attempt in 0..3 { + let held = self.balance.swap(0, Ordering::Relaxed); + let needed = units.saturating_sub(held); + let block = needed.saturating_add(WORK_BLOCK_UNITS); + // A block on top of the need, while the budget has room for one. + // The only failure `admit_work` reports is an exceeded budget, which + // is exactly the case the next attempt handles, so it is discarded + // here rather than returned. + if block > needed && self.context.admit_work(block).is_ok() { + self.balance + .fetch_add(held + block - units, Ordering::Relaxed); + return Ok(()); + } + // Exactly what this charge needs, so the budget fails at the unit + // that exceeds it rather than a block early. + match self.context.admit_work(needed) { + Ok(()) => { + self.balance + .fetch_add(held + needed - units, Ordering::Relaxed); + return Ok(()); + } + Err(exceeded) => { + // Give back this meter's own idle remainder first, then take + // back what other idle meters hold, and try again. + self.context.refund_work(held); + if attempt == 2 || self.context.reclaim_grants(&self.balance) == 0 { + return Err(exceeded); + } + } + } } - // Near the limit, spend the remaining budget one charge at a time so - // that the failure happens at the same unit as it would at one thread. - self.context.admit_work(units - self.granted)?; - self.granted = 0; - Ok(()) + Err(ProcedureError::BudgetExceeded { + resource: "work", + limit: self.context.0.limits.work_units, + }) } /// Shared execution state, for an exact poll or a nested reservation. @@ -439,8 +529,7 @@ impl WorkMeter { impl Drop for WorkMeter { fn drop(&mut self) { - self.context.refund_work(self.granted); - self.granted = 0; + self.context.release_meter(&self.balance); } } diff --git a/crates/grust-procedures/tests/contracts/parallel.rs b/crates/grust-procedures/tests/contracts/parallel.rs index a350fa28..b33f4c7f 100644 --- a/crates/grust-procedures/tests/contracts/parallel.rs +++ b/crates/grust-procedures/tests/contracts/parallel.rs @@ -66,7 +66,7 @@ fn a_budget_one_unit_short_fails_at_one_worker() { let (failures, usage) = charge_everywhere(&execution, PER_WORKER); assert_eq!(failures, 1, "exactly the worker that ran out should fail"); assert!( - usage.work_units <= total - 1, + usage.work_units < total, "spent {} of a {} unit budget", usage.work_units, total - 1 From b287c8551fde6a501c582bc70ce027b6d8f23ca6 Mon Sep 17 00:00:00 2001 From: Alexy Khrabrov Date: Sun, 20 Sep 2026 05:47:58 +0000 Subject: [PATCH 6/8] Keep the fixed reduction chunk where it belongs Components and projection statistics do not sum floats: unions are applied to shared atomics, and a self-loop count is an integer sum, which is the same however it is grouped. Both were taking the fixed reduction chunk anyway, which cost components half its speed on com-Orkut, 37x down to 20x. They follow the worker count again. Only PageRank's dangling mass and residual, which are float reductions, need a grouping independent of the worker count. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Sak2FQcdgs5pkL2ruKrSUZ --- crates/grust-algorithms/src/statistics.rs | 6 ++- crates/grust-algorithms/src/traversal.rs | 59 +++++++++++++---------- 2 files changed, 39 insertions(+), 26 deletions(-) diff --git a/crates/grust-algorithms/src/statistics.rs b/crates/grust-algorithms/src/statistics.rs index e8799aab..5af14751 100644 --- a/crates/grust-algorithms/src/statistics.rs +++ b/crates/grust-algorithms/src/statistics.rs @@ -72,10 +72,14 @@ impl GraphProjection { // associative, so the total is the same however the edges were divided. let self_loops = match crate::parallel::workers(context, self.edge_count()) { Some(workers) => { - let parts = crate::parallel::map_chunks( + // An integer sum is the same however it is grouped, so these + // chunks may follow the worker count. + let chunk = crate::parallel::chunk_len(self.edge_count(), workers); + let parts = crate::parallel::map_chunks_sized( context, workers, self.edges(), + chunk, |_, slice, meter| { meter.charge(slice.len())?; Ok(slice diff --git a/crates/grust-algorithms/src/traversal.rs b/crates/grust-algorithms/src/traversal.rs index 2791f034..d4c8b815 100644 --- a/crates/grust-algorithms/src/traversal.rs +++ b/crates/grust-algorithms/src/traversal.rs @@ -309,35 +309,44 @@ fn union_find(graph: &GraphProjection, workers: usize) -> Result { context, )?; let parents = &parents.values; - crate::parallel::map_chunks(context, workers, graph.edges(), |_, slice, meter| { - for edge in slice { - meter.charge(1)?; - let mut left = root(parents, edge.source); - let mut right = root(parents, edge.target); - while left != right { + // Unions are applied to shared atomics, not summed, so the grouping cannot + // change the partition and the chunks may follow the worker count. + let chunk = crate::parallel::chunk_len(graph.edge_count(), workers); + crate::parallel::map_chunks_sized( + context, + workers, + graph.edges(), + chunk, + |_, slice, meter| { + for edge in slice { meter.charge(1)?; - let (larger, smaller) = if left > right { - (left, right) - } else { - (right, left) - }; - match parents[larger].compare_exchange( - larger, - smaller, - Ordering::Relaxed, - Ordering::Relaxed, - ) { - Ok(_) => break, - // Another worker linked this root first; follow where it went. - Err(observed) => { - left = root(parents, observed); - right = root(parents, smaller); + let mut left = root(parents, edge.source); + let mut right = root(parents, edge.target); + while left != right { + meter.charge(1)?; + let (larger, smaller) = if left > right { + (left, right) + } else { + (right, left) + }; + match parents[larger].compare_exchange( + larger, + smaller, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + // Another worker linked this root first; follow where it went. + Err(observed) => { + left = root(parents, observed); + right = root(parents, smaller); + } } } } - } - Ok(()) - })?; + Ok(()) + }, + )?; let mut labels = Buffer::indexed(n, 0usize, context)?; crate::parallel::for_each_chunk( context, From bca732ca9c1f09d5918e4c45da46f8b2361821b6 Mon Sep 17 00:00:00 2001 From: Alexy Khrabrov Date: Sun, 20 Sep 2026 06:34:03 +0000 Subject: [PATCH 7/8] One parallel layer, one reverse index The two parallel modules become their union. The catalog's range and block helpers stay as they are: balanced_ranges beats chunking by count on any skewed degree distribution, which is every social graph. The worker count, the pool and the work meters come from this branch, because they are bounded by what the execution asked for rather than by the machine. width() stays for now, with a doc note, since seventeen catalog kernels still call it; step 5 retires it. The projection has one reverse index instead of two. Reachability had a narrow one and the catalog added a transpose with weights and edge slots; a single build that carries weights costs less than two builds over the same arcs. Strongly connected components and PageRank both read incoming() now. Edge slots became optional and a transpose leaves them out, because no caller of a transpose reads them: eight bytes an arc that SCC and PageRank would have paid to ignore, about a gigabyte on a directed graph of com-Orkut's size. Weighted PageRank follows from that. The in-arc index carries each arc's weight, so the weighted projection pulls like the unweighted one instead of falling back to the sequential push, which is now only the oracle. Measured on roadNet-CA with synthetic weights, twenty iterations, warm: 19.0x at sixteen workers where there was no parallel path at all before. From the review: a stress test provokes the reclaim retry window directly, two hundred rounds of sixteen meters starting together on a budget equal to their work, with no refusal observed; and the doc comment now says what a caller sees if all three retries lose, and how to leave headroom against it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Sak2FQcdgs5pkL2ruKrSUZ --- crates/grust-algorithms/examples/scaling.rs | 17 ++- crates/grust-algorithms/src/biconnected.rs | 2 +- crates/grust-algorithms/src/flow.rs | 2 +- crates/grust-algorithms/src/pagerank.rs | 101 ++++++++++++++---- crates/grust-algorithms/src/parallel.rs | 2 - crates/grust-algorithms/src/projection.rs | 18 +--- .../src/projection/adjacency.rs | 81 ++++++-------- crates/grust-algorithms/src/shortest.rs | 2 +- crates/grust-algorithms/src/spanning.rs | 2 +- crates/grust-algorithms/src/traversal.rs | 4 +- crates/grust-algorithms/tests/parallel.rs | 57 ++++++++-- crates/grust-procedures/src/resources.rs | 13 +++ .../tests/contracts/parallel.rs | 46 ++++++++ 13 files changed, 238 insertions(+), 109 deletions(-) diff --git a/crates/grust-algorithms/examples/scaling.rs b/crates/grust-algorithms/examples/scaling.rs index f9748aca..85f7a723 100644 --- a/crates/grust-algorithms/examples/scaling.rs +++ b/crates/grust-algorithms/examples/scaling.rs @@ -1,7 +1,8 @@ //! Time the parallel kernels against their sequential paths on a real graph. //! -//! scaling [--workers 1,2,4,8,16] [--iterations 10] -//! [--undirected] [--kernels degree,pagerank,bfs,wcc] +//! scaling [--workers 0,1,2,4,8,16] [--iterations 10] +//! [--undirected] [--weighted] [--repeat 2] +//! [--kernels degree,pagerank,bfs,wcc] //! //! The edge list is one `source target` pair per line, `#` comments ignored, //! as SNAP and the strain harness datasets ship them. The same projection is @@ -32,6 +33,7 @@ fn main() -> Result<(), Box> { // one-off cost from the iteration it amortises over. let mut repeat = 1usize; let mut orientation = Orientation::Outgoing; + let mut weighted = false; let mut kernels = vec![ "degree".to_string(), "pagerank".to_string(), @@ -50,6 +52,9 @@ fn main() -> Result<(), Box> { "--iterations" => iterations = value()?.parse()?, "--repeat" => repeat = value()?.parse()?, "--undirected" => orientation = Orientation::Undirected, + // Weights are synthesised from the edge position, so a run is + // reproducible and every row has a different scale. + "--weighted" => weighted = true, "--kernels" => kernels = value()?.split(',').map(String::from).collect(), other => return Err(format!("unknown argument `{other}`").into()), } @@ -128,7 +133,13 @@ fn main() -> Result<(), Box> { id: None, }) .collect(), - None, + weighted.then(|| { + pairs + .iter() + .enumerate() + .map(|(ordinal, _)| 1.0 + (ordinal % 9) as f64) + .collect() + }), orientation, &context, )?; diff --git a/crates/grust-algorithms/src/biconnected.rs b/crates/grust-algorithms/src/biconnected.rs index ee91e5f6..4f1ae7ed 100644 --- a/crates/grust-algorithms/src/biconnected.rs +++ b/crates/grust-algorithms/src/biconnected.rs @@ -153,7 +153,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result { let arc = *next; *next += 1; meter.tick(1)?; - let slot = adjacency.edge_slots.values[arc]; + let slot = adjacency.edge_slot(arc); let other = adjacency.targets.values[arc]; if slot == arrived_by || other == node { continue; diff --git a/crates/grust-algorithms/src/flow.rs b/crates/grust-algorithms/src/flow.rs index 1e4e741b..a2010883 100644 --- a/crates/grust-algorithms/src/flow.rs +++ b/crates/grust-algorithms/src/flow.rs @@ -146,7 +146,7 @@ pub fn max_flow(graph: &GraphProjection, source: &str, target: &str) -> Result) -> Result }); } let adjacency = graph.outgoing(); - // An unweighted projection has a probability of 1/out-degree on every arc, - // so the iteration can be computed by pulling into each target instead of + // The iteration can be computed by pulling into each target instead of // pushing out of each source. Pulling is what makes it parallel: targets own // their own sums, so no worker writes where another might read, and each sum - // stays in reverse-CSR order, so the result does not depend on the division - // of work. A weighted projection needs the source's weight total per arc and - // keeps the sequential push below. + // stays in in-arc order, so the result does not depend on the division of + // work. A weighted projection pulls too, because the projection's in-arc + // index carries each arc's weight; the push loop below stays as the + // sequential path and the oracle the pull is tested against. let workers = crate::parallel::workers( context, n.saturating_add(adjacency.targets.values.len()) .saturating_mul(2), ); - if let Some(workers) = workers - && !graph.is_weighted() - { + if let Some(workers) = workers { return pull(graph, options, &teleport.values, scores, workers); } let mut next = Buffer::filled(n, 0.0, context)?; @@ -215,15 +213,16 @@ pub fn pagerank(graph: &GraphProjection, options: PageRankOptions<'_>) -> Result }) } -/// The parallel iteration for an unweighted projection. +/// The parallel iteration, weighted or not. /// /// Each iteration computes, for every target `v`, -/// `next[v] = base * teleport[v] + damping * Σ score[u] / out_degree(u)` -/// over the in-arcs of `v`, which is the same distribution the push loop above -/// produces, summed in a different but equally fixed order. Every reduction over -/// nodes, the dangling mass and the residual, combines per-chunk partial sums in -/// chunk order, so the sequence of iterations and the final scores are identical -/// at one worker and at sixteen. +/// `next[v] = base * teleport[v] + damping * Σ score[u] · p(u → v)` +/// over the in-arcs of `v`, where `p` is the same arc probability the push loop +/// derives from the source's scaled weight total. It is therefore the same +/// distribution, summed in a different but equally fixed order. Every reduction +/// over nodes, the dangling mass and the residual, combines per-chunk partial +/// sums in fixed-size chunks in chunk order, so the sequence of iterations and +/// the final scores are identical at one worker and at sixteen. fn pull( graph: &GraphProjection, options: PageRankOptions<'_>, @@ -235,7 +234,61 @@ fn pull( let n = graph.node_count(); let adjacency = graph.outgoing(); let offsets = &adjacency.offsets.values; - let reverse = graph.reverse()?; + let reverse = graph.incoming()?; + // Per-source arc probability, exactly as the push loop derives it: each row + // is scaled by its largest weight before summing, so two arcs of f64::MAX + // still sum finitely, and a row of zero weight is dangling. + let weighted = graph.is_weighted(); + let mut scales = Buffer::indexed(n, 0.0f64, context)?; + let mut totals = Buffer::indexed(n, 0.0f64, context)?; + { + crate::parallel::for_each_chunk( + context, + workers, + &mut scales.values, + |first, slice, meter| { + for (index, scale) in slice.iter_mut().enumerate() { + let node = first + index; + let range = offsets[node]..offsets[node + 1]; + meter.charge(1 + range.len())?; + if !weighted { + *scale = if range.is_empty() { 0.0 } else { 1.0 }; + continue; + } + let mut largest = 0.0f64; + for arc in range { + largest = largest.max(adjacency.weight(arc)); + } + *scale = largest; + } + Ok(()) + }, + )?; + let scaled = &scales.values; + crate::parallel::for_each_chunk( + context, + workers, + &mut totals.values, + |first, slice, meter| { + for (index, total) in slice.iter_mut().enumerate() { + let node = first + index; + let range = offsets[node]..offsets[node + 1]; + meter.charge(range.len())?; + if scaled[node] <= 0.0 { + continue; + } + let mut sum = 0.0; + for arc in range { + sum += adjacency.weight(arc) / scaled[node]; + } + *total = sum; + } + Ok(()) + }, + )?; + } + let scales = &scales.values; + let totals = &totals.values; let mut next = Buffer::indexed(n, 0.0f64, context)?; let mut residual = f64::INFINITY; for iteration in 1..=options.max_iterations { @@ -248,8 +301,9 @@ fn pull( meter.charge(slice.len())?; let mut sum = 0.0; for (index, &score) in slice.iter().enumerate() { - let node = first + index; - if offsets[node + 1] == offsets[node] { + // A row of zero outgoing weight is dangling even where arcs + // exist, which is the push kernel's rule too. + if scales[first + index] <= 0.0 { sum += score; } } @@ -272,10 +326,13 @@ fn pull( let mut sum = 0.0; for arc in arcs { let source = reverse.targets.values[arc]; - let out_degree = offsets[source + 1] - offsets[source]; - // A node with no outgoing arc contributes through the - // dangling mass in `base`, never through an arc. - sum += scores_now[source] / out_degree as f64; + // A dangling row contributes through `base`, never + // through an arc; its total is zero and would divide. + if totals[source] <= 0.0 { + continue; + } + let probability = (reverse.weight(arc) / scales[source]) / totals[source]; + sum += scores_now[source] * probability; } let updated = base * teleport[node] + options.damping * sum; if !updated.is_finite() { diff --git a/crates/grust-algorithms/src/parallel.rs b/crates/grust-algorithms/src/parallel.rs index 67f64058..0969c16d 100644 --- a/crates/grust-algorithms/src/parallel.rs +++ b/crates/grust-algorithms/src/parallel.rs @@ -38,7 +38,6 @@ use grust_procedures::{ExecutionContext, Result}; - /// Work units below which kernels stay sequential. /// /// Measured on quegee (Xeon Platinum 8124M, 8 physical cores and 16 threads) @@ -303,7 +302,6 @@ fn pool(workers: usize) -> crate::Result> { Ok(pool) } - /// Tasks the caller's pool can run at once; 1 without the `parallel` feature. pub(crate) fn width() -> usize { #[cfg(feature = "parallel")] diff --git a/crates/grust-algorithms/src/projection.rs b/crates/grust-algorithms/src/projection.rs index 5a3398d0..6515a2ee 100644 --- a/crates/grust-algorithms/src/projection.rs +++ b/crates/grust-algorithms/src/projection.rs @@ -11,7 +11,7 @@ use crate::buffer::Buffer; mod adjacency; mod origin; -pub(crate) use adjacency::{Adjacency, ReverseTopology}; +pub(crate) use adjacency::Adjacency; pub use origin::{ProjectionRepresentation, ProjectionSelection}; /// Explicit orientation applied once while preparing topology. @@ -59,7 +59,6 @@ struct ProjectionData { node_by_id: HashMap, edges: Buffer, outgoing: Adjacency, - reverse: Mutex>>, incoming: Mutex>>, context: ExecutionContext, _retained: MemoryReservation, @@ -189,7 +188,6 @@ impl GraphProjection { node_by_id, edges, outgoing, - reverse: Mutex::new(None), incoming: Mutex::new(None), context: context.clone(), _retained: retained, @@ -267,20 +265,6 @@ impl GraphProjection { *cached = Some(Arc::clone(&incoming)); Ok(InArcs::Built(incoming)) } - pub(crate) fn reverse(&self) -> Result> { - self.inner.context.checkpoint()?; - let mut cached = self - .inner - .reverse - .lock() - .map_err(|_| ProcedureError::ResourceStatePoisoned)?; - if let Some(reverse) = &*cached { - return Ok(Arc::clone(reverse)); - } - let reverse = Arc::new(self.inner.outgoing.reversed(&self.inner.context)?); - *cached = Some(Arc::clone(&reverse)); - Ok(reverse) - } } /// In-arcs of a projection: its own rows when undirected, else a built transpose. diff --git a/crates/grust-algorithms/src/projection/adjacency.rs b/crates/grust-algorithms/src/projection/adjacency.rs index 3ba7aa02..0258e796 100644 --- a/crates/grust-algorithms/src/projection/adjacency.rs +++ b/crates/grust-algorithms/src/projection/adjacency.rs @@ -5,7 +5,10 @@ use super::*; pub(crate) struct Adjacency { pub(crate) offsets: Buffer, pub(crate) targets: Buffer, - pub(crate) edge_slots: Buffer, + /// Original edge position per arc. Present on a projection's own adjacency, + /// absent on a transpose, whose callers identify arcs by endpoint rather + /// than by edge and would otherwise pay eight bytes an arc to ignore. + edge_slots: Option>, pub(crate) weights: Option>, } @@ -43,7 +46,7 @@ impl Adjacency { let mut result = Self { offsets, targets: Buffer::filled(count, 0, context)?, - edge_slots: Buffer::filled(count, 0, context)?, + edge_slots: Some(Buffer::filled(count, 0, context)?), weights: weights .map(|_| Buffer::filled(count, 0.0, context)) .transpose()?, @@ -89,7 +92,9 @@ impl Adjacency { ) { let index = positions[source]; self.targets.values[index] = target; - self.edge_slots.values[index] = edge; + if let Some(slots) = &mut self.edge_slots { + slots.values[index] = edge; + } if let (Some(weights), Some(weight)) = (&mut self.weights, weight) { weights.values[index] = weight; } @@ -99,46 +104,34 @@ impl Adjacency { pub(crate) fn range(&self, node: usize) -> std::ops::Range { self.offsets.values[node]..self.offsets.values[node + 1] } + /// Original edge position of an arc, on an adjacency that carries them. + /// + /// # Panics + /// A transpose carries no edge slots. Every construction that a kernel + /// reaches through `outgoing()` fills them, so a panic here would mean a + /// kernel read a transpose as if it were the projection's own adjacency. + #[track_caller] + pub(crate) fn edge_slot(&self, arc: usize) -> usize { + self.edge_slots + .as_ref() + .expect("arc slots belong to a projection's own adjacency, not a transpose") + .values[arc] + } + pub(crate) fn weight(&self, arc: usize) -> f64 { self.weights .as_ref() .map_or(1.0, |weights| weights.values[arc]) } - pub(crate) fn reversed(&self, context: &ExecutionContext) -> Result { - let n = self.offsets.values.len() - 1; - let mut offsets = Buffer::filled(n + 1, 0usize, context)?; - for &target in &self.targets.values { - context.charge_work(1)?; - offsets.values[target + 1] += 1; - } - prefix(&mut offsets.values, context)?; - let mut reverse = ReverseTopology { - offsets, - targets: Buffer::filled(self.targets.values.len(), 0, context)?, - }; - let mut positions = Buffer::capacity(n, context)?; - for chunk in reverse.offsets.values[..n].chunks(1024) { - context.charge_work(chunk.len())?; - positions.values.extend_from_slice(chunk); - } - for source in 0..n { - context.charge_work(1)?; - for arc in self.range(source) { - context.charge_work(1)?; - let target = self.targets.values[arc]; - reverse.targets.values[positions.values[target]] = source; - positions.values[target] += 1; - } - } - Ok(reverse) - } -} - -impl Adjacency { /// The same arcs grouped by target: row `v` lists the sources of arcs into - /// `v`, each with its weight and original edge slot. Within a row, arcs keep - /// the order of their sources, so the result is a function of this CSR alone. + /// `v`, each with its weight. Within a row, arcs keep the order of their + /// sources, so the result is a function of this CSR alone. + /// + /// This is the projection's only reverse index. Reachability alone once had + /// a second, narrower one; one build that carries weights costs less than + /// two builds over the same arcs, and no caller of a transpose reads edge + /// slots, so those are left out rather than duplicated. pub(crate) fn transposed(&self, context: &ExecutionContext) -> Result { let n = self.offsets.values.len() - 1; let arcs = self.targets.values.len(); @@ -153,7 +146,6 @@ impl Adjacency { let mut positions = Buffer::capacity(n, context)?; positions.values.extend_from_slice(&offsets.values[..n]); let mut targets = Buffer::filled(arcs, 0usize, context)?; - let mut edge_slots = Buffer::filled(arcs, 0usize, context)?; let mut weights = self .weights .as_ref() @@ -166,7 +158,6 @@ impl Adjacency { let slot = positions.values[self.targets.values[arc]]; positions.values[self.targets.values[arc]] += 1; targets.values[slot] = source; - edge_slots.values[slot] = self.edge_slots.values[arc]; if let (Some(into), Some(from)) = (&mut weights, &self.weights) { into.values[slot] = from.values[arc]; } @@ -175,7 +166,7 @@ impl Adjacency { Ok(Adjacency { offsets, targets, - edge_slots, + edge_slots: None, weights, }) } @@ -190,15 +181,3 @@ fn prefix(offsets: &mut [usize], context: &ExecutionContext) -> Result<()> { } Ok(()) } - -/// SCC needs reverse reachability only. No duplicated weights or edge IDs. -pub(crate) struct ReverseTopology { - offsets: Buffer, - pub(crate) targets: Buffer, -} - -impl ReverseTopology { - pub(crate) fn range(&self, node: usize) -> std::ops::Range { - self.offsets.values[node]..self.offsets.values[node + 1] - } -} diff --git a/crates/grust-algorithms/src/shortest.rs b/crates/grust-algorithms/src/shortest.rs index c4a7cf1d..fd8484b9 100644 --- a/crates/grust-algorithms/src/shortest.rs +++ b/crates/grust-algorithms/src/shortest.rs @@ -224,7 +224,7 @@ fn run(graph: &GraphProjection, source: &str, paths: bool) -> Result<(Distances, if candidate < distances.values[next] { distances.values[next] = candidate; if let Some(parents) = &mut parents { - parents.values[next] = Some((node, adjacency.edge_slots.values[arc])); + parents.values[next] = Some((node, adjacency.edge_slot(arc))); } heap.improve(next, candidate, context)?; } diff --git a/crates/grust-algorithms/src/spanning.rs b/crates/grust-algorithms/src/spanning.rs index 17834da8..093b2a27 100644 --- a/crates/grust-algorithms/src/spanning.rs +++ b/crates/grust-algorithms/src/spanning.rs @@ -118,7 +118,7 @@ pub fn spanning_tree( let range = adjacency.range(node); meter.tick(1 + range.len())?; for arc in range { - weight.values[adjacency.edge_slots.values[arc]] = adjacency.weight(arc); + weight.values[adjacency.edge_slot(arc)] = adjacency.weight(arc); } } } diff --git a/crates/grust-algorithms/src/traversal.rs b/crates/grust-algorithms/src/traversal.rs index d4c8b815..78273099 100644 --- a/crates/grust-algorithms/src/traversal.rs +++ b/crates/grust-algorithms/src/traversal.rs @@ -399,7 +399,9 @@ pub fn strongly_connected_components(graph: &GraphProjection) -> Result ExecutionContext { /// A deterministic graph: a permutation cycle so every node has a successor, /// plus pseudo-random chords from a fixed linear congruential sequence, so the /// degree distribution is skewed enough that a naive split would be unbalanced. +/// The same graph on an execution that never asked for threads, which is what +/// selects the sequential kernels. +fn sequential_graph(weighted: bool) -> GraphProjection { + build( + ExecutionContext::new(ExecutionLimits { + memory_bytes: 1 << 30, + work_units: usize::MAX, + batch_rows: 8192, + deadline: None, + }) + .expect("valid limits"), + weighted, + ) +} + fn graph(workers: usize, weighted: bool) -> GraphProjection { - let context = context(workers); + build(context(workers), weighted) +} + +fn build(context: ExecutionContext, weighted: bool) -> GraphProjection { let nodes = (0..NODES).map(|id| id.to_string().into()).collect(); let mut edges = Vec::with_capacity(EDGES); let mut state = 0x2545_F491_4F6C_DD1Du64; @@ -117,14 +135,35 @@ fn pagerank_agrees_with_the_sequential_push_and_does_not_move_with_threads() { } #[test] -fn pagerank_keeps_the_sequential_path_for_weighted_projections() { - let options = PageRankOptions::default(); - let sequential = pagerank(&graph(1, true), options).expect("sequential pagerank"); - let parallel = pagerank(&graph(16, true), options).expect("pagerank at sixteen threads"); - // Weighted projections take the push kernel on both paths, so the numbers - // are identical rather than merely close. - assert_eq!(parallel.values(), sequential.values()); - assert_eq!(parallel.iterations(), sequential.iterations()); +fn weighted_pagerank_pulls_and_agrees_with_the_sequential_push() { + let options = PageRankOptions { + tolerance: 1e-10, + max_iterations: 100, + ..Default::default() + }; + // A weighted projection pulls as well, over in-arcs that carry their weight. + let sequential = pagerank(&sequential_graph(true), options).expect("sequential pagerank"); + let one = pagerank(&graph(1, true), options).expect("pagerank at one worker"); + let sixteen = pagerank(&graph(16, true), options).expect("pagerank at sixteen threads"); + for (node, (&pushed, &pulled)) in sequential.values().iter().zip(sixteen.values()).enumerate() { + assert!( + (pushed - pulled).abs() <= 1e-12 + 1e-9 * pushed.abs(), + "node {node}: push {pushed} against pull {pulled}" + ); + } + let total: f64 = sixteen.values().iter().sum(); + assert!((total - 1.0).abs() < 1e-9, "mass {total}"); + // And the pull is the same number at one worker as at sixteen, bit for bit. + let bits = |result: &grust_algorithms::PageRank| -> Vec { + result + .values() + .iter() + .map(|score| score.to_bits()) + .collect() + }; + assert_eq!(bits(&one), bits(&sixteen)); + assert_eq!(one.iterations(), sixteen.iterations()); + assert_eq!(one.residual().to_bits(), sixteen.residual().to_bits()); } #[test] diff --git a/crates/grust-procedures/src/resources.rs b/crates/grust-procedures/src/resources.rs index 328cbac2..5bc80e87 100644 --- a/crates/grust-procedures/src/resources.rs +++ b/crates/grust-procedures/src/resources.rs @@ -465,6 +465,19 @@ impl WorkMeter { /// could be refused because of who else exists — the failure the contract /// test `an_idle_meters_unspent_block_does_not_refuse_work_that_fits` /// reproduces. + /// + /// One window remains, and a caller should know its shape. A grant being + /// moved inside another meter's own admission is held in that meter's stack + /// rather than in its balance, so it cannot be reclaimed; three attempts + /// cover the moment, and if all three lose, the charge reports + /// `BudgetExceeded` although the work would have fitted. The result is a + /// query that fails rather than a budget that is exceeded, it needs several + /// meters admitting in the same instant with the budget nearly gone, and + /// `many_meters_racing_for_the_last_of_a_budget_that_exactly_fits` provokes + /// exactly that: two hundred rounds of sixteen meters starting together on + /// a budget equal to their work, with no refusal observed. A caller that + /// cannot tolerate even a rare spurious refusal should leave one block per + /// worker of headroom in the budget. #[cold] fn admit(&mut self, units: usize) -> Result<()> { self.context.check_state(DeadlineCheck::Sampled)?; diff --git a/crates/grust-procedures/tests/contracts/parallel.rs b/crates/grust-procedures/tests/contracts/parallel.rs index b33f4c7f..2673750a 100644 --- a/crates/grust-procedures/tests/contracts/parallel.rs +++ b/crates/grust-procedures/tests/contracts/parallel.rs @@ -261,3 +261,49 @@ fn skewed_work_that_exactly_fits_succeeds_at_sixteen_threads() { ); assert_eq!(execution.usage().expect("usage").work_units, total); } + +/// The reclaim path retries three times, because a grant held inside another +/// meter's own admission cannot be taken back. That window is narrow but real, +/// so this provokes it: many meters, all charging at once, against a budget +/// equal to the work, repeated enough times that a rare loss would show. +#[test] +fn many_meters_racing_for_the_last_of_a_budget_that_exactly_fits() { + const ROUNDS: usize = 200; + const PER_WORKER: usize = 4 * WORK_BLOCK_UNITS; + let mut refusals = 0; + for _ in 0..ROUNDS { + let total = WORKERS * PER_WORKER; + let execution = ExecutionContext::new(limits(total, None)).expect("valid limits"); + let failures = AtomicUsize::new(0); + let start = std::sync::Barrier::new(WORKERS); + std::thread::scope(|scope| { + for _ in 0..WORKERS { + let (execution, failures, start) = (execution.clone(), &failures, &start); + scope.spawn(move || { + let mut meter = execution.work_meter(); + // Charge at the same moment, so admissions collide. + start.wait(); + for _ in 0..PER_WORKER { + if meter.charge(1).is_err() { + failures.fetch_add(1, Ordering::Relaxed); + return; + } + } + }); + } + }); + let failed = failures.load(Ordering::Relaxed); + refusals += failed; + if failed == 0 { + assert_eq!( + execution.usage().expect("usage").work_units, + total, + "usage should account for exactly the work performed" + ); + } + } + assert_eq!( + refusals, 0, + "{refusals} refusals of work that fits, over {ROUNDS} rounds of {WORKERS} racing meters" + ); +} From cec897f8759214e272362264eec2f3a4eeff493e Mon Sep 17 00:00:00 2001 From: Alexy Khrabrov Date: Sat, 19 Sep 2026 23:48:06 -0700 Subject: [PATCH 8/8] Decide a work-budget refusal under exclusion, not by retrying many_meters_racing_for_the_last_of_a_budget_that_exactly_fits failed in 55 of 80 runs on a ten-core host, and the skewed sixteen-thread contract in 7 of 80. Meters that ran out together each moved their balance onto their own stack, found the others' balances empty, reclaimed nothing, and returned BudgetExceeded on the first attempt; the three retries never ran. A block is invisible between reaching the shared counter and reaching its meter's balance. Admitting a block now holds the grant registry shared across that moment, and a meter about to be refused holds it exclusively, so every unspent unit is in some balance when it reclaims and the refusal is the one a single thread would meet. 0 failures in 360 runs, debug and release. Shared holders never wait for each other; on this host the charge loop and PageRank, components and breadth-first search at ten workers are within run-to-run noise of the version without exclusion. Co-Authored-By: Claude Fable 5.1 --- crates/grust-procedures/src/resources.rs | 141 ++++++++---------- .../tests/contracts/parallel.rs | 11 +- 2 files changed, 73 insertions(+), 79 deletions(-) diff --git a/crates/grust-procedures/src/resources.rs b/crates/grust-procedures/src/resources.rs index 5bc80e87..afc2a3bc 100644 --- a/crates/grust-procedures/src/resources.rs +++ b/crates/grust-procedures/src/resources.rs @@ -1,7 +1,7 @@ //! Shared cooperative limits with reservations retained by buffer owners. use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, RwLock}; use std::time::Instant; use crate::{ProcedureError, Result}; @@ -58,9 +58,10 @@ struct Shared { cancelled: AtomicBool, charges_since_deadline_read: AtomicUsize, /// Unspent balances of the live work meters, so a meter that runs out can - /// take back what idle ones hold. Touched only when a meter is created, - /// dropped, or refused, never on a charge. - grants: Mutex>>, + /// take back what idle ones hold. Never touched on a charge. A meter + /// admitting a block holds it shared; creating, dropping and refusing hold + /// it exclusively. See [`WorkMeter`]'s admission for why. + grants: RwLock>>, state: Mutex, } @@ -102,7 +103,7 @@ impl ExecutionContext { work_units: AtomicUsize::new(0), cancelled: AtomicBool::new(false), charges_since_deadline_read: AtomicUsize::new(0), - grants: Mutex::new(Vec::new()), + grants: RwLock::new(Vec::new()), state: Mutex::new(State::default()), }))) } @@ -197,7 +198,7 @@ impl ExecutionContext { // chunk of work, so this lock is taken a handful of times per region. self.0 .grants - .lock() + .write() .unwrap_or_else(|poison| poison.into_inner()) .push(Arc::clone(&balance)); WorkMeter { @@ -206,31 +207,11 @@ impl ExecutionContext { } } - /// Return every other live meter's unspent grant to the budget, and report - /// how much came back. - fn reclaim_grants(&self, own: &Arc) -> usize { - let grants = self - .0 - .grants - .lock() - .unwrap_or_else(|poison| poison.into_inner()); - let mut reclaimed = 0usize; - for balance in grants.iter() { - if Arc::ptr_eq(balance, own) { - continue; - } - reclaimed = reclaimed.saturating_add(balance.swap(0, Ordering::Relaxed)); - } - drop(grants); - self.refund_work(reclaimed); - reclaimed - } - fn release_meter(&self, own: &Arc) { self.refund_work(own.swap(0, Ordering::Relaxed)); self.0 .grants - .lock() + .write() .unwrap_or_else(|poison| poison.into_inner()) .retain(|balance| !Arc::ptr_eq(balance, own)); } @@ -457,7 +438,7 @@ impl WorkMeter { } /// Admit a block; failing that, exactly what was asked for; failing that, - /// take back every idle grant and try once more. + /// take back every idle grant and ask once more. /// /// The last step is what keeps admission independent of the number of /// meters. Without it, a block granted to a meter that has stopped charging @@ -466,59 +447,69 @@ impl WorkMeter { /// test `an_idle_meters_unspent_block_does_not_refuse_work_that_fits` /// reproduces. /// - /// One window remains, and a caller should know its shape. A grant being - /// moved inside another meter's own admission is held in that meter's stack - /// rather than in its balance, so it cannot be reclaimed; three attempts - /// cover the moment, and if all three lose, the charge reports - /// `BudgetExceeded` although the work would have fitted. The result is a - /// query that fails rather than a budget that is exceeded, it needs several - /// meters admitting in the same instant with the budget nearly gone, and - /// `many_meters_racing_for_the_last_of_a_budget_that_exactly_fits` provokes - /// exactly that: two hundred rounds of sixteen meters starting together on - /// a budget equal to their work, with no refusal observed. A caller that - /// cannot tolerate even a rare spurious refusal should leave one block per - /// worker of headroom in the budget. + /// **Why a refusal here is exact, not merely unlikely.** A refusal is + /// right only if every admitted, unspent unit has been found and handed + /// back first. Units are invisible for one moment: after a meter's block + /// has been added to the shared counter and before it appears in that + /// meter's balance. So a meter admitting a block holds the registry + /// *shared* across exactly that moment, and a meter about to be refused + /// holds it *exclusively*. Under the exclusive hold no block is in + /// transit, every unspent unit sits in some balance, reclaiming empties + /// them all, and the counter then holds only work performed. The refusal + /// that follows is the one a single thread would have met at the same unit. + /// + /// An earlier version retried three times with no exclusion and refused + /// work that fitted in most runs of + /// `many_meters_racing_for_the_last_of_a_budget_that_exactly_fits` on a + /// ten-core host: meters that ran out together each emptied their own + /// balance onto their stack, found the others' empty, and gave up. + /// + /// Shared holders never wait for each other, so while the budget has room + /// this costs one uncontended acquisition per [`WORK_BLOCK_UNITS`] per + /// worker. Exclusive holds happen only when a block no longer fits, which + /// is the end of the budget, and when a meter is created or dropped. #[cold] fn admit(&mut self, units: usize) -> Result<()> { self.context.check_state(DeadlineCheck::Sampled)?; - // Three attempts: grant, exact, and one more of each after reclaiming. - // A grant in transit inside another meter's own admission cannot be - // reclaimed, so a retry covers that window instead of a lock. - for attempt in 0..3 { - let held = self.balance.swap(0, Ordering::Relaxed); - let needed = units.saturating_sub(held); - let block = needed.saturating_add(WORK_BLOCK_UNITS); - // A block on top of the need, while the budget has room for one. - // The only failure `admit_work` reports is an exceeded budget, which - // is exactly the case the next attempt handles, so it is discarded - // here rather than returned. - if block > needed && self.context.admit_work(block).is_ok() { - self.balance - .fetch_add(held + block - units, Ordering::Relaxed); + // A block on top of the need, while the budget has room for one. What + // this meter already holds stays in its balance, where it can be seen. + // The only failure `admit_work` reports is an exceeded budget, which + // the exclusive path handles, so it is discarded here. + let block = units.saturating_add(WORK_BLOCK_UNITS); + if block > units { + let shared = self + .context + .0 + .grants + .read() + .unwrap_or_else(|poison| poison.into_inner()); + if self.context.admit_work(block).is_ok() { + self.balance.fetch_add(block - units, Ordering::Relaxed); return Ok(()); } - // Exactly what this charge needs, so the budget fails at the unit - // that exceeds it rather than a block early. - match self.context.admit_work(needed) { - Ok(()) => { - self.balance - .fetch_add(held + needed - units, Ordering::Relaxed); - return Ok(()); - } - Err(exceeded) => { - // Give back this meter's own idle remainder first, then take - // back what other idle meters hold, and try again. - self.context.refund_work(held); - if attempt == 2 || self.context.reclaim_grants(&self.balance) == 0 { - return Err(exceeded); - } - } - } + drop(shared); } - Err(ProcedureError::BudgetExceeded { - resource: "work", - limit: self.context.0.limits.work_units, - }) + let grants = self + .context + .0 + .grants + .write() + .unwrap_or_else(|poison| poison.into_inner()); + // Hand back what this meter holds and ask for exactly the charge, so + // the budget fails at the unit that exceeds it rather than a block early. + self.context + .refund_work(self.balance.swap(0, Ordering::Relaxed)); + if self.context.admit_work(units).is_ok() { + return Ok(()); + } + // Take back what idle meters hold. A meter charging from its balance at + // this moment loses the exchange, finds too little, and waits its turn. + let mut reclaimed = 0usize; + for balance in grants.iter() { + reclaimed = reclaimed.saturating_add(balance.swap(0, Ordering::Relaxed)); + } + self.context.refund_work(reclaimed); + self.context.admit_work(units) } /// Shared execution state, for an exact poll or a nested reservation. diff --git a/crates/grust-procedures/tests/contracts/parallel.rs b/crates/grust-procedures/tests/contracts/parallel.rs index 2673750a..27675989 100644 --- a/crates/grust-procedures/tests/contracts/parallel.rs +++ b/crates/grust-procedures/tests/contracts/parallel.rs @@ -262,10 +262,13 @@ fn skewed_work_that_exactly_fits_succeeds_at_sixteen_threads() { assert_eq!(execution.usage().expect("usage").work_units, total); } -/// The reclaim path retries three times, because a grant held inside another -/// meter's own admission cannot be taken back. That window is narrow but real, -/// so this provokes it: many meters, all charging at once, against a budget -/// equal to the work, repeated enough times that a rare loss would show. +/// A block is invisible for a moment, after it reaches the shared counter and +/// before it reaches its meter's balance. A meter refused in that moment must +/// not give up on work that fits, so a refusal is decided with the registry held +/// exclusively and a block is admitted with it held shared. This provokes the +/// moment: many meters, all running out at once, against a budget equal to the +/// work. With retries in place of exclusion this failed in most runs on a +/// ten-core host and in none on a sixteen-thread one, so run it where it bites. #[test] fn many_meters_racing_for_the_last_of_a_budget_that_exactly_fits() { const ROUNDS: usize = 200;