perf(tuner): keep every tuner computation within 80% of logical CPUs

crates/moon-core/src/db/mod.rs

@@ -50,7 +50,9 @@ pub use quote::{
     QuoteCurrency, QuoteScope, QuoteSpend, QuoteTotal, QuoteVolume, TradedVolume, UsdtTotal,
     ValuationCoverage,
 };
-pub use read_cancel::{ReadCancellation, current_is_cancelled, with_read_cancellation};
+pub use read_cancel::{
+    ReadCancellation, current_is_cancelled, current_read_cancellation, with_read_cancellation,
+};
 pub use read_fail::{FailCode, FailKind, ReadFail, ReadResult};
 pub use reader_budget::ReportReader;
 pub(crate) use rep::core_offset::{latest_segment, latest_source as latest_offset_source};

crates/moon-core/src/db/read_cancel.rs

@@ -82,6 +82,15 @@ pub fn with_read_cancellation<T>(cancellation: ReadCancellation, read: impl FnOn
     read()
 }
 
+/// The token of the read this thread is running, for work that hands part of that read to
+/// another thread (a pool) and must carry the token there with [`with_read_cancellation`].
+///
+/// Returns:
+///     The installed token, or `None` outside any [`with_read_cancellation`] scope.
+pub fn current_read_cancellation() -> Option<ReadCancellation> {
+    CURRENT.with(|current| current.borrow().clone())
+}
+
 /// Whether the read this thread is running has been cancelled — for work that is not an
 /// SQLite statement (a wait on another thread's answer, a per-row loop) and so never meets the
 /// progress handler.

crates/moon-core/src/db/tuner/threshold_search/mod.rs

@@ -20,7 +20,10 @@ mod tests;
 
 pub use compose::{ComposeBudget, composition_budget};
 pub use handle::SearchHandle;
-pub use search::{HEAVY_SEARCH_MIN_CORES, heavy_search_supported};
+pub use search::{
+    HEAVY_SEARCH_MIN_CORES, cap_global_pool, heavy_search_supported, install, install_job,
+    search_threads,
+};
 
 use std::cmp::Ordering;
 use std::sync::Arc;

crates/moon-core/src/db/tuner/threshold_search/search.rs

@@ -71,23 +71,75 @@ pub(super) fn logical_cores() -> usize {
     })
 }
 
-/// Share of the machine's logical CPUs the search may occupy, in percent.
+/// Share of the machine's logical CPUs ALL tuner computation together may occupy, in percent —
+/// the search pool and the column pool summed, and rayon's global pool capped to it as well.
 ///
-/// The whole point is that the rest of the PC stays usable while a search runs: a pool sized to
-/// every core but one is 98% of a 56-core machine, which is indistinguishable from all of it.
-/// A proportional cap keeps a fixed fraction of the machine free instead of a fixed single core,
-/// so a big box gives back proportionally more.
-const WORKER_CORE_PCT: usize = 90;
+/// The whole point is that the rest of the PC (live trading beside it) stays usable while the
+/// tuner works: a pool sized to every core but one is 98% of a 56-core machine, which is
+/// indistinguishable from all of it. A proportional cap keeps a fixed fraction of the machine
+/// free instead of a fixed single core, so a big box gives back proportionally more.
+const TUNER_CPU_PCT: usize = 80;
 
-/// Threads the search may use: [`WORKER_CORE_PCT`] of the logical CPUs, at most every CPU but
-/// one, and never fewer than one.
+/// Threads every tuner computation together may use on a machine of `cores` logical CPUs:
+/// [`TUNER_CPU_PCT`] of them rounded down, at most every CPU but one, and never fewer than one.
 ///
 /// Both bounds bind: the percentage is what keeps a large machine responsive, and the
-/// leave-one-free rule still covers the small end, where 90% of two cores rounds back up to
-/// both of them.
+/// leave-one-free rule still covers the small end, where the percentage of two or four cores
+/// would leave none of them free.
+///
+/// Args:
+///     cores: Logical CPUs of the machine.
+///
+/// Returns:
+///     The thread budget, at least one.
+pub(super) fn cpu_budget_of(cores: usize) -> usize {
+    (cores * TUNER_CPU_PCT / 100)
+        .min(cores.saturating_sub(1))
+        .max(1)
+}
+
+/// The budget of [`cpu_budget_of`] split into `(search, column)` pool threads.
+///
+/// The column pool takes one thread per eight of the budget, between one and
+/// [`COLUMN_THREADS`], and the search the rest, so the two together fit the budget wherever it
+/// holds two threads or more. Below that (one or two CPUs) each pool still gets its one thread:
+/// a pool of none cannot run, and a column must never queue behind a search.
+///
+/// Args:
+///     cores: Logical CPUs of the machine.
+///
+/// Returns:
+///     `(search threads, column threads)`, each at least one.
+pub(super) fn pool_split_of(cores: usize) -> (usize, usize) {
+    let budget = cpu_budget_of(cores);
+    let column = (budget / 8).clamp(1, COLUMN_THREADS);
+    (budget.saturating_sub(column).max(1), column)
+}
+
+/// Threads of the search pool ([`pool`]) on this machine: its share of the tuner's CPU budget.
 pub(super) fn worker_cores() -> usize {
-    let pct = logical_cores() * WORKER_CORE_PCT / 100;
-    pct.min(logical_cores().saturating_sub(1)).max(1)
+    pool_split_of(logical_cores()).0
+}
+
+/// Restarts the search runs side by side on this machine — its pool's thread count — for a time
+/// estimate that must not assume every CPU.
+pub fn search_threads() -> usize {
+    worker_cores()
+}
+
+/// Cap rayon's GLOBAL pool at the tuner's CPU budget, as a backstop: any parallel section that
+/// ever reaches it outside [`install`] / [`install_column`] still cannot take every CPU.
+///
+/// Call once at process start, before any rayon use; a later call (or a global pool someone
+/// already built) only logs.
+pub fn cap_global_pool() {
+    if let Err(e) = rayon::ThreadPoolBuilder::new()
+        .num_threads(cpu_budget_of(logical_cores()))
+        .thread_name(|i| format!("moon-rayon-{i}"))
+        .build_global()
+    {
+        log::warn!("tuner: rayon's global pool was not capped — {e}");
+    }
 }
 
 /// Logical cores a machine must have before the tuner offers its most expensive settings.
@@ -115,8 +167,9 @@ pub fn heavy_search_supported() -> bool {
 /// free: the search is started from a window that must keep repainting and answering the Stop button while it runs,
 /// and saturating every core makes that window stutter for the whole run.
 ///
-/// `None` means the fan-out falls back to rayon's GLOBAL pool, which is sized to every logical
-/// CPU — the search still finishes and still answers a stop, but the free core is gone with it.
+/// `None` means the fan-out falls back to rayon's GLOBAL pool, capped at the whole tuner budget
+/// by [`cap_global_pool`] (every CPU where that cap was not set) — the search still finishes and
+/// still answers a stop, but beside a column it may exceed the budget.
 /// Building a pool only fails when the process can no longer spawn threads, so this is a
 /// degraded path, not a supported mode.
 fn pool() -> Option<&'static rayon::ThreadPool> {
@@ -147,16 +200,37 @@ fn pool() -> Option<&'static rayon::ThreadPool> {
 ///
 /// Returns:
 ///     Whatever `f` returns.
-pub(crate) fn install<R: Send>(f: impl FnOnce() -> R + Send) -> R {
+pub fn install<R: Send>(f: impl FnOnce() -> R + Send) -> R {
     match pool() {
         Some(pool) => pool.install(f),
         None => f(),
     }
 }
 
-/// Threads of the pool the variant columns are scored on ([`install_column`]): a column is one
-/// replay of the sample, a fraction of a second on two threads, and a pool this small takes
-/// little from a machine whose search pool already holds most of it.
+/// Run a whole background job — a replay of the table, a readiness pass, a coin pick — inside
+/// the tuner's search pool, so its CPU counts against the tuner's budget rather than adding a
+/// thread on top of a running search.
+///
+/// The caller's read-cancellation token ([`crate::db::current_read_cancellation`]) is a
+/// thread-local, so it is carried onto the pool thread: without it the job's own
+/// `current_is_cancelled` checks would read `false` and an abandoned job would run to its end.
+///
+/// Args:
+///     f: The job.
+///
+/// Returns:
+///     Whatever `f` returns.
+pub fn install_job<R: Send>(f: impl FnOnce() -> R + Send) -> R {
+    let cancellation = crate::db::current_read_cancellation();
+    install(move || match cancellation {
+        Some(cancellation) => crate::db::with_read_cancellation(cancellation, f),
+        None => f(),
+    })
+}
+
+/// Most threads of the pool the variant columns are scored on ([`install_column`]): a column is
+/// one replay of the sample, a fraction of a second on two threads, and a pool this small takes
+/// little from the tuner's budget, the rest of which is the search's ([`pool_split_of`]).
 const COLUMN_THREADS: usize = 2;
 
 /// Worker pool for scoring the variant columns, apart from the search's ([`pool`]), or `None`
@@ -169,7 +243,7 @@ fn column_pool() -> Option<&'static rayon::ThreadPool> {
     static POOL: OnceLock<Option<rayon::ThreadPool>> = OnceLock::new();
     POOL.get_or_init(|| {
         rayon::ThreadPoolBuilder::new()
-            .num_threads(COLUMN_THREADS.min(logical_cores()).max(1))
+            .num_threads(pool_split_of(logical_cores()).1)
             .thread_name(|i| format!("moon-tuner-column-{i}"))
             .build()
             .inspect_err(|e| {

crates/moon-core/src/db/tuner/threshold_search/search/tests.rs

@@ -1160,3 +1160,50 @@ fn a_column_is_scored_while_a_search_holds_the_pool() {
     holder.join().expect("the holder ends once released");
     assert_eq!(scored, Ok(2016), "the column waited for the search's pool");
 }
+
+/// The search pool and the column pool run at the same time, so a split that sized each against
+/// the budget on its own (search = 80%, column = 2 on top) would put the tuner past 80% of the
+/// machine — 27 of 32 CPUs — beside live trading. Oracle: the percentages computed by hand.
+#[test]
+fn search_and_column_pools_together_fit_eighty_percent() {
+    for (cores, budget) in [(1, 1), (2, 1), (4, 3), (8, 6), (16, 12), (32, 25), (56, 44)] {
+        assert_eq!(cpu_budget_of(cores), budget, "{cores} CPUs");
+        let (search, column) = pool_split_of(cores);
+        assert!(
+            search >= 1 && column >= 1,
+            "{cores} CPUs: {search} + {column}"
+        );
+        if budget >= 2 {
+            assert!(
+                search + column <= budget,
+                "{cores} CPUs: {search} + {column}"
+            );
+        }
+        if cores >= 2 {
+            assert!(budget < cores, "{cores} CPUs: no CPU left free");
+        }
+    }
+    assert_eq!(pool_split_of(32), (23, 2));
+    assert_eq!(pool_split_of(4), (2, 1));
+}
+
+/// Below two CPUs of budget the floor of one thread per pool wins over the sum — a pool of none
+/// cannot run and a column must never queue behind a search — so a change that "fixed" the sum
+/// by dropping a pool to zero would hang the columns. Pinned as the stated exception.
+#[test]
+fn one_and_two_cpus_keep_one_thread_per_pool() {
+    assert_eq!(pool_split_of(1), (1, 1));
+    assert_eq!(pool_split_of(2), (1, 1));
+}
+
+/// A job moved onto the search pool loses its caller's thread-local read token unless it is
+/// carried there; an abandoned readiness pass would then replay every strategy to its end.
+#[test]
+fn install_job_carries_the_read_cancellation_onto_the_pool() {
+    let token = crate::db::ReadCancellation::new();
+    token.cancel();
+    let seen =
+        crate::db::with_read_cancellation(token, || install_job(crate::db::current_is_cancelled));
+    assert!(seen);
+    assert!(!install_job(crate::db::current_is_cancelled));
+}

crates/moon-core/src/db/tuner/ticks/search.rs

@@ -131,7 +131,11 @@ pub fn suggest_with_candidate(
     handle: &SearchHandle,
 ) -> Suggested {
     let mut candidate = None;
-    let result = search(deals, params, handle, &mut candidate);
+    // The whole search in the tuner's pool, not only its restart fan-out: the base score, the
+    // cut, the refusal count, the trim and the held-back score replay the sample in parallel too,
+    // and outside the pool they would reach rayon's global one. The nested `install` of the
+    // restarts runs in place.
+    let result = install(|| search(deals, params, handle, &mut candidate));
     let mut searched: Vec<String> = varied(params).iter().map(|f| f.key.to_string()).collect();
     searched.sort();
     Suggested {

crates/moon-ui-gpui/src/analytics/tuner/auto_picks.rs

@@ -546,7 +546,13 @@ impl AnalyticsView {
             false,
             cx,
             move || {
-                moon_core::db::tuner::coin_pick::suggest_coins_by_strategy(&q, settings, &worker)
+                // Every strategy's fit runs beside a search: in the tuner's pool, within its
+                // CPU budget.
+                moon_core::db::tuner::threshold_search::install_job(|| {
+                    moon_core::db::tuner::coin_pick::suggest_coins_by_strategy(
+                        &q, settings, &worker,
+                    )
+                })
             },
             move |this, result, cx| {
                 let (landed, retry) = this.auto_picks.land(seq, settings.holdout_pct, result);

crates/moon-ui-gpui/src/analytics/tuner/ready.rs

@@ -949,21 +949,25 @@ impl AnalyticsView {
             cx,
             move || {
                 let long_position_ms = moon_core::market::trade_replay::long_position_ms();
-                count_strategies(
-                    groups,
-                    |targets| {
-                        super::ticks::replay_batch(
-                            targets,
-                            &defaults,
-                            model,
-                            klines.as_ref(),
-                            long_position_ms,
-                            false,
-                        )
-                    },
-                    &progress,
-                    STRATEGY_PAUSE,
-                )
+                // Every strategy's replay is pure CPU and runs beside a search: in the tuner's
+                // pool it shares the search's CPU budget instead of adding a thread on top.
+                moon_core::db::tuner::threshold_search::install_job(|| {
+                    count_strategies(
+                        groups,
+                        |targets| {
+                            super::ticks::replay_batch(
+                                targets,
+                                &defaults,
+                                model,
+                                klines.as_ref(),
+                                long_position_ms,
+                                false,
+                            )
+                        },
+                        &progress,
+                        STRATEGY_PAUSE,
+                    )
+                })
             },
             move |this, counts, cx| this.finish_ready_pass(key, seq, Ok(counts), cx),
         );

crates/moon-ui-gpui/src/analytics/tuner/ticks/estimate.rs

@@ -102,10 +102,18 @@ pub(super) fn remaining_time(
     }
 }
 
-/// How many of a search's restarts run side by side: the machine's threads bound them.
+/// How many of a search's restarts run side by side: the search pool's threads bound them,
+/// not every CPU of the machine.
 fn side_by_side(restarts: usize) -> usize {
-    let threads = std::thread::available_parallelism().map_or(1, |n| n.get());
-    restarts.clamp(1, threads)
+    side_by_side_on(
+        restarts,
+        moon_core::db::tuner::threshold_search::search_threads(),
+    )
+}
+
+/// [`side_by_side`] on a pool of `threads` threads.
+fn side_by_side_on(restarts: usize, threads: usize) -> usize {
+    restarts.clamp(1, threads.max(1))
 }
 
 /// What one search varies: whether each group is searched, and the fields it holds.

crates/moon-ui-gpui/src/analytics/tuner/ticks/estimate/tests.rs

@@ -1,6 +1,20 @@
 use std::time::Duration;
 
-use super::{count_text, duration_text, pick_all_label};
+use super::{count_text, duration_text, pick_all_label, side_by_side, side_by_side_on};
+
+/// An estimate that assumed every CPU runs restarts would promise a 32-CPU box finishes ~28%
+/// sooner than its 23-thread search pool can: side by side is bounded by the pool, not the box.
+#[test]
+fn side_by_side_is_bounded_by_the_search_pool() {
+    assert_eq!(side_by_side_on(500, 23), 23);
+    assert_eq!(side_by_side_on(5, 23), 5);
+    assert_eq!(side_by_side_on(0, 23), 1);
+    assert_eq!(side_by_side_on(500, 0), 1);
+    assert_eq!(
+        side_by_side(100_000),
+        moon_core::db::tuner::threshold_search::search_threads()
+    );
+}
 
 /// A separate startup formula would drift from the measured running ETA after sample changes.
 #[test]