diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 860326eb..a330ca1b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,7 +20,11 @@ jobs: # insertion order and the random draws, producing a different graph — and # therefore different covered lines — on each run. That made whole-workspace # coverage wobble run-to-run and false-tripped the ratchet. One worker keeps - # the ratchet input reproducible and matched to the coverage-main baseline. + # the ratchet input reproducible and matched to the coverage-main baseline; + # it controls coverage-ratchet determinism only. MST concurrency coverage + # comes from a `std::thread` `try_union` stress test and an explicit + # `ThreadPoolBuilder` 1-versus-8-thread determinism test, both of which use + # real concurrency regardless of this pin. RAYON_NUM_THREADS: '1' steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 diff --git a/chutoro-core/src/mst/property/concurrency.rs b/chutoro-core/src/mst/property/concurrency.rs index d36bedb6..5e2fddf0 100644 --- a/chutoro-core/src/mst/property/concurrency.rs +++ b/chutoro-core/src/mst/property/concurrency.rs @@ -8,6 +8,7 @@ //! See `docs/property-testing-design.md` §4.3.3. use proptest::test_runner::{TestCaseError, TestCaseResult}; +use rayon::ThreadPoolBuilder; use crate::{EdgeHarvest, MstEdge, parallel_kruskal}; @@ -92,3 +93,61 @@ pub(super) fn run_concurrency_safety_property(fixture: &MstFixture) -> TestCaseR Ok(()) } + +/// Runs the thread-pool determinism property for the given fixture. +/// +/// Executes `parallel_kruskal` in dedicated one- and eight-thread Rayon pools +/// and asserts that both runs produce exactly the same forest. +pub(super) fn run_thread_pool_determinism_property(fixture: &MstFixture) -> TestCaseResult { + let single_thread_pool = ThreadPoolBuilder::new() + .num_threads(1) + .build() + .map_err(|error| { + TestCaseError::fail(format!("failed to build one-thread Rayon pool: {error}")) + })?; + let eight_thread_pool = ThreadPoolBuilder::new() + .num_threads(8) + .build() + .map_err(|error| { + TestCaseError::fail(format!("failed to build eight-thread Rayon pool: {error}")) + })?; + + let single_thread_harvest = EdgeHarvest::new(fixture.edges.clone()); + let single_thread_forest = single_thread_pool + .install(|| parallel_kruskal(fixture.node_count, &single_thread_harvest)) + .map_err(|error| { + TestCaseError::fail(format!( + "one-thread parallel_kruskal failed: {error} \ + (distribution={:?}, nodes={}, edges={})", + fixture.distribution, + fixture.node_count, + fixture.edges.len(), + )) + })?; + + let eight_thread_harvest = EdgeHarvest::new(fixture.edges.clone()); + let eight_thread_forest = eight_thread_pool + .install(|| parallel_kruskal(fixture.node_count, &eight_thread_harvest)) + .map_err(|error| { + TestCaseError::fail(format!( + "eight-thread parallel_kruskal failed: {error} \ + (distribution={:?}, nodes={}, edges={})", + fixture.distribution, + fixture.node_count, + fixture.edges.len(), + )) + })?; + + if single_thread_forest != eight_thread_forest { + return Err(TestCaseError::fail(format!( + "parallel_kruskal output diverged by Rayon thread count \ + (distribution={:?}, nodes={}, edges={}, one_thread={single_thread_forest:?}, \ + eight_threads={eight_thread_forest:?})", + fixture.distribution, + fixture.node_count, + fixture.edges.len(), + ))); + } + + Ok(()) +} diff --git a/chutoro-core/src/mst/property/tests.rs b/chutoro-core/src/mst/property/tests.rs index 42b2a3fa..a366fedf 100644 --- a/chutoro-core/src/mst/property/tests.rs +++ b/chutoro-core/src/mst/property/tests.rs @@ -1,9 +1,9 @@ //! Property-based test runners for the parallel Kruskal MST implementation. //! -//! Hosts proptest runners for all three properties (oracle equivalence, -//! structural invariants, concurrency safety), rstest parameterized cases -//! for targeted distribution coverage, and unit tests for the sequential -//! oracle itself. +//! Hosts proptest runners for all four properties (oracle equivalence, +//! structural invariants, repeatability, thread-pool determinism), rstest +//! parameterized cases for targeted distribution coverage, and unit tests for +//! the sequential oracle itself. use proptest::prelude::*; use rand::SeedableRng; @@ -12,7 +12,7 @@ use rand::rngs::SmallRng; use crate::CandidateEdge; use crate::test_utils::suite_proptest_config; -use super::concurrency::run_concurrency_safety_property; +use super::concurrency::{run_concurrency_safety_property, run_thread_pool_determinism_property}; use super::equivalence::run_oracle_equivalence_property; use super::oracle::{SequentialMstResult, sequential_kruskal}; use super::strategies::{generate_fixture, mst_fixture_strategy}; @@ -89,6 +89,11 @@ proptest! { fn mst_concurrency_safety(fixture in mst_fixture_strategy()) { run_concurrency_safety_property(&fixture)?; } + + #[test] + fn mst_thread_pool_determinism(fixture in mst_fixture_strategy()) { + run_thread_pool_determinism_property(&fixture)?; + } } // ======================================================================== @@ -113,6 +118,12 @@ parameterised_property_test!( "concurrency safety must hold" ); +parameterised_property_test!( + thread_pool_determinism_rstest, + run_thread_pool_determinism_property, + "thread-pool determinism must hold" +); + // ======================================================================== // TEST_CASES Consistency Check // ======================================================================== diff --git a/chutoro-core/src/mst/union_find.rs b/chutoro-core/src/mst/union_find.rs index eb84a08b..f772fe65 100644 --- a/chutoro-core/src/mst/union_find.rs +++ b/chutoro-core/src/mst/union_find.rs @@ -46,6 +46,11 @@ impl ConcurrentUnionFind { self.components.load(Ordering::Acquire) } + #[cfg(test)] + pub(super) fn root_of(&self, node: usize) -> usize { + self.find(node) + } + pub(super) fn try_union(&self, left: usize, right: usize) -> Result { loop { let left_root = self.find(left); @@ -156,3 +161,112 @@ fn choose_parent_child( lock_order(left_root, right_root) } + +#[cfg(test)] +mod tests { + //! Concurrent stress coverage for the striped-lock union-find protocol. + + use std::sync::{Arc, Barrier}; + + use rand::Rng; + use rand::SeedableRng; + use rand::rngs::SmallRng; + use rstest::rstest; + + use super::ConcurrentUnionFind; + + const NODE_COUNT: usize = 8; + const EDGE_COUNT: usize = 4_096; + + #[rstest] + #[case(42, 2)] + #[case(999, 4)] + #[case(7_777, 8)] + fn concurrent_unions_match_sequential_partition( + #[case] seed: u64, + #[case] thread_count: usize, + ) { + // Multiple workers contend for the same small lock table, exercising + // striped-lock ordering, root revalidation, and retry interleavings. + let edges = Arc::new(random_edges(seed)); + let union_find = Arc::new(ConcurrentUnionFind::new(NODE_COUNT)); + let start = Arc::new(Barrier::new(thread_count + 1)); + let chunk_size = edges.len().div_ceil(thread_count); + + let handles: Vec<_> = (0..thread_count) + .map(|worker_index| { + let edges = Arc::clone(&edges); + let union_find = Arc::clone(&union_find); + let start = Arc::clone(&start); + let first = worker_index * chunk_size; + let last = (first + chunk_size).min(edges.len()); + + std::thread::spawn(move || { + start.wait(); + for &(left, right) in &edges[first..last] { + union_find + .try_union(left, right) + .expect("generated nodes must be valid"); + } + }) + }) + .collect(); + + start.wait(); + for handle in handles { + handle.join().expect("union worker must not panic"); + } + + let concurrent_labels = normalised_labels(|node| union_find.root_of(node)); + let (oracle_labels, oracle_components) = sequential_oracle(&edges); + + assert_eq!(concurrent_labels, oracle_labels); + assert_eq!(union_find.components(), oracle_components); + } + + fn random_edges(seed: u64) -> Vec<(usize, usize)> { + let mut rng = SmallRng::seed_from_u64(seed); + (0..EDGE_COUNT) + .map(|_| (rng.gen_range(0..NODE_COUNT), rng.gen_range(0..NODE_COUNT))) + .collect() + } + + fn normalised_labels(root_of: impl Fn(usize) -> usize) -> Vec { + let mut component_minimums = [NODE_COUNT; NODE_COUNT]; + for node in 0..NODE_COUNT { + let root = root_of(node); + component_minimums[root] = component_minimums[root].min(node); + } + + (0..NODE_COUNT) + .map(|node| component_minimums[root_of(node)]) + .collect() + } + + fn sequential_oracle(edges: &[(usize, usize)]) -> (Vec, usize) { + let mut parents: Vec = (0..NODE_COUNT).collect(); + let mut components = NODE_COUNT; + + for &(left, right) in edges { + let left_root = scalar_find(&parents, left); + let right_root = scalar_find(&parents, right); + if left_root != right_root { + parents[right_root] = left_root; + components -= 1; + } + } + + ( + normalised_labels(|node| scalar_find(&parents, node)), + components, + ) + } + + fn scalar_find(parents: &[usize], node: usize) -> usize { + let mut current = node; + while parents[current] != current { + current = parents[current]; + } + current + } +}