Skip to content

Catalog and sequential kernels on WorkMeter; threads come from the execution #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alexy merged 2 commits into from
Sep 20, 2026
Merged

Catalog and sequential kernels on WorkMeter; threads come from the execution #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6003dc5
Move the catalog kernels onto WorkMeter and the execution's concurrency
alexy Sep 20, 2026
5352909
Charge the sequential kernels through a work meter
alexy Sep 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,26 @@ 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
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
Expand Down
47 changes: 24 additions & 23 deletions crates/grust-algorithms/src/betweenness.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
use crate::{
AlgorithmError, GraphProjection, Orientation, Result,
buffer::Buffer,
meter::Meter,
parallel, random,
shortest::MinHeap,
table::{NodeColumn, NodeTable},
Expand Down Expand Up @@ -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
Expand All @@ -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<Betweenness> {
let context = graph.execution();
context.checkpoint()?;
Expand All @@ -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.
Expand All @@ -107,14 +105,19 @@ 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;
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)
},
Expand Down Expand Up @@ -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<f64>,
paths: Buffer<f64>,
dependency: Buffer<f64>,
Expand All @@ -166,6 +172,7 @@ struct Workspace {
impl Workspace {
fn new(n: usize, weighted: bool, context: &ExecutionContext) -> Result<Self> {
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)?,
Expand All @@ -175,32 +182,26 @@ 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();
let mut meter = Meter::new(context);
self.order.values.clear();
self.distance.values[source] = 0.0;
self.paths.values[source] = 1.0;

// 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);
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);
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];
}
Expand All @@ -213,7 +214,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];
Expand All @@ -232,7 +233,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
Expand All @@ -255,6 +256,6 @@ impl Workspace {
self.paths.values[node] = 0.0;
self.dependency.values[node] = 0.0;
}
meter.flush()
Ok(())
}
}
12 changes: 5 additions & 7 deletions crates/grust-algorithms/src/biconnected.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
use crate::{
AlgorithmError, GraphProjection, Orientation, Result,
buffer::Buffer,
meter::Meter,
table::{NodeColumn, NodeTable},
};

Expand Down Expand Up @@ -124,7 +123,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result<Biconnectivity> {
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)?;
Expand All @@ -138,7 +137,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result<Biconnectivity> {
let mut clock = 0usize;

for root in 0..n {
meter.tick(1)?;
meter.charge(1)?;
if discovered.values[root] != UNSEEN {
continue;
}
Expand All @@ -152,7 +151,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result<Biconnectivity> {
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 {
Expand Down Expand Up @@ -196,7 +195,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result<Biconnectivity> {
"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)
Expand All @@ -210,7 +209,7 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result<Biconnectivity> {
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
Expand All @@ -220,7 +219,6 @@ pub fn biconnectivity(graph: &GraphProjection) -> Result<Biconnectivity> {
articulation
.values
.extend((0..n).filter(|&node| is_articulation.values[node]));
meter.flush()?;
Ok(Biconnectivity {
graph: graph.clone(),
components,
Expand Down
41 changes: 20 additions & 21 deletions crates/grust-algorithms/src/closeness.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
use crate::{
AlgorithmError, GraphProjection, Result,
buffer::Buffer,
meter::Meter,
parallel,
shortest::MinHeap,
table::{NodeColumn, NodeTable},
Expand Down Expand Up @@ -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<DistanceCentrality> {
let others = graph.node_count().saturating_sub(1) as f64;
Expand Down Expand Up @@ -101,7 +100,7 @@ pub fn harmonic(graph: &GraphProjection, options: HarmonicOptions) -> Result<Dis
})
}

/// Sources per block; fixed, so nothing depends on the pool width.
/// Sources per block; fixed, so nothing depends on the worker count.
const BLOCK: usize = 64;

/// Run one sweep per node and score it from (nodes reached, Σ d, Σ 1/d).
Expand All @@ -115,28 +114,31 @@ fn sweep(
let adjacency = graph.outgoing();
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(
"distance centralities need positive weights: a zero distance between distinct nodes has no reciprocal".into(),
));
}
}
meter.flush()?;
}
let mut scores = Buffer::filled(n, 0.0f64, context)?;
let workers = parallel::concurrency(
context,
n.saturating_mul(n.saturating_add(adjacency.targets.values.len())),
);
parallel::ordered_blocks(
workers,
n.div_ceil(BLOCK),
|block| {
let first = block * BLOCK;
let last = (first + BLOCK).min(n);
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));
}
Expand All @@ -155,6 +157,9 @@ fn sweep(
}

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<f64>,
order: Buffer<usize>,
heap: Option<MinHeap>,
Expand All @@ -163,6 +168,7 @@ struct Workspace {
impl Workspace {
fn new(n: usize, weighted: bool, context: &ExecutionContext) -> Result<Self> {
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()?,
Expand All @@ -171,28 +177,22 @@ 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();
let mut meter = Meter::new(context);
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);
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);
if candidate < self.distance.values[next] {
self.distance.values[next] = candidate;
heap.improve(next, candidate, context)?;
heap.improve(next, candidate, &mut self.meter)?;
}
}
}
Expand All @@ -203,7 +203,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() {
Expand All @@ -222,7 +222,6 @@ impl Workspace {
for &node in &self.order.values {
self.distance.values[node] = f64::INFINITY;
}
meter.flush()?;
Ok((reached, distance_sum, reciprocal_sum))
}
}
Loading
Loading