From 6003dc5180d30bd9c31b6362b345580da69afbb1 Mon Sep 17 00:00:00 2001 From: Alexy Khrabrov Date: Sat, 19 Sep 2026 23:54:37 -0700 Subject: [PATCH 1/2] Move the catalog kernels onto WorkMeter and the execution's concurrency Step 5 of the order in codex-to-codex.md. Meter and width() are gone. The partition helpers take a worker count read from the execution, run on the calling thread at one worker without touching rayon, and otherwise run inside the bounded pool for that count. Per-block workspaces own one meter each, because creating a meter registers it under a lock. The determinism tests ask for workers through the context instead of installing a pool, and the triangle fixture grew to clear the sequential floor it would otherwise never leave. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 10 ++ crates/grust-algorithms/src/betweenness.rs | 32 +++--- crates/grust-algorithms/src/biconnected.rs | 12 +- crates/grust-algorithms/src/closeness.rs | 25 ++-- crates/grust-algorithms/src/fastrp.rs | 29 +++-- crates/grust-algorithms/src/flow.rs | 20 ++-- crates/grust-algorithms/src/kcore.rs | 10 +- .../grust-algorithms/src/label_propagation.rs | 10 +- crates/grust-algorithms/src/leiden.rs | 15 +-- crates/grust-algorithms/src/lib.rs | 1 - crates/grust-algorithms/src/louvain.rs | 39 +++---- crates/grust-algorithms/src/meter.rs | 42 ------- crates/grust-algorithms/src/parallel.rs | 107 ++++++++++-------- crates/grust-algorithms/src/similarity.rs | 37 +++--- crates/grust-algorithms/src/spanning.rs | 21 ++-- crates/grust-algorithms/src/spectral.rs | 25 ++-- crates/grust-algorithms/src/triangles.rs | 28 +++-- crates/grust-algorithms/tests/betweenness.rs | 32 +++--- crates/grust-algorithms/tests/closeness.rs | 44 ++++--- crates/grust-algorithms/tests/fastrp.rs | 36 +++--- .../tests/label_propagation.rs | 36 +++--- crates/grust-algorithms/tests/leiden.rs | 32 +++--- crates/grust-algorithms/tests/louvain.rs | 28 ++--- crates/grust-algorithms/tests/similarity.rs | 37 +++--- crates/grust-algorithms/tests/spanning.rs | 24 ++-- crates/grust-algorithms/tests/spectral.rs | 72 ++++++------ crates/grust-algorithms/tests/triangles.rs | 25 ++-- docs/goals/graph-analytics-catalog.md | 22 ++-- 28 files changed, 392 insertions(+), 459 deletions(-) delete mode 100644 crates/grust-algorithms/src/meter.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b004dbc..91d152f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,16 @@ reconstructed from Git history, release commits, and the shipped docs. ## Unreleased +- The catalog kernels take their worker count from the execution + (`ExecutionContext::with_concurrency`) instead of from whatever rayon pool the + caller had installed, and charge work through `WorkMeter`. **Behaviour + change:** a caller that does not ask for concurrency now runs every kernel on + its own thread and starts no pool; before, a kernel called outside a pool + spread across every core of the machine. Results are unchanged and remain + bit-identical at any worker count. Below the shared floor of 16,384 work units + a kernel stays on one worker whatever was asked for. A budget is now never + exceeded by work already done: the first accounting layer admitted work after + performing it and could overshoot by a block per worker. - Add **FastRP node embeddings**: `fast_rp` and `grust.algorithms.fastRP`, with `embeddingDimension`, `iterationWeights`, `nodeSelfInfluence`, `normalizationStrength` and `seed`. Each node draws a very sparse random diff --git a/crates/grust-algorithms/src/betweenness.rs b/crates/grust-algorithms/src/betweenness.rs index 58704157..5551364f 100644 --- a/crates/grust-algorithms/src/betweenness.rs +++ b/crates/grust-algorithms/src/betweenness.rs @@ -3,7 +3,6 @@ use crate::{ AlgorithmError, GraphProjection, Orientation, Result, buffer::Buffer, - meter::Meter, parallel, random, shortest::MinHeap, table::{NodeColumn, NodeTable}, @@ -46,7 +45,7 @@ impl Betweenness { } /// Sources per block. Fixed, so which sums are formed, and in what order, does -/// not depend on the pool width. +/// not depend on the worker count. const BLOCK: usize = 64; /// For every node `v`, the sum over ordered pairs `s != v != t` of the share of @@ -65,9 +64,9 @@ const BLOCK: usize = 64; /// otherwise. A zero weight makes the number of shortest paths ill-defined and /// is rejected. /// -/// Sources run in blocks of 64 on the caller's rayon pool and are merged in -/// block order; scores, and the work charged, are identical at any pool width. -/// Memory held at once does grow with the pool: one workspace per running block. +/// Sources run in blocks of 64 on as many workers as the execution asked for and are merged in +/// block order; scores, and the work charged, are identical at any worker count. +/// Memory held at once does grow with the workers: one workspace per running block. pub fn betweenness(graph: &GraphProjection, options: BetweennessOptions) -> Result { let context = graph.execution(); context.checkpoint()?; @@ -80,16 +79,15 @@ pub fn betweenness(graph: &GraphProjection, options: BetweennessOptions) -> Resu } let weighted = adjacency.weights.is_some(); if let Some(weights) = &adjacency.weights { - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); for &weight in &weights.values { - meter.tick(1)?; + meter.charge(1)?; if weight == 0.0 { return Err(AlgorithmError::InvalidArguments( "betweenness needs positive weights: a zero weight makes the number of shortest paths ill-defined".into(), )); } } - meter.flush()?; } // The sources, ascending, so an exhaustive sample is the exact run bit for bit. @@ -107,7 +105,12 @@ pub fn betweenness(graph: &GraphProjection, options: BetweennessOptions) -> Resu let source_count = sources.values.len(); let mut scores = Buffer::filled(n, 0.0f64, context)?; + let workers = parallel::concurrency( + context, + source_count.saturating_mul(n.saturating_add(adjacency.targets.values.len())), + ); parallel::ordered_blocks( + workers, source_count.div_ceil(BLOCK), |block| { let first = block * BLOCK; @@ -155,6 +158,9 @@ pub fn betweenness(graph: &GraphProjection, options: BetweennessOptions) -> Resu /// Per-block scratch: one single-source pass at a time, summed into `local`. struct Workspace { + /// One meter for the block: creating a meter registers it, which is too + /// much to pay per node. + meter: grust_procedures::WorkMeter, distance: Buffer, paths: Buffer, dependency: Buffer, @@ -166,6 +172,7 @@ struct Workspace { impl Workspace { fn new(n: usize, weighted: bool, context: &ExecutionContext) -> Result { Ok(Self { + meter: context.work_meter(), distance: Buffer::filled(n, f64::INFINITY, context)?, paths: Buffer::filled(n, 0.0, context)?, dependency: Buffer::filled(n, 0.0, context)?, @@ -182,7 +189,6 @@ impl Workspace { context: &ExecutionContext, ) -> Result<()> { let adjacency = graph.outgoing(); - let mut meter = Meter::new(context); self.order.values.clear(); self.distance.values[source] = 0.0; self.paths.values[source] = 1.0; @@ -193,7 +199,7 @@ impl Workspace { while let Some((cost, node)) = heap.pop(context)? { self.order.values.push(node); let range = adjacency.range(node); - meter.tick(1 + range.len())?; + self.meter.charge(1 + range.len())?; for arc in range { let next = adjacency.targets.values[arc]; let candidate = cost + adjacency.weight(arc); @@ -213,7 +219,7 @@ impl Workspace { let node = self.order.values[head]; head += 1; let range = adjacency.range(node); - meter.tick(1 + range.len())?; + self.meter.charge(1 + range.len())?; let candidate = self.distance.values[node] + 1.0; for arc in range { let next = adjacency.targets.values[arc]; @@ -232,7 +238,7 @@ impl Workspace { // share of that successor's paths that came through it. for &node in self.order.values.iter().rev() { let range = adjacency.range(node); - meter.tick(1 + range.len())?; + self.meter.charge(1 + range.len())?; for arc in range { let next = adjacency.targets.values[arc]; if next != node @@ -255,6 +261,6 @@ impl Workspace { self.paths.values[node] = 0.0; self.dependency.values[node] = 0.0; } - meter.flush() + Ok(()) } } diff --git a/crates/grust-algorithms/src/biconnected.rs b/crates/grust-algorithms/src/biconnected.rs index 4f1ae7ed..7796480c 100644 --- a/crates/grust-algorithms/src/biconnected.rs +++ b/crates/grust-algorithms/src/biconnected.rs @@ -4,7 +4,6 @@ use crate::{ AlgorithmError, GraphProjection, Orientation, Result, buffer::Buffer, - meter::Meter, table::{NodeColumn, NodeTable}, }; @@ -124,7 +123,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result { let n = graph.node_count(); let edges = graph.edges(); let adjacency = graph.outgoing(); - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); let mut discovered = Buffer::filled(n, UNSEEN, context)?; let mut low = Buffer::filled(n, UNSEEN, context)?; @@ -138,7 +137,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result { let mut clock = 0usize; for root in 0..n { - meter.tick(1)?; + meter.charge(1)?; if discovered.values[root] != UNSEEN { continue; } @@ -152,7 +151,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result { if *next < adjacency.range(node).end { let arc = *next; *next += 1; - meter.tick(1)?; + meter.charge(1)?; let slot = adjacency.edge_slot(arc); let other = adjacency.targets.values[arc]; if slot == arrived_by || other == node { @@ -196,7 +195,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result { "tree edge missing from its component".into(), ) })?; - meter.tick(pending.values.len() - start)?; + meter.charge(pending.values.len() - start)?; let name = pending.values[start..] .iter() .map(|&slot| edges[slot].ordinal) @@ -210,7 +209,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result { is_articulation.values[root] = root_children > 1; } - meter.tick(n + edges.len())?; + meter.charge(n + edges.len())?; let count = |flags: &[bool]| flags.iter().filter(|&&flag| flag).count(); let mut bridges = Buffer::capacity(count(&is_bridge.values), context)?; bridges @@ -220,7 +219,6 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result { articulation .values .extend((0..n).filter(|&node| is_articulation.values[node])); - meter.flush()?; Ok(Biconnectivity { graph: graph.clone(), components, diff --git a/crates/grust-algorithms/src/closeness.rs b/crates/grust-algorithms/src/closeness.rs index f6b659ff..01bb8ff7 100644 --- a/crates/grust-algorithms/src/closeness.rs +++ b/crates/grust-algorithms/src/closeness.rs @@ -3,7 +3,6 @@ use crate::{ AlgorithmError, GraphProjection, Result, buffer::Buffer, - meter::Meter, parallel, shortest::MinHeap, table::{NodeColumn, NodeTable}, @@ -68,7 +67,7 @@ impl DistanceCentrality { /// rejected, because a zero distance between distinct nodes has no reciprocal. /// Parallel edges and self-loops change no distance. /// -/// Scores and charged work are identical at any pool width; memory held at once +/// Scores and charged work are identical at any worker count; memory held at once /// is one workspace per running block of 64 sources. pub fn closeness(graph: &GraphProjection, options: ClosenessOptions) -> Result { let others = graph.node_count().saturating_sub(1) as f64; @@ -101,7 +100,7 @@ pub fn harmonic(graph: &GraphProjection, options: HarmonicOptions) -> Result, order: Buffer, heap: Option, @@ -163,6 +169,7 @@ struct Workspace { impl Workspace { fn new(n: usize, weighted: bool, context: &ExecutionContext) -> Result { Ok(Self { + meter: context.work_meter(), distance: Buffer::filled(n, f64::INFINITY, context)?, order: Buffer::capacity(n, context)?, heap: weighted.then(|| MinHeap::new(n, context)).transpose()?, @@ -178,7 +185,6 @@ impl Workspace { context: &ExecutionContext, ) -> Result<(usize, f64, f64)> { let adjacency = graph.outgoing(); - let mut meter = Meter::new(context); self.order.values.clear(); self.distance.values[source] = 0.0; if let Some(heap) = &mut self.heap { @@ -186,7 +192,7 @@ impl Workspace { while let Some((cost, node)) = heap.pop(context)? { self.order.values.push(node); let range = adjacency.range(node); - meter.tick(1 + range.len())?; + self.meter.charge(1 + range.len())?; for arc in range { let next = adjacency.targets.values[arc]; let candidate = cost + adjacency.weight(arc); @@ -203,7 +209,7 @@ impl Workspace { let node = self.order.values[head]; head += 1; let range = adjacency.range(node); - meter.tick(1 + range.len())?; + self.meter.charge(1 + range.len())?; for arc in range { let next = adjacency.targets.values[arc]; if self.distance.values[next].is_infinite() { @@ -222,7 +228,6 @@ impl Workspace { for &node in &self.order.values { self.distance.values[node] = f64::INFINITY; } - meter.flush()?; Ok((reached, distance_sum, reciprocal_sum)) } } diff --git a/crates/grust-algorithms/src/fastrp.rs b/crates/grust-algorithms/src/fastrp.rs index 61c9fdb9..1929c945 100644 --- a/crates/grust-algorithms/src/fastrp.rs +++ b/crates/grust-algorithms/src/fastrp.rs @@ -4,7 +4,6 @@ use crate::{ AlgorithmError, GraphProjection, Result, buffer::Buffer, - meter::Meter, parallel, random, table::{NodeColumn, NodeTable}, }; @@ -74,7 +73,7 @@ impl FastRp { } } -/// Nodes per chunk; fixed, so nothing depends on the pool width. +/// Nodes per chunk; fixed, so nothing depends on the worker count. const CHUNK: usize = 1024; /// Embed every node. @@ -83,7 +82,7 @@ const CHUNK: usize = 1024; /// with probabilities 1/6, 2/3, 1/6, scaled by `degree ^ normalization_strength` /// (a node of degree zero scales by zero unless the strength is zero). Entry /// `j` of node `v` is a pure function of `(seed, v, j)`, so neither the order -/// of work nor the pool can change it. +/// of work nor the worker count can change it. /// /// **Each round** replaces a node's vector with the weighted mean of the /// vectors of the nodes its arcs reach, so after `k` rounds a node reflects its @@ -128,11 +127,16 @@ pub fn fast_rp(graph: &GraphProjection, options: FastRpOptions<'_>) -> Result) -> Result) -> Result) -> Result Result 0.0 }; @@ -112,7 +111,7 @@ pub fn max_flow(graph: &GraphProjection, source: &str, target: &str) -> Result Result Result 0.0 && level.values[next] == UNSEEN { @@ -187,7 +186,7 @@ pub fn max_flow(graph: &GraphProjection, source: &str, target: &str) -> Result Result Result Result Result Result { } let n = graph.node_count(); let adjacency = graph.outgoing(); - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); // Degree without self-loops. Under the undirected arc contract a loop // appears once in its node's row, every other edge once in each endpoint's. @@ -63,7 +62,7 @@ pub fn k_core(graph: &GraphProjection) -> Result { let mut maximum = 0usize; for node in 0..n { let range = adjacency.range(node); - meter.tick(1 + range.len())?; + meter.charge(1 + range.len())?; let loops = adjacency.targets.values[range.clone()] .iter() .filter(|&&target| target == node) @@ -93,7 +92,7 @@ pub fn k_core(graph: &GraphProjection) -> Result { position.values[node] = slot; } } - meter.tick(n)?; + meter.charge(n)?; // Peel in nondecreasing degree. Removing `node` lowers each heavier // neighbour by one per arc, moving it one bin down by swapping it with the @@ -101,7 +100,7 @@ pub fn k_core(graph: &GraphProjection) -> Result { for index in 0..n { let node = order.values[index]; let range = adjacency.range(node); - meter.tick(1 + range.len())?; + meter.charge(1 + range.len())?; for arc in range { let other = adjacency.targets.values[arc]; if other == node || degree.values[other] <= degree.values[node] { @@ -121,7 +120,6 @@ pub fn k_core(graph: &GraphProjection) -> Result { degree.values[other] -= 1; } } - meter.flush()?; let mut cores = Buffer::capacity(n, context)?; let mut degeneracy = 0i64; diff --git a/crates/grust-algorithms/src/label_propagation.rs b/crates/grust-algorithms/src/label_propagation.rs index 290c20ff..eddf9d4d 100644 --- a/crates/grust-algorithms/src/label_propagation.rs +++ b/crates/grust-algorithms/src/label_propagation.rs @@ -3,7 +3,6 @@ use crate::{ AlgorithmError, GraphProjection, Result, buffer::Buffer, - meter::Meter, random, table::{NodeColumn, NodeTable, TableScalar}, }; @@ -99,20 +98,20 @@ pub fn label_propagation( // Weight per label for the node in hand, and which labels it touched. let mut support = Buffer::filled(n, 0.0f64, context)?; let mut touched = Buffer::::capacity(n, context)?; - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); let mut iterations = 0; let mut converged = false; while iterations < options.max_iterations && !converged { if let Some(seed) = options.seed { - meter.tick(n)?; + meter.charge(n)?; random::shuffle(&mut order.values, seed, iterations as u64); } iterations += 1; let mut changed = false; for &node in &order.values { let range = incoming.range(node); - meter.tick(1 + range.len())?; + meter.charge(1 + range.len())?; for arc in range { let weight = incoming.weight(arc); if weight == 0.0 { @@ -151,7 +150,7 @@ pub fn label_propagation( // Name each community by its smallest member: a label's first owner may // have left it. - meter.tick(2 * n)?; + meter.charge(2 * n)?; let mut smallest = Buffer::filled(n, usize::MAX, context)?; for (node, &label) in labels.values.iter().enumerate() { smallest.values[label] = smallest.values[label].min(node); @@ -159,7 +158,6 @@ pub fn label_propagation( for label in &mut labels.values { *label = smallest.values[*label]; } - meter.flush()?; Ok(LabelPropagation { graph: graph.clone(), communities: labels, diff --git a/crates/grust-algorithms/src/leiden.rs b/crates/grust-algorithms/src/leiden.rs index 4fca8e3e..fd2209e3 100644 --- a/crates/grust-algorithms/src/leiden.rs +++ b/crates/grust-algorithms/src/leiden.rs @@ -5,7 +5,6 @@ use crate::{ GraphProjection, Orientation, Result, buffer::Buffer, louvain::{self, Level, Louvain, LouvainOptions}, - meter::Meter, random, }; use grust_procedures::ExecutionContext; @@ -78,9 +77,9 @@ pub fn leiden(graph: &GraphProjection, options: LeidenOptions) -> Result break; } levels += 1; - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); for slot in &mut assignment.values { - meter.tick(1)?; + meter.charge(1)?; *slot = dense.values[*slot]; } // The next level's nodes are these groups; each starts in the community @@ -88,7 +87,7 @@ pub fn leiden(graph: &GraphProjection, options: LeidenOptions) -> Result let mut first_group = Buffer::filled(level.nodes(), usize::MAX, context)?; let mut next_initial = Buffer::filled(count, 0usize, context)?; for node in 0..level.nodes() { - meter.tick(1)?; + meter.charge(1)?; let community = outcome.community.values[node]; let group = dense.values[node]; if first_group.values[community] == usize::MAX { @@ -96,7 +95,6 @@ pub fn leiden(graph: &GraphProjection, options: LeidenOptions) -> Result } next_initial.values[group] = first_group.values[community]; } - meter.flush()?; level = louvain::coarsen(&level, &dense, count, directed, context)?; initial = Some(next_initial); } @@ -120,7 +118,7 @@ fn refine( context: &ExecutionContext, ) -> Result> { let n = level.nodes(); - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); let gamma = options.resolution; // Both matrix entries of an undirected arc, as in the move phase. let scale = if directed { 1.0 } else { 2.0 }; @@ -138,7 +136,7 @@ fn refine( let mut community_in = Buffer::filled(n, 0.0f64, context)?; for node in 0..n { let row = level.row(node); - meter.tick(1 + row.len() + in_rows(node).len())?; + meter.charge(1 + row.len() + in_rows(node).len())?; let own = level.inside.values[node]; let mut out = own; for arc in row { @@ -198,7 +196,7 @@ fn refine( for &node in &order.values { let home = community.values[node]; - meter.tick(1 + level.row(node).len() + in_rows(node).len())?; + meter.charge(1 + level.row(node).len() + in_rows(node).len())?; if !alone.values[node] || !well_connected( external.values[node], @@ -268,6 +266,5 @@ fn refine( } touched.values.clear(); } - meter.flush()?; Ok(group) } diff --git a/crates/grust-algorithms/src/lib.rs b/crates/grust-algorithms/src/lib.rs index ef18c5d4..4c367875 100644 --- a/crates/grust-algorithms/src/lib.rs +++ b/crates/grust-algorithms/src/lib.rs @@ -32,7 +32,6 @@ pub use leiden::{Leiden, LeidenOptions, leiden}; mod louvain; pub use kcore::{KCore, k_core}; pub use louvain::{Louvain, LouvainOptions, louvain}; -mod meter; mod ordering; mod parallel; pub use ordering::{NodeOrder, TopologicalOrder, depth_first, topological_sort}; diff --git a/crates/grust-algorithms/src/louvain.rs b/crates/grust-algorithms/src/louvain.rs index 4ac6652a..11e41de2 100644 --- a/crates/grust-algorithms/src/louvain.rs +++ b/crates/grust-algorithms/src/louvain.rs @@ -17,7 +17,6 @@ use crate::{ AlgorithmError, GraphProjection, Orientation, Result, buffer::Buffer, - meter::Meter, random, table::{NodeColumn, NodeTable, TableScalar}, }; @@ -134,7 +133,7 @@ impl Level { /// **Determinism.** Nodes are visited in row order, or in the order `seed` /// fixes; ties go to staying put, then to the community whose current id is /// smallest. Moves are applied one at a time, so the result does not depend on -/// the rayon pool: asynchronous parallel moves, which NetworKit's PLM uses, are +/// the worker count: asynchronous parallel moves, which NetworKit's PLM uses, are /// not reproducible, and reproducibility is the contract here. pub fn louvain(graph: &GraphProjection, options: LouvainOptions) -> Result { let context = graph.execution(); @@ -158,12 +157,11 @@ pub fn louvain(graph: &GraphProjection, options: LouvainOptions) -> Result Result { let n = offsets.values.len() - 1; let arcs = targets.values.len(); - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); let mut reversed = Buffer::filled(n + 1, 0usize, context)?; for &target in &targets.values { reversed.values[target + 1] += 1; @@ -347,7 +343,7 @@ fn transpose( let mut carried = Buffer::filled(arcs, 0.0f64, context)?; for source in 0..n { let range = offsets.values[source]..offsets.values[source + 1]; - meter.tick(1 + range.len())?; + meter.charge(1 + range.len())?; for arc in range { let slot = next.values[targets.values[arc]]; next.values[targets.values[arc]] += 1; @@ -355,7 +351,6 @@ fn transpose( carried.values[slot] = weights.values[arc]; } } - meter.flush()?; Ok((reversed, sources, carried)) } @@ -380,14 +375,14 @@ pub(super) fn move_nodes( context: &ExecutionContext, ) -> Result { let n = level.nodes(); - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); // Strengths, and the total M they sum to. let mut k_out = Buffer::capacity(n, context)?; let mut k_in = Buffer::capacity(n, context)?; for node in 0..n { let row = level.row(node); - meter.tick(1 + row.len())?; + meter.charge(1 + row.len())?; let own = level.inside.values[node]; k_out .values @@ -452,7 +447,7 @@ pub(super) fn move_nodes( seen.values[home] = true; touched.values.push(home); let row = level.row(node); - meter.tick(1 + row.len())?; + meter.charge(1 + row.len())?; for arc in row { let other = community.values[level.targets.values[arc]]; if !seen.values[other] { @@ -465,7 +460,7 @@ pub(super) fn move_nodes( } if let Some((offsets, sources, weights)) = &level.incoming { let row = offsets.values[node]..offsets.values[node + 1]; - meter.tick(row.len())?; + meter.charge(row.len())?; for arc in row { let other = community.values[sources.values[arc]]; if !seen.values[other] { @@ -519,7 +514,6 @@ pub(super) fn move_nodes( break; } } - meter.flush()?; Ok(Moves { community, moved, @@ -555,7 +549,7 @@ pub(super) fn coarsen( context: &ExecutionContext, ) -> Result { let n = level.nodes(); - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); // Members grouped by community, by counting sort. let mut start = Buffer::filled(count + 1, 0usize, context)?; @@ -573,7 +567,7 @@ pub(super) fn coarsen( members.values[next.values[community]] = node; next.values[community] += 1; } - meter.tick(n)?; + meter.charge(n)?; let mut offsets = Buffer::filled(count + 1, 0usize, context)?; let mut targets = Buffer::capacity(level.targets.values.len(), context)?; @@ -587,7 +581,7 @@ pub(super) fn coarsen( for &node in &members.values[start.values[community]..start.values[community + 1]] { inside.values[community] += level.inside.values[node]; let row = level.row(node); - meter.tick(1 + row.len())?; + meter.charge(1 + row.len())?; for arc in row { let other = dense.values[level.targets.values[arc]]; let weight = level.weights.values[arc]; @@ -613,7 +607,6 @@ pub(super) fn coarsen( } offsets.values[community + 1] = targets.values.len(); } - meter.flush()?; let incoming = directed .then(|| transpose(&offsets, &targets, &weights, context)) .transpose()?; diff --git a/crates/grust-algorithms/src/meter.rs b/crates/grust-algorithms/src/meter.rs deleted file mode 100644 index 6449337d..00000000 --- a/crates/grust-algorithms/src/meter.rs +++ /dev/null @@ -1,42 +0,0 @@ -//! Work accounting in blocks. A kernel's inner loop does a few nanoseconds of -//! work per arc; admitting each arc separately costs more than the arc. A meter -//! counts locally and admits at most every 1024 units, so budgets, cancellation -//! and deadlines are still observed promptly and totals stay exact. - -use grust_procedures::{ExecutionContext, Result}; - -const BLOCK: usize = 1024; - -pub(crate) struct Meter<'a> { - context: &'a ExecutionContext, - pending: usize, -} - -impl<'a> Meter<'a> { - pub(crate) fn new(context: &'a ExecutionContext) -> Self { - Self { - context, - pending: 0, - } - } - - /// Count `units` of work, admitting a block once enough has accumulated. - #[inline] - pub(crate) fn tick(&mut self, units: usize) -> Result<()> { - self.pending += units; - if self.pending >= BLOCK { - self.flush()?; - } - Ok(()) - } - - /// Admit whatever is still counted. Call before returning a result: work - /// left unadmitted would make a budget look larger than it is. - pub(crate) fn flush(&mut self) -> Result<()> { - if self.pending > 0 { - self.context - .charge_work(std::mem::take(&mut self.pending))?; - } - Ok(()) - } -} diff --git a/crates/grust-algorithms/src/parallel.rs b/crates/grust-algorithms/src/parallel.rs index 0969c16d..7e01e314 100644 --- a/crates/grust-algorithms/src/parallel.rs +++ b/crates/grust-algorithms/src/parallel.rs @@ -302,16 +302,19 @@ 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")] - { - rayon::current_num_threads().max(1) - } - #[cfg(not(feature = "parallel"))] - { - 1 - } +/// Workers for a kernel that has one implementation rather than a sequential +/// and a parallel one: what the execution asked for, above the shared floor, and +/// otherwise one. At one worker the helpers below run on the calling thread and +/// touch no pool, so a caller that never asked for threads never starts any. +pub(crate) fn concurrency(context: &ExecutionContext, units: usize) -> usize { + workers(context, units).unwrap_or(1).max(1) +} + +/// Run `body` inside the pool for `workers`. Callers take this path only above +/// one worker. +#[cfg(feature = "parallel")] +fn inside(workers: usize, body: impl FnOnce() -> Result + Send) -> Result { + pool(workers)?.install(body) } /// Split `0..weights.len()` into at most `parts` contiguous ranges of roughly @@ -341,52 +344,54 @@ pub(crate) fn balanced_ranges(weights: &[usize], parts: usize) -> Vec( + workers: usize, ranges: &[std::ops::Range], - task: impl Fn(std::ops::Range) -> Result + Sync, + task: impl Fn(std::ops::Range) -> Result + Sync + Send, ) -> Result> { #[cfg(feature = "parallel")] - { + if workers > 1 { use rayon::prelude::*; - ranges - .par_iter() - .map(|range| task(range.clone())) - .collect::>>() - } - #[cfg(not(feature = "parallel"))] - { - ranges.iter().map(|range| task(range.clone())).collect() + return inside(workers, || { + ranges + .par_iter() + .map(|range| task(range.clone())) + .collect::>>() + }); } + let _ = workers; + ranges.iter().map(|range| task(range.clone())).collect() } /// Run `task(block)` for `0..blocks` and hand each result to `merge` **in block -/// order**. Blocks run `width()` at a time, so at most that many results are +/// order**. Blocks run `workers` at a time, so at most that many results are /// alive at once; which blocks exist, what each computes and the order they are /// merged in do not depend on the pool, so neither does a floating-point sum /// built by `merge`. pub(crate) fn ordered_blocks( + workers: usize, blocks: usize, - task: impl Fn(usize) -> Result + Sync, + task: impl Fn(usize) -> Result + Sync + Send, mut merge: impl FnMut(usize, T) -> Result<()>, ) -> Result<()> { - let wave = width(); + let wave = workers.max(1); let mut first = 0; while first < blocks { let last = (first + wave).min(blocks); - let indices: Vec = (first..last).collect(); #[cfg(feature = "parallel")] - let results = { + let results = if wave > 1 { use rayon::prelude::*; - indices - .par_iter() - .map(|&block| task(block)) - .collect::>>()? + inside(wave, || { + (first..last) + .into_par_iter() + .map(&task) + .collect::>>() + })? + } else { + (first..last).map(&task).collect::>>()? }; #[cfg(not(feature = "parallel"))] - let results = indices - .iter() - .map(|&block| task(block)) - .collect::>>()?; - for (block, result) in indices.into_iter().zip(results) { + let results = (first..last).map(&task).collect::>>()?; + for (block, result) in (first..last).zip(results) { merge(block, result)?; } first = last; @@ -399,20 +404,23 @@ pub(crate) fn ordered_blocks( /// the caller folds over the returned vector groups its terms the same way at /// any pool width. pub(crate) fn for_chunks( + workers: usize, values: &mut [T], size: usize, - task: impl Fn(usize, &mut [T]) -> Result + Sync, + task: impl Fn(usize, &mut [T]) -> Result + Sync + Send, ) -> Result> { #[cfg(feature = "parallel")] - { + if workers > 1 { use rayon::prelude::*; - values - .par_chunks_mut(size) - .enumerate() - .map(|(index, chunk)| task(index * size, chunk)) - .collect() + return inside(workers, || { + values + .par_chunks_mut(size) + .enumerate() + .map(|(index, chunk)| task(index * size, chunk)) + .collect() + }); } - #[cfg(not(feature = "parallel"))] + let _ = workers; values .chunks_mut(size) .enumerate() @@ -423,16 +431,21 @@ pub(crate) fn for_chunks( /// Sort by a **total** order. The result is then the same whatever the pool /// does, so the parallel sort is safe to use where a result must not vary. pub(crate) fn sort_total( + workers: usize, values: &mut [T], - order: impl Fn(&T, &T) -> std::cmp::Ordering + Sync, -) { + order: impl Fn(&T, &T) -> std::cmp::Ordering + Sync + Send, +) -> Result<()> { #[cfg(feature = "parallel")] - { + if workers > 1 { use rayon::prelude::*; - values.par_sort_unstable_by(order); + return inside(workers, || { + values.par_sort_unstable_by(order); + Ok(()) + }); } - #[cfg(not(feature = "parallel"))] + let _ = workers; values.sort_unstable_by(order); + Ok(()) } #[cfg(test)] diff --git a/crates/grust-algorithms/src/similarity.rs b/crates/grust-algorithms/src/similarity.rs index 5c43aaa3..56c04ec1 100644 --- a/crates/grust-algorithms/src/similarity.rs +++ b/crates/grust-algorithms/src/similarity.rs @@ -3,7 +3,6 @@ use crate::{ AlgorithmError, GraphProjection, Result, buffer::Buffer, - meter::Meter, parallel, table::{NodeColumn, NodeTable}, }; @@ -104,7 +103,7 @@ impl NodeSimilarity { /// so the work budget is the guard; `upper_degree_cutoff` does not help there, /// as it bounds a node's own neighbours, not how many point at it. /// -/// Rows and charged work are identical at any pool width. +/// Rows and charged work are identical at any worker count. pub fn node_similarity( graph: &GraphProjection, options: NodeSimilarityOptions, @@ -133,7 +132,12 @@ pub fn node_similarity( let blocks = n.div_ceil(BLOCK); let mut chunks: Buffer> = Buffer::capacity(blocks, context)?; let mut total = 0usize; + // Cost is the number of (node, shared neighbour, node) triples; the entries + // squared over the nodes is its order of magnitude. + let entries = sets.entries.values.len(); + let workers = parallel::concurrency(context, entries.saturating_mul(entries / n.max(1) + 1)); parallel::ordered_blocks( + workers, blocks, |block| { let first = block * BLOCK; @@ -154,15 +158,15 @@ pub fn node_similarity( }, )?; - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); let mut rows = Buffer::capacity(total, context)?; for chunk in chunks.values.drain(..) { - meter.tick(chunk.values.len())?; + meter.charge(chunk.values.len())?; rows.values.extend_from_slice(&chunk.values); } drop(chunks); if options.top_n > 0 { - meter.tick(total.saturating_mul(total.max(2).ilog2() as usize))?; + meter.charge(total.saturating_mul(total.max(2).ilog2() as usize))?; rows.values.sort_unstable_by(|a, b| { b.similarity .total_cmp(&a.similarity) @@ -172,7 +176,7 @@ pub fn node_similarity( rows.values.truncate(options.top_n); } let count = rows.values.len(); - meter.tick(count)?; + meter.charge(count)?; let mut first = Buffer::capacity(count, context)?; let mut second = Buffer::capacity(count, context)?; let mut similarity = Buffer::capacity(count, context)?; @@ -181,7 +185,6 @@ pub fn node_similarity( second.values.push(row.second); similarity.values.push(row.similarity); } - meter.flush()?; Ok(NodeSimilarity { graph: graph.clone(), first, @@ -190,7 +193,7 @@ pub fn node_similarity( }) } -/// Nodes per block; fixed, so nothing depends on the pool width. +/// Nodes per block; fixed, so nothing depends on the worker count. const BLOCK: usize = 64; #[derive(Clone, Copy)] @@ -227,7 +230,7 @@ impl Sets { let adjacency = graph.outgoing(); let weighted = adjacency.weights.is_some(); let n = graph.node_count(); - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); let mut offsets = Buffer::capacity(n + 1, context)?; let mut entries: Buffer<(usize, f64)> = Buffer::capacity(adjacency.targets.values.len(), context)?; @@ -238,7 +241,7 @@ impl Sets { for node in 0..n { let range = adjacency.range(node); let degree = range.len(); - meter.tick(1 + degree.saturating_mul(degree.max(2).ilog2() as usize))?; + meter.charge(1 + degree.saturating_mul(degree.max(2).ilog2() as usize))?; let start = entries.values.len(); for arc in range { let target = adjacency.targets.values[arc]; @@ -286,7 +289,7 @@ impl Sets { for index in 1..=n { member_offsets.values[index] += member_offsets.values[index - 1]; } - meter.tick(n + entries.values.len())?; + meter.charge(n + entries.values.len())?; let mut members = Buffer::filled(member_offsets.values[n], (0usize, 0.0f64), context)?; let mut position = Buffer::capacity(n, context)?; position @@ -302,7 +305,6 @@ impl Sets { position.values[target] += 1; } } - meter.flush()?; Ok(Self { offsets, entries, @@ -315,6 +317,9 @@ impl Sets { } struct Workspace { + /// One meter for the block: creating a meter registers it, which is too + /// much to pay per node. + meter: grust_procedures::WorkMeter, /// `S` per candidate for the node in hand. shared: Buffer, seen: Buffer, @@ -326,6 +331,7 @@ struct Workspace { impl Workspace { fn new(n: usize, top_k: usize, context: &ExecutionContext) -> Result { Ok(Self { + meter: context.work_meter(), shared: Buffer::filled(n, 0.0, context)?, seen: Buffer::filled(n, false, context)?, candidates: Buffer::capacity(n, context)?, @@ -346,14 +352,13 @@ impl Workspace { if !sets.admitted.values[node] { return Ok(()); } - let mut meter = Meter::new(context); let cosine = options.metric == SimilarityMetric::Cosine; for &(target, weight) in &sets.entries.values[sets.offsets.values[node]..sets.offsets.values[node + 1]] { let members = &sets.members.values [sets.member_offsets.values[target]..sets.member_offsets.values[target + 1]]; - meter.tick(1 + members.len())?; + self.meter.charge(1 + members.len())?; for &(other, other_weight) in members { if other == node { continue; @@ -392,7 +397,7 @@ impl Workspace { } } let found = self.ranked.values.len(); - meter.tick( + self.meter.charge( self.candidates.values.len() + found.saturating_mul(found.max(2).ilog2() as usize), )?; self.candidates.values.clear(); @@ -419,6 +424,6 @@ impl Workspace { } self.rows.values.extend_from_slice(&self.ranked.values); self.ranked.values.clear(); - meter.flush() + Ok(()) } } diff --git a/crates/grust-algorithms/src/spanning.rs b/crates/grust-algorithms/src/spanning.rs index 093b2a27..3ebeb3b4 100644 --- a/crates/grust-algorithms/src/spanning.rs +++ b/crates/grust-algorithms/src/spanning.rs @@ -3,7 +3,6 @@ use crate::{ AlgorithmError, GraphProjection, Orientation, Result, buffer::Buffer, - meter::Meter, parallel, table::{NodeColumn, NodeTable, TableScalar}, }; @@ -92,7 +91,7 @@ impl SpanningForest { /// what a Prim run from that node would return. /// /// Kruskal with union by size and path halving. Only the sort runs on the -/// caller's rayon pool; its order is total, so the result does not depend on it. +/// execution's workers; its order is total, so the result does not depend on them. /// Defined on undirected graphs; project with `orientation: "undirected"`. pub fn spanning_tree( graph: &GraphProjection, @@ -109,14 +108,14 @@ pub fn spanning_tree( let n = graph.node_count(); let edges = graph.edges(); let adjacency = graph.outgoing(); - let mut meter = Meter::new(context); + let mut meter = context.work_meter(); // Weight per edge, read off either of its arcs. let mut weight = Buffer::filled(edges.len(), 1.0f64, context)?; if adjacency.weights.is_some() { for node in 0..n { let range = adjacency.range(node); - meter.tick(1 + range.len())?; + meter.charge(1 + range.len())?; for arc in range { weight.values[adjacency.edge_slot(arc)] = adjacency.weight(arc); } @@ -126,10 +125,11 @@ pub fn spanning_tree( let mut order = Buffer::capacity(edges.len(), context)?; order.values.extend(0..edges.len()); let count = edges.len(); - meter.tick(count.saturating_mul(count.max(2).ilog2() as usize))?; - meter.flush()?; + meter.charge(count.saturating_mul(count.max(2).ilog2() as usize))?; let maximum = options.objective == SpanningObjective::Maximum; - parallel::sort_total(&mut order.values, |&a: &usize, &b: &usize| { + let workers = + parallel::concurrency(context, count.saturating_mul(count.max(2).ilog2() as usize)); + parallel::sort_total(workers, &mut order.values, |&a: &usize, &b: &usize| { let by_weight = weight.values[a].total_cmp(&weight.values[b]); let by_weight = if maximum { by_weight.reverse() @@ -137,7 +137,7 @@ pub fn spanning_tree( by_weight }; by_weight.then(edges[a].ordinal.cmp(&edges[b].ordinal)) - }); + })?; let mut parent = Buffer::capacity(n, context)?; parent.values.extend(0..n); @@ -155,7 +155,7 @@ pub fn spanning_tree( if remaining == 0 { break; } - meter.tick(1)?; + meter.charge(1)?; let a = find(&mut parent.values, edges[slot].source); let b = find(&mut parent.values, edges[slot].target); if a == b { @@ -173,7 +173,7 @@ pub fn spanning_tree( } let wanted = source.map(|node| find(&mut parent.values, node)); - meter.tick(edges.len())?; + meter.charge(edges.len())?; let mut kept = Buffer::capacity(n.saturating_sub(1).min(edges.len()), context)?; let mut weights = Buffer::capacity(n.saturating_sub(1).min(edges.len()), context)?; let mut total = 0.0; @@ -186,7 +186,6 @@ pub fn spanning_tree( total += weight.values[slot]; } } - meter.flush()?; Ok(SpanningForest { graph: graph.clone(), edges: kept, diff --git a/crates/grust-algorithms/src/spectral.rs b/crates/grust-algorithms/src/spectral.rs index 4f57c2ea..0f7473c2 100644 --- a/crates/grust-algorithms/src/spectral.rs +++ b/crates/grust-algorithms/src/spectral.rs @@ -4,7 +4,6 @@ use crate::{ AlgorithmError, GraphProjection, Result, buffer::Buffer, - meter::Meter, parallel, projection::Adjacency, table::{NodeColumn, NodeTable, TableScalar}, @@ -104,7 +103,7 @@ impl IteratedScores { } /// Nodes per chunk. Fixed, so every sum groups its terms the same way whatever -/// the pool, and results are identical at any width. +/// the worker count, and results are identical at any. const CHUNK: usize = 4096; fn validate(options: IterationOptions) -> Result<()> { @@ -125,13 +124,19 @@ fn pull( (keep, beta, alpha): (f64, f64, f64), context: &ExecutionContext, ) -> Result { - let sums = parallel::for_chunks(into, CHUNK, |first, chunk| { - let mut meter = Meter::new(context); + // Chunks are fixed, so the worker count cannot change a sum; it only + // decides how many chunks run at once. + let workers = parallel::concurrency( + context, + into.len().saturating_add(arcs.targets.values.len()), + ); + let sums = parallel::for_chunks(workers, into, CHUNK, |first, chunk| { + let mut meter = context.work_meter(); let mut squares = 0.0; for (offset, value) in chunk.iter_mut().enumerate() { let node = first + offset; let range = arcs.range(node); - meter.tick(1 + range.len())?; + meter.charge(1 + range.len())?; let mut sum = 0.0; for arc in range { sum += arcs.weight(arc) * from[arcs.targets.values[arc]]; @@ -139,7 +144,6 @@ fn pull( *value = keep * from[node] + beta + alpha * sum; squares += *value * *value; } - meter.flush()?; Ok(squares) })?; Ok(sums.into_iter().sum()) @@ -152,7 +156,8 @@ fn settle( divisor: f64, context: &ExecutionContext, ) -> Result { - let sums = parallel::for_chunks(values, CHUNK, |first, chunk| { + let workers = parallel::concurrency(context, values.len()); + let sums = parallel::for_chunks(workers, values, CHUNK, |first, chunk| { context.charge_work(chunk.len())?; let mut change = 0.0; for (offset, value) in chunk.iter_mut().enumerate() { @@ -181,7 +186,7 @@ fn nonfinite(what: &str) -> AlgorithmError { /// graph with no cycle there is nothing to be principal; the iteration drifts /// toward the sinks and usually stops at the limit with `converged == false`. /// -/// Identical at any pool width. +/// Identical at any worker count. pub fn eigenvector(graph: &GraphProjection, options: IterationOptions) -> Result { let context = graph.execution(); context.checkpoint()?; @@ -224,7 +229,7 @@ pub fn eigenvector(graph: &GraphProjection, options: IterationOptions) -> Result /// valid values. Instead a run that does not settle returns /// `converged == false`, and one whose scores overflow fails and says why. /// -/// Identical at any pool width. +/// Identical at any worker count. pub fn katz(graph: &GraphProjection, options: KatzOptions) -> Result { let context = graph.execution(); context.checkpoint()?; @@ -275,7 +280,7 @@ pub fn katz(graph: &GraphProjection, options: KatzOptions) -> Result Result { let context = graph.execution(); context.checkpoint()?; diff --git a/crates/grust-algorithms/src/triangles.rs b/crates/grust-algorithms/src/triangles.rs index 980d5fe0..ac777d9c 100644 --- a/crates/grust-algorithms/src/triangles.rs +++ b/crates/grust-algorithms/src/triangles.rs @@ -6,7 +6,6 @@ use std::sync::atomic::{AtomicU64, Ordering}; use crate::{ AlgorithmError, GraphProjection, Orientation, Result, buffer::Buffer, - meter::Meter, parallel, table::{NodeColumn, NodeTable, TableScalar}, }; @@ -76,9 +75,9 @@ impl Triangles { /// /// Each triangle is found once, at its lowest-ranked corner, where rank orders /// nodes by (degree, row): O(A^1.5) intersections. Nodes are split into blocks -/// of equal estimated work and counted on the caller's rayon pool into atomic +/// of equal estimated work and counted on as many workers as the execution asked for into atomic /// per-node counters; the counts, and the work charged, are the same at any -/// pool width. +/// worker count. pub fn triangles(graph: &GraphProjection, options: TriangleOptions) -> Result { let context = graph.execution(); context.checkpoint()?; @@ -90,14 +89,14 @@ pub fn triangles(graph: &GraphProjection, options: TriangleOptions) -> Result Result Result Result Result Result Result = (0..edges.len()).map(|i| (1 + i % 4) as f64).collect(); for weighted in [false, true] { let run = |threads: usize| { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(threads) - .build() - .unwrap(); - pool.install(|| { - let context = context(); - let projection = graph( - 500, - &edges, - weighted.then_some(&weights[..]), - Orientation::Undirected, - &context, - ); - let before = context.usage().unwrap().work_units; - let result = betweenness(&projection, BetweennessOptions::default()).unwrap(); - let work = context.usage().unwrap().work_units - before; - let bits: Vec = result.values().iter().map(|v| v.to_bits()).collect(); - (bits, work) - }) + let context = context().with_concurrency(threads).unwrap(); + let projection = graph( + 500, + &edges, + weighted.then_some(&weights[..]), + Orientation::Undirected, + &context, + ); + let before = context.usage().unwrap().work_units; + let result = betweenness(&projection, BetweennessOptions::default()).unwrap(); + let work = context.usage().unwrap().work_units - before; + let bits: Vec = result.values().iter().map(|v| v.to_bits()).collect(); + (bits, work) }; let (bits, work) = run(1); for threads in [2, 3, 8] { diff --git a/crates/grust-algorithms/tests/closeness.rs b/crates/grust-algorithms/tests/closeness.rs index 4a3f6d37..261d8ac1 100644 --- a/crates/grust-algorithms/tests/closeness.rs +++ b/crates/grust-algorithms/tests/closeness.rs @@ -287,31 +287,25 @@ fn distance_centralities_are_identical_at_any_pool_width_and_charge_the_same_wor let weights: Vec = (0..edges.len()).map(|i| (1 + i % 7) as f64 / 3.0).collect(); for weighted in [false, true] { let run = |threads: usize| { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(threads) - .build() - .unwrap(); - pool.install(|| { - let context = context(); - let projection = graph( - 500, - &edges, - weighted.then_some(&weights[..]), - Orientation::Outgoing, - &context, - ); - let before = context.usage().unwrap().work_units; - let close = closeness(&projection, ClosenessOptions::default()).unwrap(); - let harm = harmonic(&projection, HarmonicOptions::default()).unwrap(); - let work = context.usage().unwrap().work_units - before; - let bits: Vec = close - .values() - .iter() - .chain(harm.values()) - .map(|v| v.to_bits()) - .collect(); - (bits, work) - }) + let context = context().with_concurrency(threads).unwrap(); + let projection = graph( + 500, + &edges, + weighted.then_some(&weights[..]), + Orientation::Outgoing, + &context, + ); + let before = context.usage().unwrap().work_units; + let close = closeness(&projection, ClosenessOptions::default()).unwrap(); + let harm = harmonic(&projection, HarmonicOptions::default()).unwrap(); + let work = context.usage().unwrap().work_units - before; + let bits: Vec = close + .values() + .iter() + .chain(harm.values()) + .map(|v| v.to_bits()) + .collect(); + (bits, work) }; let (bits, work) = run(1); for threads in [2, 3, 8] { diff --git a/crates/grust-algorithms/tests/fastrp.rs b/crates/grust-algorithms/tests/fastrp.rs index 375512ed..3c28abd7 100644 --- a/crates/grust-algorithms/tests/fastrp.rs +++ b/crates/grust-algorithms/tests/fastrp.rs @@ -297,27 +297,21 @@ fn fast_rp_is_identical_at_any_pool_width_and_charges_the_same_work() { .map(|_| (random.below(n), random.below(n))) .collect(); let run = |threads: usize| { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(threads) - .build() - .unwrap(); - pool.install(|| { - let context = context(); - let projection = graph(n, &edges, None, Orientation::Undirected, &context); - let before = context.usage().unwrap().work_units; - let result = fast_rp( - &projection, - FastRpOptions { - dimension: 32, - normalization_strength: -0.5, - self_influence: 0.2, - ..Default::default() - }, - ) - .unwrap(); - let bits: Vec = result.values().iter().map(|v| v.to_bits()).collect(); - (bits, context.usage().unwrap().work_units - before) - }) + let context = context().with_concurrency(threads).unwrap(); + let projection = graph(n, &edges, None, Orientation::Undirected, &context); + let before = context.usage().unwrap().work_units; + let result = fast_rp( + &projection, + FastRpOptions { + dimension: 32, + normalization_strength: -0.5, + self_influence: 0.2, + ..Default::default() + }, + ) + .unwrap(); + let bits: Vec = result.values().iter().map(|v| v.to_bits()).collect(); + (bits, context.usage().unwrap().work_units - before) }; let first = run(1); for threads in [2, 3, 8] { diff --git a/crates/grust-algorithms/tests/label_propagation.rs b/crates/grust-algorithms/tests/label_propagation.rs index 5c4d122e..c087f273 100644 --- a/crates/grust-algorithms/tests/label_propagation.rs +++ b/crates/grust-algorithms/tests/label_propagation.rs @@ -259,27 +259,21 @@ fn scrambled(n: usize, edges: usize) -> Vec<(usize, usize)> { fn a_seed_reproduces_its_result_at_any_pool_width() { let edges = scrambled(2000, 6000); let run = |threads: usize, seed| { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(threads) - .build() - .unwrap(); - pool.install(|| { - let context = context(); - let projection = graph(2000, &edges, None, Orientation::Undirected, &context); - let before = context.usage().unwrap().work_units; - let result = label_propagation( - &projection, - LabelPropagationOptions { - max_iterations: 50, - seed, - }, - ) - .unwrap(); - ( - result.communities().to_vec(), - context.usage().unwrap().work_units - before, - ) - }) + let context = context().with_concurrency(threads).unwrap(); + let projection = graph(2000, &edges, None, Orientation::Undirected, &context); + let before = context.usage().unwrap().work_units; + let result = label_propagation( + &projection, + LabelPropagationOptions { + max_iterations: 50, + seed, + }, + ) + .unwrap(); + ( + result.communities().to_vec(), + context.usage().unwrap().work_units - before, + ) }; let first = run(1, Some(7)); assert_eq!(run(8, Some(7)), first); diff --git a/crates/grust-algorithms/tests/leiden.rs b/crates/grust-algorithms/tests/leiden.rs index 7fc19940..f66e7247 100644 --- a/crates/grust-algorithms/tests/leiden.rs +++ b/crates/grust-algorithms/tests/leiden.rs @@ -290,25 +290,19 @@ fn leiden_is_reproducible_and_stays_connected_on_a_larger_graph() { edges.push((random.below(n), random.below(n), 1.0)); } let run = |threads: usize, seed| { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(threads) - .build() - .unwrap(); - pool.install(|| { - let context = context(); - let projection = graph(n, &edges, Orientation::Undirected, &context); - let options = LouvainOptions { - seed, - ..Default::default() - }; - let result = leiden(&projection, options).unwrap(); - let plain = louvain(&projection, options).unwrap(); - ( - result.communities().to_vec(), - result.modularity().to_bits(), - plain.modularity(), - ) - }) + let context = context().with_concurrency(threads).unwrap(); + let projection = graph(n, &edges, Orientation::Undirected, &context); + let options = LouvainOptions { + seed, + ..Default::default() + }; + let result = leiden(&projection, options).unwrap(); + let plain = louvain(&projection, options).unwrap(); + ( + result.communities().to_vec(), + result.modularity().to_bits(), + plain.modularity(), + ) }; let first = run(1, Some(5)); assert_eq!(run(8, Some(5)), first); diff --git a/crates/grust-algorithms/tests/louvain.rs b/crates/grust-algorithms/tests/louvain.rs index dec052ee..7b96e317 100644 --- a/crates/grust-algorithms/tests/louvain.rs +++ b/crates/grust-algorithms/tests/louvain.rs @@ -321,23 +321,17 @@ fn resolution_moves_between_components_and_singletons() { fn louvain_is_deterministic_under_a_fixed_seed_and_any_pool_width() { let edges = pseudo_random_edges(300, 1500, 99); let run = |seed: Option, threads: usize| { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(threads) - .build() - .unwrap(); - pool.install(|| { - let context = context(); - let projection = graph(300, &edges, Orientation::Undirected, &context); - let result = louvain( - &projection, - LouvainOptions { - seed, - ..Default::default() - }, - ) - .unwrap(); - (result.communities().to_vec(), result.modularity().to_bits()) - }) + let context = context().with_concurrency(threads).unwrap(); + let projection = graph(300, &edges, Orientation::Undirected, &context); + let result = louvain( + &projection, + LouvainOptions { + seed, + ..Default::default() + }, + ) + .unwrap(); + (result.communities().to_vec(), result.modularity().to_bits()) }; assert_eq!(run(None, 1), run(None, 8)); assert_eq!(run(Some(7), 1), run(Some(7), 8)); diff --git a/crates/grust-algorithms/tests/similarity.rs b/crates/grust-algorithms/tests/similarity.rs index dadcf07c..4c42e414 100644 --- a/crates/grust-algorithms/tests/similarity.rs +++ b/crates/grust-algorithms/tests/similarity.rs @@ -327,28 +327,21 @@ fn node_similarity_is_identical_at_any_pool_width_and_charges_the_same_work() { let weights: Vec = (0..edges.len()).map(|i| (1 + i % 5) as f64 / 3.0).collect(); for metric in METRICS { let run_with = |threads: usize| { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(threads) - .build() - .unwrap(); - pool.install(|| { - let context = context(); - let projection = - graph(600, &edges, Some(&weights), Orientation::Outgoing, &context); - let before = context.usage().unwrap().work_units; - let rows: Vec<_> = run( - &projection, - NodeSimilarityOptions { - metric, - top_k: 5, - ..Default::default() - }, - ) - .into_iter() - .map(|(a, b, s)| (a, b, s.to_bits())) - .collect(); - (rows, context.usage().unwrap().work_units - before) - }) + let context = context().with_concurrency(threads).unwrap(); + let projection = graph(600, &edges, Some(&weights), Orientation::Outgoing, &context); + let before = context.usage().unwrap().work_units; + let rows: Vec<_> = run( + &projection, + NodeSimilarityOptions { + metric, + top_k: 5, + ..Default::default() + }, + ) + .into_iter() + .map(|(a, b, s)| (a, b, s.to_bits())) + .collect(); + (rows, context.usage().unwrap().work_units - before) }; let first = run_with(1); assert!(first.0.len() > 1000); diff --git a/crates/grust-algorithms/tests/spanning.rs b/crates/grust-algorithms/tests/spanning.rs index 06725060..1ba07b77 100644 --- a/crates/grust-algorithms/tests/spanning.rs +++ b/crates/grust-algorithms/tests/spanning.rs @@ -218,21 +218,15 @@ fn the_forest_is_identical_at_any_pool_width_and_charges_the_same_work() { .collect(); let weights: Vec = (0..edges.len()).map(|_| random.below(50) as f64).collect(); let run = |threads: usize| { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(threads) - .build() - .unwrap(); - pool.install(|| { - let context = context(); - let projection = graph(n, &edges, Some(&weights), Orientation::Undirected, &context); - let before = context.usage().unwrap().work_units; - let result = spanning_tree(&projection, SpanningTreeOptions::default()).unwrap(); - ( - result.edges().to_vec(), - result.total_weight().to_bits(), - context.usage().unwrap().work_units - before, - ) - }) + let context = context().with_concurrency(threads).unwrap(); + let projection = graph(n, &edges, Some(&weights), Orientation::Undirected, &context); + let before = context.usage().unwrap().work_units; + let result = spanning_tree(&projection, SpanningTreeOptions::default()).unwrap(); + ( + result.edges().to_vec(), + result.total_weight().to_bits(), + context.usage().unwrap().work_units - before, + ) }; let first = run(1); assert!(first.0.len() > 19_000); diff --git a/crates/grust-algorithms/tests/spectral.rs b/crates/grust-algorithms/tests/spectral.rs index 05a36fb5..397a3385 100644 --- a/crates/grust-algorithms/tests/spectral.rs +++ b/crates/grust-algorithms/tests/spectral.rs @@ -359,45 +359,39 @@ fn the_iterations_are_identical_at_any_pool_width_and_charge_the_same_work() { let edges = scrambled(n, 60_000); let weights: Vec = (0..edges.len()).map(|i| (1 + i % 7) as f64 / 3.0).collect(); let run = |threads: usize| { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(threads) - .build() - .unwrap(); - pool.install(|| { - let context = context(); - let projection = graph(n, &edges, Some(&weights), Orientation::Outgoing, &context); - let before = context.usage().unwrap().work_units; - let options = IterationOptions { - max_iterations: 30, - ..Default::default() - }; - let results = [ - eigenvector(&projection, options).unwrap(), - hits(&projection, options).unwrap(), - katz( - &projection, - KatzOptions { - alpha: 0.001, - iteration: options, - ..Default::default() - }, - ) - .unwrap(), - ]; - let bits: Vec = results - .iter() - .flat_map(|result| { - ["score", "hub", "authority"] - .into_iter() - .filter_map(|name| result.values(name)) - .flatten() - .map(|v| v.to_bits()) - .chain([result.residual().to_bits(), result.iterations() as u64]) - .collect::>() - }) - .collect(); - (bits, context.usage().unwrap().work_units - before) - }) + let context = context().with_concurrency(threads).unwrap(); + let projection = graph(n, &edges, Some(&weights), Orientation::Outgoing, &context); + let before = context.usage().unwrap().work_units; + let options = IterationOptions { + max_iterations: 30, + ..Default::default() + }; + let results = [ + eigenvector(&projection, options).unwrap(), + hits(&projection, options).unwrap(), + katz( + &projection, + KatzOptions { + alpha: 0.001, + iteration: options, + ..Default::default() + }, + ) + .unwrap(), + ]; + let bits: Vec = results + .iter() + .flat_map(|result| { + ["score", "hub", "authority"] + .into_iter() + .filter_map(|name| result.values(name)) + .flatten() + .map(|v| v.to_bits()) + .chain([result.residual().to_bits(), result.iterations() as u64]) + .collect::>() + }) + .collect(); + (bits, context.usage().unwrap().work_units - before) }; let first = run(1); for threads in [2, 3, 8] { diff --git a/crates/grust-algorithms/tests/triangles.rs b/crates/grust-algorithms/tests/triangles.rs index e91e484b..7455c897 100644 --- a/crates/grust-algorithms/tests/triangles.rs +++ b/crates/grust-algorithms/tests/triangles.rs @@ -233,24 +233,21 @@ fn scrambled(n: usize, edges: usize) -> Vec<(usize, usize)> { #[test] fn triangles_are_identical_at_any_pool_width_and_charge_the_same_work() { - let edges = scrambled(400, 6000); + // Large enough to clear the floor below which a kernel stays on one worker; + // a smaller fixture would pass by never going parallel. + const N: usize = 700; + let edges = scrambled(N, 14_000); let run = |threads: usize| { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(threads) - .build() - .unwrap(); - pool.install(|| { - let context = context(); - let projection = graph(400, &edges, Orientation::Undirected, &context); - let before = context.usage().unwrap().work_units; - let result = triangles(&projection, TriangleOptions::default()).unwrap(); - let work = context.usage().unwrap().work_units - before; - (result.triangles().to_vec(), result.triangle_count(), work) - }) + let context = context().with_concurrency(threads).unwrap(); + let projection = graph(N, &edges, Orientation::Undirected, &context); + let before = context.usage().unwrap().work_units; + let result = triangles(&projection, TriangleOptions::default()).unwrap(); + let work = context.usage().unwrap().work_units - before; + (result.triangles().to_vec(), result.triangle_count(), work) }; let (counts, total, work) = run(1); assert!(total > 0, "the fixture should contain triangles"); - let (expected, _, expected_total) = by_definition(400, &edges); + let (expected, _, expected_total) = by_definition(N, &edges); assert_eq!(counts, expected); assert_eq!(total, expected_total); for threads in [2, 3, 8] { diff --git a/docs/goals/graph-analytics-catalog.md b/docs/goals/graph-analytics-catalog.md index ceb7e260..fdd7f2c9 100644 --- a/docs/goals/graph-analytics-catalog.md +++ b/docs/goals/graph-analytics-catalog.md @@ -592,11 +592,15 @@ M4 (Tier A), then per milestone.** Each is a minor version: new public API. same treatment once: a `PairTable` and an `EdgeTable`. - **`Meter`** (`meter.rs`) is the block-charging helper rule 5 asks for: `tick` in the loop, `flush` before returning. -- **Thread count comes from the caller's rayon pool**, not from - `ExecutionLimits`: adding a field there breaks 33 construction sites and every - downstream crate. A caller installs a pool; rule 4 makes the result identical - at any width, and tests run each parallel kernel in a one-thread and a - many-thread pool. +- **Thread count comes from the execution**: + `ExecutionContext::with_concurrency(n)`, read through + `parallel::concurrency(context, units)`. A caller that asks for nothing gets + one worker on its own thread and no pool is ever started; nothing reads the + CPU count. (The first version read the installed rayon pool, which outside a + pool is the whole machine — wrong inside Sail.) Rule 4 makes the result + identical at any count, and each parallel kernel's test asserts it at 1, 2, 3 + and 8 on a fixture large enough to clear the sequential floor, because a + smaller one passes by never going parallel. - **Undirected-only kernels refuse a directed projection with a message that contains "undirected".** Nutmeg's schema probe and Grust's catalog test both rely on that word to retry on an undirected projection. Keep it. @@ -613,13 +617,15 @@ M4 (Tier A), then per milestone.** Each is a minor version: new public API. contiguous blocks of equal estimated work (`balanced_ranges`), run one task per block (`map_ranges`), combine in block order. Cross-block writes go through atomic counters (`Buffer::filled_with`), never per-task vectors, so memory and - charged work do not depend on the pool width. Each task owns a `Meter`. + charged work do not depend on the worker count. Each task owns a `WorkMeter` + (`context.work_meter()`), created per block, never per node: creating one + registers it under a lock. Triangles proves it: identical counts *and* identical work at 1, 2, 3, 8 threads. - **Per-source kernels use `parallel::ordered_blocks`.** Betweenness splits its sources into fixed blocks of 64, runs `width()` blocks at a time, and merges per-block vectors in block order: float sums and charged work are bit-identical at any width, but live memory is one workspace per running - block, so it does grow with the pool. Closeness and harmonic (group 6) should + block, so it does grow with the workers. Closeness and harmonic (group 6) should reuse it and `shortest::MinHeap` (now `pub(crate)`). Betweenness needs only out-arcs: the backward pass rescans them instead of storing predecessors. GDS's `samplingSeed` maps to `seed`; GDS does not rescale a sampled score and @@ -645,7 +651,7 @@ M4 (Tier A), then per milestone.** Each is a minor version: new public API. `edgeOrdinal` (then `componentId`). Spanning tree and max flow should do the same. - **Iterative kernels pull, in fixed chunks.** `parallel::for_chunks` cuts a - vector into chunks of 4,096 whatever the pool, each node pulls its next value + vector into chunks of 4,096 whatever the worker count, each node pulls its next value from `incoming()`, and per-chunk partial sums are folded in chunk order. That is what makes a float reduction bit-identical at any width; `balanced_ranges` cannot, because its boundaries move with the width. `articleRank` is not done: From 535290972be67d66358ef25d12bc6747dabe4cd3 Mon Sep 17 00:00:00 2001 From: Alexy Khrabrov Date: Sat, 19 Sep 2026 23:59:08 -0700 Subject: [PATCH 2/2] Charge the sequential kernels through a work meter The second half of step 5: item 3 of quegee's 03:05Z follow-ups. dijkstra, shortestPaths, bfs, dfs, topologicalSort, scc, wcc and the shared MinHeap no longer pay one compare-exchange on the shared counter per visited entry. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 10 ++++++ crates/grust-algorithms/src/betweenness.rs | 15 +++------ crates/grust-algorithms/src/closeness.rs | 16 +++------ crates/grust-algorithms/src/ordering.rs | 12 ++++--- crates/grust-algorithms/src/shortest.rs | 22 +++++++----- crates/grust-algorithms/src/traversal.rs | 39 ++++++++++++---------- 6 files changed, 61 insertions(+), 53 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 91d152f1..416ac071 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,16 @@ reconstructed from Git history, release commits, and the shipped docs. ## Unreleased +- The sequential kernels — `dijkstra`, `shortestPaths`, `bfs`, `dfs`, + `topologicalSort`, `scc`, `wcc` — and the heap the path kernels share charge + work through a `WorkMeter` instead of one compare-exchange on the shared + counter per visited entry. Budgets, cancellation and results are unchanged, + and the exact-work-charge tests pass as they were. On a million-node, + ten-million-arc random graph, one laptop, release, no deadline, best of nine, + per-entry against metered: `dijkstra` 690 ms to 513, `bfs` 312 to 203, `dfs` + 358 to 302, `scc` 682 to 563, `wcc` 310 to 186, with the per-entry figures + reproduced after the metered run. These are single-host probes, not the + paired benchmark harness, which has yet to run on this change. - The catalog kernels take their worker count from the execution (`ExecutionContext::with_concurrency`) instead of from whatever rayon pool the caller had installed, and charge work through `WorkMeter`. **Behaviour diff --git a/crates/grust-algorithms/src/betweenness.rs b/crates/grust-algorithms/src/betweenness.rs index 5551364f..2c6bae70 100644 --- a/crates/grust-algorithms/src/betweenness.rs +++ b/crates/grust-algorithms/src/betweenness.rs @@ -117,7 +117,7 @@ pub fn betweenness(graph: &GraphProjection, options: BetweennessOptions) -> Resu let last = (first + BLOCK).min(source_count); let mut workspace = Workspace::new(n, weighted, context)?; for &source in &sources.values[first..last] { - workspace.accumulate(graph, source, context)?; + workspace.accumulate(graph, source)?; } Ok(workspace.local) }, @@ -182,12 +182,7 @@ impl Workspace { }) } - fn accumulate( - &mut self, - graph: &GraphProjection, - source: usize, - context: &ExecutionContext, - ) -> Result<()> { + fn accumulate(&mut self, graph: &GraphProjection, source: usize) -> Result<()> { let adjacency = graph.outgoing(); self.order.values.clear(); self.distance.values[source] = 0.0; @@ -195,8 +190,8 @@ impl Workspace { // Forward: settle nodes in nondecreasing distance, counting shortest paths. if let Some(heap) = &mut self.heap { - heap.improve(source, 0.0, context)?; - while let Some((cost, node)) = heap.pop(context)? { + heap.improve(source, 0.0, &mut self.meter)?; + while let Some((cost, node)) = heap.pop(&mut self.meter)? { self.order.values.push(node); let range = adjacency.range(node); self.meter.charge(1 + range.len())?; @@ -206,7 +201,7 @@ impl Workspace { if candidate < self.distance.values[next] { self.distance.values[next] = candidate; self.paths.values[next] = self.paths.values[node]; - heap.improve(next, candidate, context)?; + heap.improve(next, candidate, &mut self.meter)?; } else if candidate == self.distance.values[next] { self.paths.values[next] += self.paths.values[node]; } diff --git a/crates/grust-algorithms/src/closeness.rs b/crates/grust-algorithms/src/closeness.rs index 01bb8ff7..4dbca005 100644 --- a/crates/grust-algorithms/src/closeness.rs +++ b/crates/grust-algorithms/src/closeness.rs @@ -138,8 +138,7 @@ fn sweep( let mut workspace = Workspace::new(n, weighted, context)?; let mut part = Buffer::capacity(last - first, context)?; for source in first..last { - let (reached, distance_sum, reciprocal_sum) = - workspace.run(graph, source, context)?; + let (reached, distance_sum, reciprocal_sum) = workspace.run(graph, source)?; part.values .push(score(reached, distance_sum, reciprocal_sum)); } @@ -178,18 +177,13 @@ impl Workspace { /// Settle everything reachable from `source`, summing in settle order, which /// the queue and the heap's tie-break make a function of the graph alone. - fn run( - &mut self, - graph: &GraphProjection, - source: usize, - context: &ExecutionContext, - ) -> Result<(usize, f64, f64)> { + fn run(&mut self, graph: &GraphProjection, source: usize) -> Result<(usize, f64, f64)> { let adjacency = graph.outgoing(); self.order.values.clear(); self.distance.values[source] = 0.0; if let Some(heap) = &mut self.heap { - heap.improve(source, 0.0, context)?; - while let Some((cost, node)) = heap.pop(context)? { + heap.improve(source, 0.0, &mut self.meter)?; + while let Some((cost, node)) = heap.pop(&mut self.meter)? { self.order.values.push(node); let range = adjacency.range(node); self.meter.charge(1 + range.len())?; @@ -198,7 +192,7 @@ impl Workspace { let candidate = cost + adjacency.weight(arc); if candidate < self.distance.values[next] { self.distance.values[next] = candidate; - heap.improve(next, candidate, context)?; + heap.improve(next, candidate, &mut self.meter)?; } } } diff --git a/crates/grust-algorithms/src/ordering.rs b/crates/grust-algorithms/src/ordering.rs index 89bb6188..30da94e4 100644 --- a/crates/grust-algorithms/src/ordering.rs +++ b/crates/grust-algorithms/src/ordering.rs @@ -30,6 +30,7 @@ pub enum TopologicalOrder { /// ties; each reachable node appears once. Weights are ignored. pub fn depth_first(graph: &GraphProjection, source: &str) -> Result { let context = graph.execution(); + let mut meter = context.work_meter(); context.checkpoint()?; let source = graph.source(source)?; let adjacency = graph.outgoing(); @@ -40,7 +41,7 @@ pub fn depth_first(graph: &GraphProjection, source: &str) -> Result { nodes.values.push(source); stack.values.push((source, adjacency.range(source).start)); while let Some(&mut (node, ref mut next)) = stack.values.last_mut() { - context.charge_work(1)?; + meter.charge(1)?; if *next == adjacency.range(node).end { stack.values.pop(); continue; @@ -71,20 +72,21 @@ enum Color { /// On cyclic input returns the first back-edge witness in node/adjacency order. pub fn topological_sort(graph: &GraphProjection) -> Result { let context = graph.execution(); + let mut meter = context.work_meter(); let n = graph.node_count(); let adjacency = graph.outgoing(); let mut colors = Buffer::filled(n, Color::Unseen, context)?; let mut finished = Buffer::capacity(n, context)?; let mut stack = Buffer::capacity(n, context)?; for seed in 0..n { - context.charge_work(1)?; + meter.charge(1)?; if colors.values[seed] != Color::Unseen { continue; } colors.values[seed] = Color::Active; stack.values.push((seed, adjacency.range(seed).start)); while let Some(&mut (node, ref mut next)) = stack.values.last_mut() { - context.charge_work(1)?; + meter.charge(1)?; if *next == adjacency.range(node).end { colors.values[node] = Color::Finished; finished.values.push(node); @@ -102,7 +104,7 @@ pub fn topological_sort(graph: &GraphProjection) -> Result { let mut witness = Buffer::capacity(n.saturating_add(1), context)?; let mut inside = false; for &(ancestor, _) in &stack.values { - context.charge_work(1)?; + meter.charge(1)?; inside |= ancestor == target; if inside { witness.values.push(ancestor); @@ -119,7 +121,7 @@ pub fn topological_sort(graph: &GraphProjection) -> Result { } } for index in 0..finished.values.len() / 2 { - context.charge_work(1)?; + meter.charge(1)?; finished.values.swap(index, n - index - 1); } Ok(TopologicalOrder::Acyclic(NodeOrder { diff --git a/crates/grust-algorithms/src/shortest.rs b/crates/grust-algorithms/src/shortest.rs index fd8484b9..bc43608e 100644 --- a/crates/grust-algorithms/src/shortest.rs +++ b/crates/grust-algorithms/src/shortest.rs @@ -199,6 +199,7 @@ type Parents = Buffer>; fn run(graph: &GraphProjection, source: &str, paths: bool) -> Result<(Distances, Option)> { let context = graph.execution(); + let mut meter = context.work_meter(); context.checkpoint()?; let source = graph.source(source)?; let n = graph.node_count(); @@ -208,12 +209,12 @@ fn run(graph: &GraphProjection, source: &str, paths: bool) -> Result<(Distances, .transpose()?; let mut heap = MinHeap::new(n, context)?; distances.values[source] = 0.0; - heap.improve(source, 0.0, context)?; + heap.improve(source, 0.0, &mut meter)?; let adjacency = graph.outgoing(); - while let Some((cost, node)) = heap.pop(context)? { - context.charge_work(1)?; + while let Some((cost, node)) = heap.pop(&mut meter)? { + meter.charge(1)?; for arc in adjacency.range(node) { - context.charge_work(1)?; + meter.charge(1)?; let next = adjacency.targets.values[arc]; let candidate = cost + adjacency.weight(arc); if !candidate.is_finite() { @@ -226,7 +227,7 @@ fn run(graph: &GraphProjection, source: &str, paths: bool) -> Result<(Distances, if let Some(parents) = &mut parents { parents.values[next] = Some((node, adjacency.edge_slot(arc))); } - heap.improve(next, candidate, context)?; + heap.improve(next, candidate, &mut meter)?; } } } @@ -264,7 +265,7 @@ impl MinHeap { &mut self, node: usize, cost: f64, - context: &ExecutionContext, + meter: &mut grust_procedures::WorkMeter, ) -> Result<()> { let mut index = self.positions.values[node]; if index == usize::MAX { @@ -275,7 +276,7 @@ impl MinHeap { self.entries.values[index].0 = cost; } while index > 0 { - context.charge_work(1)?; + meter.charge(1)?; let parent = (index - 1) / 2; if !self.less(index, parent) { break; @@ -286,7 +287,10 @@ impl MinHeap { Ok(()) } - pub(crate) fn pop(&mut self, context: &ExecutionContext) -> Result> { + pub(crate) fn pop( + &mut self, + meter: &mut grust_procedures::WorkMeter, + ) -> Result> { if self.entries.values.is_empty() { return Ok(None); } @@ -298,7 +302,7 @@ impl MinHeap { self.positions.values[self.entries.values[0].1] = 0; let mut index = 0; while index < self.entries.values.len() / 2 { - context.charge_work(1)?; + meter.charge(1)?; let left = index * 2 + 1; let right = left + 1; let child = if right < self.entries.values.len() && self.less(right, left) { diff --git a/crates/grust-algorithms/src/traversal.rs b/crates/grust-algorithms/src/traversal.rs index 78273099..d2b31e29 100644 --- a/crates/grust-algorithms/src/traversal.rs +++ b/crates/grust-algorithms/src/traversal.rs @@ -66,6 +66,7 @@ fn bfs_sources<'a>( sources: impl Iterator, ) -> Result { let context = graph.execution(); + let mut meter = context.work_meter(); context.checkpoint()?; let adjacency = graph.outgoing(); let workers = crate::parallel::workers_above( @@ -79,7 +80,7 @@ fn bfs_sources<'a>( { let mut seen = Buffer::filled(graph.node_count(), false, context)?; for source in sources { - context.charge_work(1)?; + meter.charge(1)?; let source = graph.source(source)?; if !seen.values[source] { seen.values[source] = true; @@ -98,11 +99,11 @@ fn bfs_sources<'a>( } let mut head = 0; while head < queue.values.len() { - context.charge_work(1)?; + meter.charge(1)?; let node = queue.values[head]; head += 1; for arc in adjacency.range(node) { - context.charge_work(1)?; + meter.charge(1)?; let next = adjacency.targets.values[arc]; if distances.values[next].is_infinite() { distances.values[next] = distances.values[node] + 1.0; @@ -217,6 +218,7 @@ fn levels(graph: &GraphProjection, roots: &[usize], workers: usize) -> Result Result { let context = graph.execution(); + let mut meter = context.work_meter(); let n = graph.node_count(); let workers = crate::parallel::workers( context, @@ -227,14 +229,14 @@ pub fn weakly_connected_components(graph: &GraphProjection) -> Result Result Result Result { while parents[node] != node { - context.charge_work(1)?; + meter.charge(1)?; parents[node] = parents[parents[node]]; node = parents[node]; } @@ -370,20 +372,21 @@ fn union_find(graph: &GraphProjection, workers: usize) -> Result { /// in admitted heap buffers, including on chains with millions of vertices. pub fn strongly_connected_components(graph: &GraphProjection) -> Result { let context = graph.execution(); + let mut meter = context.work_meter(); let n = graph.node_count(); let adjacency = graph.outgoing(); let mut seen = Buffer::filled(n, false, context)?; let mut order = Buffer::capacity(n, context)?; let mut stack = Buffer::capacity(n, context)?; for seed in 0..n { - context.charge_work(1)?; + meter.charge(1)?; if seen.values[seed] { continue; } seen.values[seed] = true; stack.values.push((seed, adjacency.range(seed).start)); while let Some(&mut (node, ref mut next_arc)) = stack.values.last_mut() { - context.charge_work(1)?; + meter.charge(1)?; if *next_arc == adjacency.range(node).end { order.values.push(node); stack.values.pop(); @@ -405,7 +408,7 @@ pub fn strongly_connected_components(graph: &GraphProjection) -> Result Result Result