diff --git a/Cargo.lock b/Cargo.lock index fc93401..af35e8a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -168,6 +168,7 @@ dependencies = [ "aura-std", "chrono", "chrono-tz", + "rayon", "serde", "serde_json", ] @@ -404,6 +405,25 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossbeam-deque" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -479,6 +499,12 @@ dependencies = [ "syn", ] +[[package]] +name = "either" +version = "1.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" + [[package]] name = "equivalent" version = "1.0.2" @@ -803,6 +829,26 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "regex" version = "1.12.3" diff --git a/crates/aura-engine/Cargo.toml b/crates/aura-engine/Cargo.toml index c59bb46..02674ae 100644 --- a/crates/aura-engine/Cargo.toml +++ b/crates/aura-engine/Cargo.toml @@ -19,6 +19,14 @@ serde = { workspace = true } # serde_json renders RunReport JSON for both the registry (disk) and stdout # (RunReport::to_json) — one shape, no hand-rolled writer (amended C16). serde_json = { workspace = true } +# rayon is admitted under the amended C16 per-case dependency policy (INDEX.md): +# it backs `run_indexed`'s work-stealing pool for the sweep/campaign executor. +# It stays within C1 (parallelism across sims, never within one) — each sim +# still runs its own deterministic, synchronous event loop on one thread; rayon +# only fans the *disjoint* sims of a family out across threads, replacing +# `std::thread::scope`. No dep on a workspace precedent (chrono's pattern: +# direct versioned dep). +rayon = "1" [dev-dependencies] # aura-std is a TEST-ONLY dependency: the engine's own integration tests diff --git a/crates/aura-engine/src/mc.rs b/crates/aura-engine/src/mc.rs index 730dd6b..f4450d4 100644 --- a/crates/aura-engine/src/mc.rs +++ b/crates/aura-engine/src/mc.rs @@ -110,6 +110,26 @@ fn quantile(sorted: &[f64], p: f64) -> f64 { } } +/// The single assembly path behind both [`monte_carlo`] and the thread-count- +/// explicit test wrapper: run `run_one` over every seed via `run_indexed` (on +/// whichever pool is ambient when called), build the per-seed draws in +/// seed-input order, and reduce them into the aggregate. One source of truth so +/// the two callers can never drift on assembly logic while their pool selection +/// differs. +fn assemble_mc(base_point: &[Scalar], seeds: &[u64], run_one: &F) -> McFamily +where + F: Fn(u64, &[Scalar]) -> RunReport + Sync, +{ + let reports = run_indexed(seeds.len(), |i| run_one(seeds[i], base_point)); + let draws: Vec = seeds + .iter() + .zip(reports) + .map(|(&seed, report)| McDraw { seed, report }) + .collect(); + let aggregate = McAggregate::from_draws(&draws); + McFamily { draws, aggregate } +} + /// Run `run_one(seed, base_point)` over every seed, disjointly in parallel (C1), /// and collect the family in seed-input order plus the aggregate. The varying /// dimension is the *seed* (C12 axis 4: MC = sweep over seeds), not a tuning @@ -124,15 +144,18 @@ pub fn monte_carlo(base_point: &[Scalar], seeds: &[u64], run_one: F) -> McFam where F: Fn(u64, &[Scalar]) -> RunReport + Sync, { - let nthreads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1); - monte_carlo_with_threads(base_point, seeds, nthreads, run_one) + debug_assert!(!seeds.is_empty(), "monte_carlo requires a non-empty seed set"); + assemble_mc(base_point, seeds, &run_one) } -/// The thread-count-explicit core of [`monte_carlo`]. Module-private: the public -/// `monte_carlo` derives the count, while the tests drive it at 1 and at N to pin -/// determinism under concurrency (C1). A thin adapter over -/// [`run_indexed`](crate::sweep): it runs each seed disjointly and zips the -/// reports back onto their seeds in seed-input order. +/// The thread-count-explicit wrapper of [`monte_carlo`]. Module-private: the +/// public `monte_carlo` runs on the ambient pool; the tests drive this at 1 and +/// at N to pin determinism (C1). `assemble_mc` (run_indexed + draw assembly + +/// aggregate) runs inside a local rayon pool of `nthreads` workers, sharing the +/// exact assembly path `monte_carlo` uses — no second copy to drift. Bounds +/// match `monte_carlo` (`F: Sync`): the install closure only borrows `run_one`, +/// never moves it. +#[cfg(test)] fn monte_carlo_with_threads( base_point: &[Scalar], seeds: &[u64], @@ -143,14 +166,11 @@ where F: Fn(u64, &[Scalar]) -> RunReport + Sync, { debug_assert!(!seeds.is_empty(), "monte_carlo requires a non-empty seed set"); - let reports = run_indexed(seeds.len(), nthreads, |i| run_one(seeds[i], base_point)); - let draws: Vec = seeds - .iter() - .zip(reports) - .map(|(&seed, report)| McDraw { seed, report }) - .collect(); - let aggregate = McAggregate::from_draws(&draws); - McFamily { draws, aggregate } + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(nthreads) + .build() + .expect("rayon thread pool"); + pool.install(|| assemble_mc(base_point, seeds, &run_one)) } /// Distribution of `E[R]` under a moving-block bootstrap of an OOS per-trade R series diff --git a/crates/aura-engine/src/sweep.rs b/crates/aura-engine/src/sweep.rs index c348e1e..257e23a 100644 --- a/crates/aura-engine/src/sweep.rs +++ b/crates/aura-engine/src/sweep.rs @@ -12,7 +12,6 @@ use aura_core::{zip_params, Cell, ParamSpec, Scalar, ScalarKind}; use crate::RunReport; use crate::harness::SplitMix64; -use std::sync::atomic::{AtomicUsize, Ordering}; /// A validated cartesian grid over a blueprint's param-space: one discrete /// value-list per param slot, in `param_space()` order. Enumerates a family of @@ -366,77 +365,20 @@ impl SweepFamily { } } -/// Run `run_one` over every point the `space` produces (`Space::points` — -/// `GridSpace`'s cartesian product or `RandomSpace`'s seeded draws), disjointly -/// in parallel (C1), and collect the family in enumeration order. `run_one` -/// builds + bootstraps + runs + -/// summarizes one point; it shares nothing mutable, so it is `Sync` and the runs -/// are lock-free. Parallelism is `available_parallelism()` workers — `std` only, -/// via `std::thread::scope`. -pub fn sweep(space: &S, run_one: F) -> SweepFamily +/// The single assembly path behind both [`sweep`] and the thread-count-explicit +/// test wrapper: run `run_one` over each already-enumerated `point` via +/// `run_indexed` (on whichever pool is ambient when called — the process-wide +/// global pool, or a locally `install`ed one) and collect the family in +/// enumeration order. Kept as one source of truth so the two callers can never +/// drift on assembly logic (grid points, `RandomSpace` draws) while their pool +/// selection differs. +fn assemble_sweep(space: Vec, points: Vec>, run_one: &F) -> SweepFamily where - S: Space, F: Fn(&[Cell]) -> RunReport + Sync, { - let nthreads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1); - sweep_with_threads(space, nthreads, run_one) -} - -/// Run `n` disjoint jobs in parallel and collect their results in **job-index -/// order** — the shared disjoint-parallel core both `sweep` (over grid points) -/// and `monte_carlo` (over seeds) drive (C1: order is the input order, not the -/// completion order). Workers pull job indices from a shared atomic cursor -/// (work-stealing load-balances uneven per-job cost); each tags its result with -/// the index, and the results are sorted on that index after the scope joins. -/// Only the cursor is shared; the results side is lock-free. `nthreads` is -/// clamped to `[1, n.max(1)]` (a 0-job call yields an empty vec; a 0 thread -/// count coerces to 1). -pub(crate) fn run_indexed(n: usize, nthreads: usize, run_one: F) -> Vec -where - T: Send, - F: Fn(usize) -> T + Sync, -{ - let nthreads = nthreads.clamp(1, n.max(1)); - let cursor = AtomicUsize::new(0); - - let mut results: Vec<(usize, T)> = std::thread::scope(|scope| { - let handles: Vec<_> = (0..nthreads) - .map(|_| { - scope.spawn(|| { - let mut local: Vec<(usize, T)> = Vec::new(); - loop { - let i = cursor.fetch_add(1, Ordering::Relaxed); - if i >= n { - break; - } - local.push((i, run_one(i))); - } - local - }) - }) - .collect(); - handles.into_iter().flat_map(|h| h.join().unwrap()).collect() - }); - - results.sort_by_key(|&(i, _)| i); - results.into_iter().map(|(_, t)| t).collect() -} - -/// The thread-count-explicit core of [`sweep`]. Module-private: the public -/// `sweep` derives the count, while the tests drive it at 1 and at N to pin -/// determinism under concurrency (C1). A thin adapter over [`run_indexed`]: it -/// enumerates the `space`'s param-space points (`Space::points`), runs each -/// disjointly, and zips the reports back onto their points in enumeration -/// (odometer) order. -fn sweep_with_threads(space: &S, nthreads: usize, run_one: F) -> SweepFamily -where - S: Space, - F: Fn(&[Cell]) -> RunReport + Sync, -{ - let points = space.points(); - let reports = run_indexed(points.len(), nthreads, |i| run_one(&points[i])); + let reports = run_indexed(points.len(), |i| run_one(&points[i])); SweepFamily { - space: space.param_specs().to_vec(), + space, points: points .into_iter() .zip(reports) @@ -445,6 +387,74 @@ where } } +/// Run `run_one` over every point the `space` produces (`Space::points` — +/// `GridSpace`'s cartesian product or `RandomSpace`'s seeded draws), disjointly +/// in parallel (C1), and collect the family in enumeration order. `run_one` +/// builds + bootstraps + runs + summarizes one point; it shares nothing +/// mutable, so it is `Sync` and the runs are lock-free. Parallelism is the +/// shared rayon pool (`run_indexed`): a nested sweep / walk-forward shares its +/// workers rather than each spawning its own `available_parallelism()` +/// OS-thread batch. +pub fn sweep(space: &S, run_one: F) -> SweepFamily +where + S: Space, + F: Fn(&[Cell]) -> RunReport + Sync, +{ + let points = space.points(); + assemble_sweep(space.param_specs().to_vec(), points, &run_one) +} + +/// Run `n` disjoint jobs in parallel and collect their results in **job-index +/// order** — the shared disjoint-parallel core that `sweep` (over grid points), +/// `monte_carlo` (over seeds), and `walk_forward` (over window bounds) drive +/// (C1: order is the input order, not the completion order). Jobs run on the +/// ambient rayon pool: the process-wide global pool at top level, or — inside a +/// `ThreadPool::install` scope (the `*_with_threads` test wrappers) — that local +/// pool. Nested `run_indexed` calls (a `walk_forward` window running an inner +/// `sweep`) share the same workers via work-stealing, so total live workers +/// never exceed the pool size — no per-call OS-thread batch, no multiplicative +/// oversubscription. `IndexedParallelIterator::collect` restores enumeration +/// order regardless of which worker ran which job; a 0-job call yields an empty +/// vec. `run_one` is shared by reference across workers (`Sync`), never moved. +// `run_one` is captured by reference in the `map` closure below (needs only +// `F: Sync`, not `F: Send`): the seemingly-redundant closure is load-bearing — +// passing `run_one` directly would move it into `Map` and require `F: Send`, a +// bound the `_with_threads` wrappers' `install`-borrowed callers don't carry. +#[allow(clippy::redundant_closure)] +pub(crate) fn run_indexed(n: usize, run_one: F) -> Vec +where + T: Send, + F: Fn(usize) -> T + Sync, +{ + use rayon::prelude::*; + (0..n).into_par_iter().map(|i| run_one(i)).collect() +} + +/// The thread-count-explicit wrapper of [`sweep`]. Module-private: the public +/// `sweep` runs on the ambient pool, while the tests drive this at 1 and at N +/// to pin determinism under concurrency (C1). The point enumeration runs +/// outside the pool; `assemble_sweep` (run_indexed + family assembly) runs +/// inside a local rayon pool of `nthreads` workers via `ThreadPool::install`, +/// sharing the exact assembly path `sweep` uses — no second copy to drift. A +/// 1-worker pool runs the jobs inline (no deadlock) — the sequential-equivalent +/// reference. Bounds match `sweep` (`F: Sync`): the install closure only +/// *borrows* `run_one` and moves the already-owned `space`/`points`, so no +/// `Send` bound on `F` is needed. +#[cfg(test)] +fn sweep_with_threads(space: &S, nthreads: usize, run_one: F) -> SweepFamily +where + S: Space, + F: Fn(&[Cell]) -> RunReport + Sync, +{ + let points = space.points(); + let param_specs = space.param_specs().to_vec(); + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(nthreads) + .build() + .expect("rayon thread pool"); + pool.install(|| assemble_sweep(param_specs, points, &run_one)) +} + #[cfg(test)] mod tests { use super::*; @@ -803,6 +813,68 @@ mod tests { assert_eq!(one, sweep(&grid, run_point)); } + #[test] + fn nested_run_indexed_shares_the_pool() { + use std::sync::atomic::{AtomicUsize, Ordering}; + // A shared gauge: live leaf jobs on entry, max seen. Under one shared pool, + // nested run_indexed (outer jobs each spawning an inner run_indexed) must + // never exceed the pool size — the pre-rayon per-call scope would have run + // POOL*POOL threads here. The sleep lets concurrency actually build; the + // assertion is an upper bound, so the test is not timing-flaky. + const POOL: usize = 4; + let live = AtomicUsize::new(0); + let max = AtomicUsize::new(0); + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(POOL) + .build() + .expect("rayon thread pool"); + pool.install(|| { + run_indexed(POOL * 2, |_outer| { + run_indexed(POOL * 2, |_inner| { + let now = live.fetch_add(1, Ordering::SeqCst) + 1; + max.fetch_max(now, Ordering::SeqCst); + std::thread::sleep(std::time::Duration::from_millis(2)); + live.fetch_sub(1, Ordering::SeqCst); + 0u8 + }) + }); + }); + let peak = max.load(Ordering::SeqCst); + assert!( + peak <= POOL, + "nested run_indexed peaked at {peak} live workers, exceeding the {POOL}-worker pool" + ); + } + + #[test] + #[ignore = "benchmark: cargo test -p aura-engine bench_nested_pool -- --ignored --nocapture"] + fn bench_nested_pool_walltime() { + use std::time::Instant; + // Synthetic nested workload: an outer fan-out whose each job runs an inner + // fan-out of CPU-bound leaves — the walk_forward(windows) -> sweep(members) + // shape, without any archive. Measures wall time under the shared pool; run + // before/after the rayon swap to see the oversubscription collapse. Not a + // committed baseline (that is #251). + fn busy(rounds: u64) -> u64 { + (0..rounds).fold(0u64, |a, x| a.wrapping_add(x.wrapping_mul(x).rotate_left(7))) + } + const OUTER: usize = 48; + const INNER: usize = 12; + let start = Instant::now(); + let checksum: u64 = run_indexed(OUTER, |_w| { + run_indexed(INNER, |i| busy(4_000_000 + i as u64)) + .into_iter() + .fold(0u64, u64::wrapping_add) + }) + .into_iter() + .fold(0u64, u64::wrapping_add); + let elapsed = start.elapsed(); + println!( + "nested run_indexed ({OUTER}x{INNER} CPU-bound leaves): {elapsed:?} \ + (checksum {checksum})" + ); + } + #[test] fn distinct_points_produce_distinct_metrics() { let family = sweep(&sma_cross_grid(), run_point); diff --git a/crates/aura-engine/src/walkforward.rs b/crates/aura-engine/src/walkforward.rs index baae39e..2564ef8 100644 --- a/crates/aura-engine/src/walkforward.rs +++ b/crates/aura-engine/src/walkforward.rs @@ -174,6 +174,33 @@ impl WalkForwardResult { } } +/// The single assembly path behind both [`walk_forward`] and the thread-count- +/// explicit test wrapper: run `run_window` over each already-rolled `bounds` via +/// `run_indexed` (on whichever pool is ambient when called), stitch the OOS +/// equity, and assemble the result. One source of truth so the two callers can +/// never drift on assembly logic while their pool selection differs. +fn assemble_walk_forward( + space: Vec, + bounds: Vec, + run_window: &F, +) -> WalkForwardResult +where + F: Fn(WindowBounds) -> WindowRun + Sync, +{ + let runs = run_indexed(bounds.len(), |i| run_window(bounds[i])); + let windows: Vec = bounds + .into_iter() + .zip(runs) + .map(|(bounds, run)| WindowOutcome { bounds, run }) + .collect(); + let stitched_oos_equity = stitch(&windows); + debug_assert!( + windows.iter().all(|w| w.run.chosen_params.len() == space.len()), + "every window's chosen point must match the param-space arity (same blueprint)" + ); + WalkForwardResult { space, windows, stitched_oos_equity } +} + /// Roll the windows, run `run_window` on each disjointly in parallel (C1, via the /// shared [`run_indexed`](crate::sweep) core), then stitch the OOS equity into one /// continuous curve. The varying dimension is the *data window* (C12 axis 3). @@ -186,16 +213,20 @@ pub fn walk_forward(roller: WindowRoller, space: Vec, run_window: where F: Fn(WindowBounds) -> WindowRun + Sync, { - let nthreads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1); - walk_forward_with_threads(roller, space, nthreads, run_window) + let bounds: Vec = roller.collect(); + debug_assert!(!bounds.is_empty(), "walk_forward requires a non-empty roll"); + assemble_walk_forward(space, bounds, &run_window) } -/// The thread-count-explicit core of [`walk_forward`]. Module-private: the public -/// `walk_forward` derives the count, while the tests drive it at 1 and at N to pin -/// determinism under concurrency (C1). A thin adapter over -/// [`run_indexed`](crate::sweep): it collects the roller's bounds (tiny — four -/// timestamps each, no tick data), runs each window disjointly, zips the runs back -/// onto their bounds in roll order, and stitches the OOS segments. +/// The thread-count-explicit wrapper of [`walk_forward`]. Module-private: the +/// public `walk_forward` runs on the ambient pool; the tests drive this at 1 +/// and at N to pin determinism (C1). The roll enumeration runs outside the +/// pool; `assemble_walk_forward` (run_indexed + stitch + result assembly) runs +/// inside a local rayon pool of `nthreads` workers, sharing the exact assembly +/// path `walk_forward` uses — no second copy to drift. Bounds match +/// `walk_forward` (`F: Sync`): the install closure only borrows `run_window`, +/// never moves it. +#[cfg(test)] fn walk_forward_with_threads( roller: WindowRoller, space: Vec, @@ -207,18 +238,11 @@ where { let bounds: Vec = roller.collect(); debug_assert!(!bounds.is_empty(), "walk_forward requires a non-empty roll"); - let runs = run_indexed(bounds.len(), nthreads, |i| run_window(bounds[i])); - let windows: Vec = bounds - .into_iter() - .zip(runs) - .map(|(bounds, run)| WindowOutcome { bounds, run }) - .collect(); - let stitched_oos_equity = stitch(&windows); - debug_assert!( - windows.iter().all(|w| w.run.chosen_params.len() == space.len()), - "every window's chosen point must match the param-space arity (same blueprint)" - ); - WalkForwardResult { space, windows, stitched_oos_equity } + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(nthreads) + .build() + .expect("rayon thread pool"); + pool.install(|| assemble_walk_forward(space, bounds, &run_window)) } /// Fold the per-window OOS equity segments into one continuous curve: each