Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
68 changes: 68 additions & 0 deletions crates/graphforge-exec/src/algorithm_analyze.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3981,6 +3981,26 @@ mod tests {
)
}

fn execute_with_compute_threads(
graph: &AdjacencyGraph,
algorithm: AnalyzeAlgorithm,
directed: bool,
threads: usize,
) -> Result<AlgorithmOutput, AlgorithmError> {
let mut registry = AlgorithmRegistry::default();
register_analyze_algorithms(&mut registry, directed)?;
let control = AlgorithmControl::new(
AlgorithmLimits::default().with_compute_threads(threads),
AlgorithmCancellation::default(),
)
.with_compute_pool(Arc::new(crate::ComputePool::new(threads).unwrap()));
registry.execute(Algorithm::Analyze(algorithm), graph, &control)
}

fn output_fingerprint(output: &AlgorithmOutput) -> String {
format!("{:?}|{:?}", output.schema, output.rows())
}

fn execute_minimum_k_spanning_tree(
graph: &AdjacencyGraph,
k: usize,
Expand Down Expand Up @@ -6945,6 +6965,54 @@ mod tests {
);
}

#[test]
fn bridges_keeps_serial_fingerprint_under_thread_budgets() {
let graph = AdjacencyGraph::with_test_undirected_multigraph(
10,
&[
(10, 0, 1),
(11, 1, 2),
(12, 2, 0),
(13, 1, 3),
(14, 3, 4),
(15, 4, 5),
(16, 5, 3),
(17, 5, 6),
(18, 6, 7),
(19, 7, 5),
(20, 7, 8),
(21, 8, 9),
(22, 7, 8),
],
);
let serial =
execute_with_compute_threads(&graph, AnalyzeAlgorithm::Bridges, false, 1).unwrap();
assert_eq!(
serial.rows(),
[
vec![
AlgorithmValue::Uuid(13_u128.to_be_bytes()),
AlgorithmValue::Uuid(1_u128.to_be_bytes()),
AlgorithmValue::Uuid(3_u128.to_be_bytes()),
],
vec![
AlgorithmValue::Uuid(21_u128.to_be_bytes()),
AlgorithmValue::Uuid(8_u128.to_be_bytes()),
AlgorithmValue::Uuid(9_u128.to_be_bytes()),
],
]
);
let serial_fingerprint = output_fingerprint(&serial);

for threads in [2_usize, 4, 8] {
let output =
execute_with_compute_threads(&graph, AnalyzeAlgorithm::Bridges, false, threads)
.unwrap();
assert_eq!(output.schema, serial.schema);
assert_eq!(output_fingerprint(&output), serial_fingerprint);
}
}

#[test]
fn graphsage_source_resource_contract_validates_feature_matrix() {
let empty = AdjacencyGraph::with_test_counts(0, 0);
Expand Down
15 changes: 15 additions & 0 deletions docs/development/execution-resource-policy.md
Original file line number Diff line number Diff line change
Expand Up @@ -262,6 +262,21 @@ and fingerprints match the one-thread result at `1`/`2`/`4`/`8`/automatic
configurations. Cancellation remains cooperative both before worker launch and
inside chunk scans.

## Serial bridges (#564)

`analyze(by="bridges")` is intentionally SERIAL. Bridge classification depends
on the shared low-link DFS traversal: discovery order, parent-edge identity,
multigraph parallel-edge handling, and low-link propagation determine whether an
edge is a bridge. Parallelizing that state would change the canonical edge
evidence or require synchronization around each DFS transition, so there is no
safe independent kernel for the private compute pool.

The handler keeps Rust-owned adjacency projection and bounded Arrow output,
does not use Rayon's global pool, and preserves cancellation and shared resource
limits. A fingerprint test attaches private compute pools for `1`/`2`/`4`/`8`
configured compute threads and requires identical schemas, bridge ordering, and
rows.

## Observability

`GraphForge::resource_policy()` and `GraphForge::resource_diagnostics()` expose
Expand Down