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
64 changes: 64 additions & 0 deletions crates/graphforge-exec/src/algorithm_paths.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1726,6 +1726,27 @@ mod tests {
)
}

fn execute_path_with_compute_threads(
graph: &AdjacencyGraph,
algorithm: PathAlgorithm,
source: u64,
target: Option<u64>,
threads: usize,
) -> Result<AlgorithmOutput, AlgorithmError> {
let mut registry = AlgorithmRegistry::default();
register_path_algorithms(&mut registry, uuid(source), target.map(uuid), 1, None, None)?;
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::Paths(algorithm), graph, &control)
}

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

fn execute_gomory_hu(
graph: &AdjacencyGraph,
limits: AlgorithmLimits,
Expand Down Expand Up @@ -2743,6 +2764,49 @@ mod tests {
}
}

#[test]
fn dfs_keeps_serial_fingerprint_under_thread_budgets() {
let graph = AdjacencyGraph::with_test_directed_edges(
8,
&[
(0, 2),
(0, 1),
(0, 1),
(1, 3),
(1, 4),
(3, 6),
(6, 1),
(2, 5),
(5, 7),
(7, 7),
],
);
let serial =
execute_path_with_compute_threads(&graph, PathAlgorithm::Dfs, 0, None, 1).unwrap();
assert_eq!(
serial.rows(),
vec![
traversal(0, 0, 0),
traversal(1, 1, 1),
traversal(3, 2, 2),
traversal(6, 3, 3),
traversal(4, 2, 4),
traversal(2, 1, 5),
traversal(5, 2, 6),
traversal(7, 3, 7),
]
);
let serial_fingerprint = output_fingerprint(&serial);

for threads in [2_usize, 4, 8] {
let output =
execute_path_with_compute_threads(&graph, PathAlgorithm::Dfs, 0, None, threads)
.unwrap();
assert_eq!(output.schema, serial.schema);
assert_eq!(output_fingerprint(&output), serial_fingerprint);
}
}

#[test]
fn maximum_flow_views_reject_noncanonical_k() {
for by in [
Expand Down
12 changes: 12 additions & 0 deletions docs/development/execution-resource-policy.md
Original file line number Diff line number Diff line change
Expand Up @@ -526,6 +526,18 @@ chunks merge in ascending range order so row ordering and the first undefined
partition error remain deterministic. Schemas, row order, conductance bits,
structured errors, and fingerprints match the one-thread result at
`1`/`2`/`4`/`8`/automatic configurations.
## Serial depth-first search (#540)

`paths(by="dfs")` is intentionally SERIAL. The public result is preorder with a
stable discovery index and depth. That order is defined by a single stack over
sorted neighbor lists; parallelizing frontier expansion would change discovery
order, depth ties, or require serial reassembly of every stack mutation.

The handler uses the Rust-owned adjacency projection and bounded Arrow sink,
does not use Rayon's global pool, and preserves shared cancellation and limits.
A fingerprint test attaches private compute pools for `1`/`2`/`4`/`8`
configured compute threads and requires identical schemas, preorder rows,
depths, and discovery ordinals.

## Observability

Expand Down