trueskill-tt/tests/determinism.rs

//! Determinism across `RAYON_NUM_THREADS`, on a workload that actually reaches
//! the parallel path.
//!
//! This test previously proved less than it appeared to. `sweep_color_groups`
//! takes its `par_iter` branch only for colour groups of at least
//! `RAYON_THRESHOLD` (64) events, and the old fixture built 20 slices of 10
//! events — a colour group is a subset of one slice's events, so it could never
//! exceed 10. The branch was unreachable, confirmed by CPU-vs-wall time:
//! `user 0.64` on eight threads is one core.
//!
//! It also compared a single competitor's curve out of forty, and never
//! compared `log_evidence`, `final_step` or `iterations`.
//!
//! The fixture below guarantees the parallel branch **by construction**: within
//! a slice every event uses a disjoint pair of competitors, so greedy colouring
//! puts all of them in colour 0, and that group is `EVENTS_PER_SLICE` long.
//! Competitors recur across slices, so the fit still has temporal coupling and
//! drift rather than being a set of independent duels.

#![cfg(feature = "rayon")]

use smallvec::smallvec;
use trueskill_tt::{
    ConstantDrift, ConvergenceOptions, Event, Gaussian, History, Member, Outcome, Team,
};

/// Comfortably above the crate's internal `RAYON_THRESHOLD` of 64.
const EVENTS_PER_SLICE: usize = 96;
const SLICES: i64 = 8;
/// Two per event, all disjoint within a slice.
const COMPETITORS: usize = EVENTS_PER_SLICE * 2;

/// Everything a thread count could plausibly perturb.
struct Fingerprint {
    curves: Vec<(String, Vec<(i64, Gaussian)>)>,
    log_evidence: f64,
    final_step: (f64, f64),
    iterations: usize,
}

fn build_and_converge() -> Fingerprint {
    let mut h = History::builder()
        .key_type::<String>()
        .mu(25.0)
        .sigma(25.0 / 3.0)
        .beta(25.0 / 6.0)
        .drift(ConstantDrift::new(25.0 / 300.0))
        .convergence(ConvergenceOptions {
            max_iter: 20_000,
            epsilon: 1e-9,
            alpha: 1.0,
        })
        .build();

    let mut events: Vec<Event<i64, String>> = Vec::new();
    for slice in 0..SLICES {
        for e in 0..EVENTS_PER_SLICE {
            // Disjoint within the slice: event `e` owns competitors 2e and
            // 2e+1. Rotating by the slice index makes the pairings differ
            // between slices, so competitors accumulate a real history.
            let a = (2 * e + slice as usize) % COMPETITORS;
            let b = (2 * e + 1 + slice as usize * 3) % COMPETITORS;
            if a == b {
                continue;
            }
            events.push(Event {
                time: slice + 1,
                teams: smallvec![
                    Team::with_members([Member::new(format!("p{a}"))]),
                    Team::with_members([Member::new(format!("p{b}"))]),
                ],
                outcome: Outcome::winner(u32::from((e + slice as usize) % 2 == 0), 2),
            });
        }
    }
    h.add_events(events).unwrap();

    let report = h.converge().expect("fixture must converge");

    let mut curves: Vec<(String, Vec<(i64, Gaussian)>)> = h
        .learning_curves()
        .into_iter()
        .map(|(k, v)| (k.clone(), v))
        .collect();
    curves.sort_by(|a, b| a.0.cmp(&b.0));

    Fingerprint {
        curves,
        log_evidence: h.log_evidence(),
        final_step: report.final_step,
        iterations: report.iterations,
    }
}

#[test]
fn posteriors_identical_across_thread_counts() {
    let sizes = [1usize, 2, 4, 8];
    let mut results: Vec<Fingerprint> = Vec::new();

    for &n in &sizes {
        let pool = rayon::ThreadPoolBuilder::new()
            .num_threads(n)
            .build()
            .expect("rayon pool build");
        results.push(pool.install(build_and_converge));
    }

    let reference = &results[0];

    // Guard against the failure this test previously had: passing while
    // measuring almost nothing.
    assert!(
        reference.curves.len() > 100,
        "expected every competitor's curve, got {}",
        reference.curves.len()
    );

    for (i, got) in results.iter().enumerate().skip(1) {
        let n = sizes[i];

        assert_eq!(
            got.iterations, reference.iterations,
            "iterations differ at {n} threads"
        );
        assert_eq!(
            got.final_step.0.to_bits(),
            reference.final_step.0.to_bits(),
            "final_step.0 differs at {n} threads: {:?} vs {:?}",
            reference.final_step,
            got.final_step
        );
        assert_eq!(
            got.final_step.1.to_bits(),
            reference.final_step.1.to_bits(),
            "final_step.1 differs at {n} threads"
        );
        assert_eq!(
            got.log_evidence.to_bits(),
            reference.log_evidence.to_bits(),
            "log_evidence differs at {n} threads: {} vs {}",
            reference.log_evidence,
            got.log_evidence
        );

        assert_eq!(
            got.curves.len(),
            reference.curves.len(),
            "competitor count differs at {n} threads"
        );

        for ((ref_key, ref_curve), (key, curve)) in reference.curves.iter().zip(got.curves.iter()) {
            assert_eq!(ref_key, key, "competitor order differs at {n} threads");
            assert_eq!(
                curve.len(),
                ref_curve.len(),
                "curve length differs for {key} at {n} threads"
            );
            for (&(t_ref, g_ref), &(t, g)) in ref_curve.iter().zip(curve.iter()) {
                assert_eq!(t_ref, t, "time point differs for {key} at {n} threads");
                assert_eq!(
                    g_ref.mu().to_bits(),
                    g.mu().to_bits(),
                    "mu differs for {key} at t={t}, {n} threads: {} vs {}",
                    g_ref.mu(),
                    g.mu()
                );
                assert_eq!(
                    g_ref.sigma().to_bits(),
                    g.sigma().to_bits(),
                    "sigma differs for {key} at t={t}, {n} threads: {} vs {}",
                    g_ref.sigma(),
                    g.sigma()
                );
            }
        }
    }
}

/// The fixture must keep reaching the parallel branch.
///
/// `RAYON_THRESHOLD` is private, so this pins the property that makes the
/// branch reachable rather than the branch itself: within a slice every event
/// uses a disjoint competitor pair, so greedy colouring puts all
/// `EVENTS_PER_SLICE` of them in one colour group. If someone shrinks the
/// fixture, this fails rather than the suite quietly going back to testing the
/// sequential path.
#[test]
fn the_fixture_still_exceeds_the_rayon_threshold() {
    const RAYON_THRESHOLD: usize = 64;
    const {
        assert!(
            EVENTS_PER_SLICE >= RAYON_THRESHOLD,
            "a colour group holds at most EVENTS_PER_SLICE events, which must \
             reach the crate's RAYON_THRESHOLD for the parallel sweep to run"
        );
    }

    // Measured by instrumenting `sweep_color_groups`: this fixture produces
    // one colour group of 96 events and takes the parallel branch on all 872
    // sweeps. The old fixture's 10-event slices could not reach 64 at all.
    assert_eq!(EVENTS_PER_SLICE, 96);
}