diff --git a/crates/moon-core/src/db/mod.rs b/crates/moon-core/src/db/mod.rs index d33940284..df8f81f62 100644 --- a/crates/moon-core/src/db/mod.rs +++ b/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}; diff --git a/crates/moon-core/src/db/read_cancel.rs b/crates/moon-core/src/db/read_cancel.rs index 56e3db396..740e2f682 100644 --- a/crates/moon-core/src/db/read_cancel.rs +++ b/crates/moon-core/src/db/read_cancel.rs @@ -82,6 +82,15 @@ pub fn with_read_cancellation(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 { + 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. diff --git a/crates/moon-core/src/db/tuner/threshold_search/mod.rs b/crates/moon-core/src/db/tuner/threshold_search/mod.rs index c8f89ca7c..0d73ef085 100644 --- a/crates/moon-core/src/db/tuner/threshold_search/mod.rs +++ b/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; diff --git a/crates/moon-core/src/db/tuner/threshold_search/search.rs b/crates/moon-core/src/db/tuner/threshold_search/search.rs index 2a7caea83..d42300ebd 100644 --- a/crates/moon-core/src/db/tuner/threshold_search/search.rs +++ b/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(f: impl FnOnce() -> R + Send) -> R { +pub fn install(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(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> = 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| { diff --git a/crates/moon-core/src/db/tuner/threshold_search/search/tests.rs b/crates/moon-core/src/db/tuner/threshold_search/search/tests.rs index 3b4e17edd..e874196b6 100644 --- a/crates/moon-core/src/db/tuner/threshold_search/search/tests.rs +++ b/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)); +} diff --git a/crates/moon-core/src/db/tuner/ticks/search.rs b/crates/moon-core/src/db/tuner/ticks/search.rs index fdbb6e3ba..bf790e3a9 100644 --- a/crates/moon-core/src/db/tuner/ticks/search.rs +++ b/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 = varied(params).iter().map(|f| f.key.to_string()).collect(); searched.sort(); Suggested { diff --git a/crates/moon-ui-gpui/src/analytics/tuner/auto_picks.rs b/crates/moon-ui-gpui/src/analytics/tuner/auto_picks.rs index f2fce0bd4..970e26b2f 100644 --- a/crates/moon-ui-gpui/src/analytics/tuner/auto_picks.rs +++ b/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); diff --git a/crates/moon-ui-gpui/src/analytics/tuner/ready.rs b/crates/moon-ui-gpui/src/analytics/tuner/ready.rs index ed766a318..ff1e6674e 100644 --- a/crates/moon-ui-gpui/src/analytics/tuner/ready.rs +++ b/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), ); diff --git a/crates/moon-ui-gpui/src/analytics/tuner/ticks/estimate.rs b/crates/moon-ui-gpui/src/analytics/tuner/ticks/estimate.rs index 253d8b799..8cbaf2be8 100644 --- a/crates/moon-ui-gpui/src/analytics/tuner/ticks/estimate.rs +++ b/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. diff --git a/crates/moon-ui-gpui/src/analytics/tuner/ticks/estimate/tests.rs b/crates/moon-ui-gpui/src/analytics/tuner/ticks/estimate/tests.rs index 21319310f..1e78d9103 100644 --- a/crates/moon-ui-gpui/src/analytics/tuner/ticks/estimate/tests.rs +++ b/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] diff --git a/crates/moon-ui-gpui/src/analytics/tuner/ticks/load.rs b/crates/moon-ui-gpui/src/analytics/tuner/ticks/load.rs index c9a1c9ace..a63502989 100644 --- a/crates/moon-ui-gpui/src/analytics/tuner/ticks/load.rs +++ b/crates/moon-ui-gpui/src/analytics/tuner/ticks/load.rs @@ -494,47 +494,51 @@ impl AnalyticsView { &[ReadLane::TicksReplay], false, cx, + // The table's replay is CPU and may run beside a search: in the tuner's pool, within + // its CPU budget. move || { - // One threshold for the whole table, read once: every row of one load splits - // its window the same way. - let long_position_ms = moon_core::market::trade_replay::long_position_ms(); - let mut rows = replay_batch( - targets, - &defaults, - model, - klines.as_ref(), - long_position_ms, - true, - ); - log_replay(&rows); - // Rows the job answered while the batch was read: read again, each behind - // whatever walk is running. A row the job answers during THIS loop is kept - // covered by the fold (`update_rows`), not re-read once more. - let late = super::fetch::job::finished_after(reading_since); - if !late.is_empty() { - let traces = archived_lines(&rows); - for row in rows - .iter_mut() - .filter(|r| late.contains(&r.deal.report_uid)) - { - if moon_core::db::current_is_cancelled() { - break; + moon_core::db::tuner::threshold_search::install_job(|| { + // One threshold for the whole table, read once: every row of one load splits + // its window the same way. + let long_position_ms = moon_core::market::trade_replay::long_position_ms(); + let mut rows = replay_batch( + targets, + &defaults, + model, + klines.as_ref(), + long_position_ms, + true, + ); + log_replay(&rows); + // Rows the job answered while the batch was read: read again, each behind + // whatever walk is running. A row the job answers during THIS loop is kept + // covered by the fold (`update_rows`), not re-read once more. + let late = super::fetch::job::finished_after(reading_since); + if !late.is_empty() { + let traces = archived_lines(&rows); + for row in rows + .iter_mut() + .filter(|r| late.contains(&r.deal.report_uid)) + { + if moon_core::db::current_is_cancelled() { + break; + } + let lines = traces + .get(&row.deal.report_uid) + .cloned() + .unwrap_or_default(); + replay_row( + row, + &defaults, + model, + lines, + long_position_ms, + klines.as_ref(), + ); } - let lines = traces - .get(&row.deal.report_uid) - .cloned() - .unwrap_or_default(); - replay_row( - row, - &defaults, - model, - lines, - long_position_ms, - klines.as_ref(), - ); } - } - rows + rows + }) }, move |this, rows, cx| { if this.ticks.seq != req || this.ticks.tape_seq != tape_req { diff --git a/crates/moon-ui-gpui/src/startup.rs b/crates/moon-ui-gpui/src/startup.rs index 3ee31350d..cd8b96a11 100644 --- a/crates/moon-ui-gpui/src/startup.rs +++ b/crates/moon-ui-gpui/src/startup.rs @@ -387,6 +387,9 @@ pub(crate) fn run(startup_update: Option) -> anyho if let Some(e) = diag_err { log::warn!("{e}"); } + // Before any rayon use, and after the logger so a refusal is recorded: a parallel section + // that ever escapes the tuner's own pools lands in a global pool held to the same CPU budget. + moon_core::db::tuner::threshold_search::cap_global_pool(); log::info!( "build: moonterminal={} release_base={} moonui={}", option_env!("MOONTERMINAL_GIT_REV").unwrap_or("unknown"),