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"); + } +}