From c11b38f2d6b260bff4abc9e45480c9bcc96e9ea3 Mon Sep 17 00:00:00 2001 From: Evgeny Formanenko Date: Fri, 16 Oct 2026 15:39:22 +0300 Subject: [PATCH] fix(hotblocks): give each shadow run its own thread A shadow thread that waits for a run's parallel work runs other jobs of its pool meanwhile, and in the shared shadow pool those jobs are whole other runs, so one run's time could hold several. Runs now go to single-thread workers, as query tasks do with a thread per task. Runs where the legacy engine is the slow one are counted too: the slow-run count alone cannot tell a slow engine from a slow host. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/hotblocks/src/cli.rs | 3 +- crates/hotblocks/src/metrics.rs | 29 ++++++- crates/hotblocks/src/query/executor.rs | 20 +++-- crates/hotblocks/src/query/mod.rs | 1 + crates/hotblocks/src/shadow/mod.rs | 104 ++++++++++++++++++++----- 5 files changed, 123 insertions(+), 34 deletions(-) diff --git a/crates/hotblocks/src/cli.rs b/crates/hotblocks/src/cli.rs index 8bde07a..51bd0ef 100644 --- a/crates/hotblocks/src/cli.rs +++ b/crates/hotblocks/src/cli.rs @@ -175,7 +175,8 @@ pub struct CLI { #[arg(long, value_name = "RATIO", default_value = "2")] pub shadow_slow_ratio: f64, - /// ...and at least this many milliseconds longer. + /// ...and at least this many milliseconds longer. The same rule with the + /// engines swapped counts the runs where the legacy engine is slow. #[arg(long, value_name = "MS", default_value = "50")] pub shadow_slow_margin_ms: u64, diff --git a/crates/hotblocks/src/metrics.rs b/crates/hotblocks/src/metrics.rs index d6cd071..cd3cb41 100644 --- a/crates/hotblocks/src/metrics.rs +++ b/crates/hotblocks/src/metrics.rs @@ -234,6 +234,13 @@ impl ShadowOutcome { } } +/// The engine whose time is over the slow ratio and margin of the other's. +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +pub(crate) enum ShadowSlower { + Legacy, + New +} + #[derive(Copy, Clone, Hash, Debug, Eq, PartialEq, EncodeLabelSet)] struct ShadowRunLabels { dataset: DatasetValue, @@ -249,6 +256,7 @@ struct ShadowEngineLabels { static SHADOW_QUERIES: LazyLock> = LazyLock::new(Default::default); static SHADOW_RUNS: LazyLock> = LazyLock::new(Default::default); static SHADOW_SLOW_RUNS: LazyLock> = LazyLock::new(Default::default); +static SHADOW_LEGACY_SLOW_RUNS: LazyLock> = LazyLock::new(Default::default); static SHADOW_RUN_DURATION: LazyLock> = LazyLock::new(|| Family::new_with_constructor(|| Histogram::new(buckets(0.0001, 24)))); static SHADOW_PERCENT: LazyLock> = LazyLock::new(Default::default); @@ -266,7 +274,12 @@ pub(crate) fn report_shadow_run(dataset_id: DatasetId, outcome: ShadowOutcome) { SHADOW_RUNS.get_or_create(&labels).inc(); } -pub(crate) fn report_shadow_timing(dataset_id: DatasetId, legacy: Duration, new: Duration, slow: bool) { +pub(crate) fn report_shadow_timing( + dataset_id: DatasetId, + legacy: Duration, + new: Duration, + slower: Option +) { for (engine, duration) in [("legacy", legacy), ("new", new)] { let labels = ShadowEngineLabels { dataset: DatasetValue(dataset_id), @@ -276,9 +289,12 @@ pub(crate) fn report_shadow_timing(dataset_id: DatasetId, legacy: Duration, new: .get_or_create(&labels) .observe(duration.as_secs_f64()); } - if slow { - SHADOW_SLOW_RUNS.get_or_create(&dataset_label!(dataset_id)).inc(); - } + let slow_runs = match slower { + Some(ShadowSlower::New) => &SHADOW_SLOW_RUNS, + Some(ShadowSlower::Legacy) => &SHADOW_LEGACY_SLOW_RUNS, + None => return + }; + slow_runs.get_or_create(&dataset_label!(dataset_id)).inc(); } pub(crate) fn set_shadow_percent(percent: f64) { @@ -793,6 +809,11 @@ pub fn build_metrics_registry() -> Registry { "Shadow runs where the new engine exceeded the slow ratio and margin", SHADOW_SLOW_RUNS.clone() ); + registry.register( + "shadow_legacy_slow_runs", + "Shadow runs where the legacy engine exceeded the slow ratio and margin", + SHADOW_LEGACY_SLOW_RUNS.clone() + ); registry.register( "shadow_run_duration_seconds", "Time each engine took to answer one chunk of a shadowed query", diff --git a/crates/hotblocks/src/query/executor.rs b/crates/hotblocks/src/query/executor.rs index 902befb..ab4de36 100644 --- a/crates/hotblocks/src/query/executor.rs +++ b/crates/hotblocks/src/query/executor.rs @@ -47,13 +47,14 @@ type Job = Box; /// run on the task's thread. In the shared pool they spread over every worker, /// and a new task waits in the injector until a heavy query has no parallel work /// left: rayon takes injected jobs last. -struct TaskQueue { +pub(crate) struct TaskQueue { sender: mpsc::Sender, _workers: Vec } impl TaskQueue { - fn new(threads: usize) -> anyhow::Result { + /// `threads` workers, named `-`. + pub(crate) fn new(threads: usize, name: &'static str) -> anyhow::Result { let (sender, receiver) = mpsc::channel::(); let receiver = Arc::new(Mutex::new(receiver)); @@ -61,7 +62,7 @@ impl TaskQueue { .map(|i| { let worker = rayon::ThreadPoolBuilder::new() .num_threads(1) - .thread_name(move |_| format!("query-task-{i}")) + .thread_name(move |_| format!("{name}-{i}")) .build()?; let receiver = receiver.clone(); worker.spawn(move || { @@ -81,6 +82,12 @@ impl TaskQueue { _workers: workers }) } + + pub(crate) fn spawn(&self, job: impl FnOnce() + Send + 'static) { + self.sender + .send(Box::new(job)) + .expect("the workers outlive the queue's sender") + } } /// Where a query task runs, chosen when its slot is taken. @@ -94,10 +101,7 @@ impl TaskPool { fn spawn(&self, job: impl FnOnce() + Send + 'static) { match self { Self::Shared => sqd_polars::POOL.spawn(job), - Self::ThreadPerTask(queue) => queue - .sender - .send(Box::new(job)) - .expect("the workers outlive the queue's sender") + Self::ThreadPerTask(queue) => queue.spawn(job) } } } @@ -129,7 +133,7 @@ impl QueryExecutor { settings: SettingsRef, task_workers: usize ) -> anyhow::Result { - let task_queue = TaskQueue::new(task_workers)?; + let task_queue = TaskQueue::new(task_workers, "query-task")?; Ok(Self { in_flight: Arc::new(AtomicUsize::new(0)), max_pending_tasks, diff --git a/crates/hotblocks/src/query/mod.rs b/crates/hotblocks/src/query/mod.rs index e1b4810..bd7e0aa 100644 --- a/crates/hotblocks/src/query/mod.rs +++ b/crates/hotblocks/src/query/mod.rs @@ -6,6 +6,7 @@ mod slow_tasks; mod static_snapshot; pub use executor::QueryExecutorCollector; +pub(crate) use executor::TaskQueue; pub use response::*; pub use service::*; pub use static_snapshot::StaticChunkReader; diff --git a/crates/hotblocks/src/shadow/mod.rs b/crates/hotblocks/src/shadow/mod.rs index 74ad4de..e25d933 100644 --- a/crates/hotblocks/src/shadow/mod.rs +++ b/crates/hotblocks/src/shadow/mod.rs @@ -9,7 +9,8 @@ //! //! A difference in values, an error only one engine raises, and a run where the //! new engine is much slower are logged with the query that reproduces them and -//! counted in metrics. +//! counted in metrics. A run where the legacy engine is much slower is counted +//! too: a slow-run count means little without its mirror. mod compare; @@ -35,10 +36,10 @@ use self::compare::Difference; use crate::{ log_window::{LogWindow, query_fingerprint, truncate_query}, metrics::{ - ShadowOutcome, report_shadow_query, report_shadow_run, report_shadow_run_finished, report_shadow_run_started, - report_shadow_timing + ShadowOutcome, ShadowSlower, report_shadow_query, report_shadow_run, report_shadow_run_finished, + report_shadow_run_started, report_shadow_timing }, - query::StaticChunkReader, + query::{StaticChunkReader, TaskQueue}, query_engine::{self, MetadataCatalogs}, settings::{SettingsRef, ShadowSettings}, types::DatasetKind @@ -52,25 +53,20 @@ pub struct Shadow { catalogs: Arc, /// Runs get their own threads, so that they never queue ahead of a client's /// query. The legacy engine's polars operations still use the shared pool. - pool: rayon::ThreadPool, + threads: TaskQueue, in_flight: AtomicUsize, log_window: Mutex } impl Shadow { pub fn new(max_in_flight: usize, settings: SettingsRef, catalogs: Arc) -> anyhow::Result { - let pool = rayon::ThreadPoolBuilder::new() - .num_threads(max_in_flight.max(1)) - .thread_name(|index| format!("shadow-{index}")) - .build() - .context("failed to start the shadow threads")?; Ok(Self { in_flight: AtomicUsize::new(0), log_window: Mutex::new(LogWindow::new()), settings, max_in_flight, catalogs, - pool + threads: run_threads(max_in_flight)? }) } @@ -112,18 +108,18 @@ impl Shadow { let label = outcome.label(); report_shadow_run(dataset_id, label); - let slow = match outcome { + let slower = match outcome { Outcome::Compared { legacy, new, .. } => { - let slow = is_slow(&settings, *legacy, *new); - report_shadow_timing(dataset_id, *legacy, *new, slow); - slow + let slower = slower(&settings, *legacy, *new); + report_shadow_timing(dataset_id, *legacy, *new, slower); + slower } - Outcome::Failed { .. } => false + Outcome::Failed { .. } => None }; let reason = match label { ShadowOutcome::Mismatch | ShadowOutcome::LegacyError | ShadowOutcome::NewError => label.as_str(), - ShadowOutcome::Match if slow => "slow", + ShadowOutcome::Match if slower == Some(ShadowSlower::New) => "slow", _ => return }; let fingerprint = query_fingerprint(dataset_id, &run.query.body); @@ -146,9 +142,26 @@ impl Shadow { } } -fn is_slow(settings: &ShadowSettings, legacy: Duration, new: Duration) -> bool { - let margin = new.saturating_sub(legacy); - new.as_secs_f64() > legacy.as_secs_f64() * settings.slow_ratio && margin >= settings.slow_margin() +/// One single-thread worker for each run slot. A worker that waits for parallel +/// work in a shared pool runs other jobs of that pool meanwhile, so one run's +/// time would hold the whole of other runs. +fn run_threads(max_in_flight: usize) -> anyhow::Result { + TaskQueue::new(max_in_flight.max(1), "shadow").context("failed to start the shadow threads") +} + +fn slower(settings: &ShadowSettings, legacy: Duration, new: Duration) -> Option { + if is_slower(settings, new, legacy) { + Some(ShadowSlower::New) + } else if is_slower(settings, legacy, new) { + Some(ShadowSlower::Legacy) + } else { + None + } +} + +fn is_slower(settings: &ShadowSettings, time: Duration, other: Duration) -> bool { + let margin = time.saturating_sub(other); + time.as_secs_f64() > other.as_secs_f64() * settings.slow_ratio && margin >= settings.slow_margin() } /// A shadowed query, carried by the response it shadows. @@ -185,7 +198,7 @@ impl ShadowQuery { chunk: chunk.clone(), bounds }; - self.shadow.pool.spawn(move || { + self.shadow.threads.spawn(move || { let outcome = run.execute(); run.query.shadow.record(&run, &outcome); drop(slot); @@ -464,3 +477,52 @@ fn panic_message(panic: &Box) -> &str { .or_else(|| panic.downcast_ref::<&str>().copied()) .unwrap_or("non-string panic payload") } + +#[cfg(test)] +mod tests { + use std::{sync::mpsc, thread}; + + use super::*; + + #[test] + fn a_run_never_runs_another_inside_its_time() { + let threads = run_threads(2).unwrap(); + let (started, run_started) = mpsc::channel(); + let (other_done, other_finished) = mpsc::channel(); + let (done, finished) = mpsc::channel(); + + threads.spawn(move || { + // The second half waits for the other run. Where another worker can take + // it, this one waits for that worker and takes the other run meanwhile. + let ((), other_thread) = rayon::join( + || { + started.send(()).unwrap(); + thread::sleep(Duration::from_millis(100)); + }, + move || other_finished.recv_timeout(Duration::from_secs(10)).unwrap() + ); + done.send((thread::current().id(), other_thread)).unwrap(); + }); + run_started.recv().unwrap(); + threads.spawn(move || other_done.send(thread::current().id()).unwrap()); + + let (thread, other_thread) = finished.recv_timeout(Duration::from_secs(20)).unwrap(); + assert_ne!(thread, other_thread, "the other run ran inside this one"); + } + + #[test] + fn either_engine_can_be_the_slower() { + let settings = ShadowSettings { + percent: 100.0, + slow_ratio: 2.0, + slow_margin_ms: 50, + log_limit: 0 + }; + let ms = Duration::from_millis; + + assert_eq!(slower(&settings, ms(10), ms(100)), Some(ShadowSlower::New)); + assert_eq!(slower(&settings, ms(100), ms(10)), Some(ShadowSlower::Legacy)); + assert_eq!(slower(&settings, ms(10), ms(40)), None, "under the margin"); + assert_eq!(slower(&settings, ms(100), ms(190)), None, "under the ratio"); + } +}