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/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..85f7a723 --- /dev/null +++ b/crates/grust-algorithms/examples/scaling.rs @@ -0,0 +1,212 @@ +//! Time the parallel kernels against their sequential paths on a real graph. +//! +//! 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 +//! 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; + // 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 weighted = false; + 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()?, + "--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()), + } + } + + 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(), + weighted.then(|| { + pairs + .iter() + .enumerate() + .map(|(ordinal, _)| 1.0 + (ordinal % 9) as f64) + .collect() + }), + orientation, + &context, + )?; + let build = build.elapsed().as_secs_f64(); + for kernel in &kernels { + 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/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/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/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 converged: true, }); } - let mut next = Buffer::filled(n, 0.0, context)?; let adjacency = graph.outgoing(); + // 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 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 { + 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 +212,172 @@ pub fn pagerank(graph: &GraphProjection, options: PageRankOptions<'_>) -> Result converged: false, }) } + +/// The parallel iteration, weighted or not. +/// +/// Each iteration computes, for every target `v`, +/// `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<'_>, + 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.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 { + // 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() { + // 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; + } + } + 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]; + // 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() { + 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..0969c16d 100644 --- a/crates/grust-algorithms/src/parallel.rs +++ b/crates/grust-algorithms/src/parallel.rs @@ -1,11 +1,306 @@ -//! 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 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, +//! 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. +/// +/// 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 +/// 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 { + 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")] + { + context.concurrency_requested() + } + #[cfg(not(feature = "parallel"))] + { + let _ = context; + None + } +} + +/// 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); + } + 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( + 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::*; + // 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_init( + || context.work_meter(), + |meter, (index, slice)| body(index * chunk, slice, 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_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.max(1); + 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_init( + || context.work_meter(), + |meter, (index, slice)| body(index * chunk, slice, meter), + ) + .collect() + }) +} + +#[cfg(not(feature = "parallel"))] +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.max(1); + let mut meter = context.work_meter(); + items + .chunks(chunk) + .enumerate() + .map(|(index, slice)| body(index * chunk, slice, &mut meter)) + .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. +/// +/// 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; + 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); + if pools.len() >= MAX_CACHED_POOLS { + pools.clear(); + } + 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/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/statistics.rs b/crates/grust-algorithms/src/statistics.rs index f05970e4..5af14751 100644 --- a/crates/grust-algorithms/src/statistics.rs +++ b/crates/grust-algorithms/src/statistics.rs @@ -66,12 +66,40 @@ 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) => { + // 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 + .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 60d549d7..78273099 100644 --- a/crates/grust-algorithms/src/traversal.rs +++ b/crates/grust-algorithms/src/traversal.rs @@ -67,17 +67,35 @@ fn bfs_sources<'a>( ) -> Result { let context = graph.execution(); context.checkpoint()?; + let adjacency = graph.outgoing(); + 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(); + { + 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 +113,118 @@ 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); + // 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 + }; + // A frontier's expansion writes each discovered node once, so its + // chunking cannot change the answer and may follow the worker count. + let chunk = crate::parallel::chunk_len(frontier.len(), level_workers); + let lists = crate::parallel::map_chunks_sized( + context, + level_workers, + &frontier, + chunk, + |_, 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 +275,97 @@ 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; + // 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 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 { @@ -183,7 +399,9 @@ pub fn strongly_connected_components(graph: &GraphProjection) -> Result 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. +/// 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 { + 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; + 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 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] +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"); + } +} + +// --- 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/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..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}; @@ -53,12 +53,25 @@ const DEADLINE_SAMPLE_UNITS: usize = 1024; #[derive(Debug)] struct Shared { limits: ExecutionLimits, + concurrency: Option, work_units: AtomicUsize, 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. 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, } +/// 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,13 +99,57 @@ impl ExecutionContext { } Ok(Self(Arc::new(Shared { limits, + concurrency: None, work_units: AtomicUsize::new(0), cancelled: AtomicBool::new(false), charges_since_deadline_read: AtomicUsize::new(0), + grants: RwLock::new(Vec::new()), state: Mutex::new(State::default()), }))) } + /// 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 + /// 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 = 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 + } + /// Read immutable limits. pub fn limits(&self) -> ExecutionLimits { self.0.limits @@ -122,9 +179,50 @@ 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 { + 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 + .write() + .unwrap_or_else(|poison| poison.into_inner()) + .push(Arc::clone(&balance)); + WorkMeter { + context: self.clone(), + balance, + } + } + + fn release_meter(&self, own: &Arc) { + self.refund_work(own.swap(0, Ordering::Relaxed)); + self.0 + .grants + .write() + .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. 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 +247,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 +299,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 +400,143 @@ 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, + /// 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 { + /// 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()?; + // 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; failing that, exactly what was asked for; failing that, + /// 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 + /// 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. + /// + /// **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)?; + // 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(()); + } + drop(shared); + } + 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. + 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.release_meter(&self.balance); + } +} + /// 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..27675989 --- /dev/null +++ b/crates/grust-procedures/tests/contracts/parallel.rs @@ -0,0 +1,312 @@ +//! 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, + "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() + ); +} + +// --- 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); +} + +/// 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; + 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" + ); +}