fix(hotblocks): give each shadow run its own thread

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,
 

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<Family<DatasetLabel, Counter>> = LazyLock::new(Default::default);
 static SHADOW_RUNS: LazyLock<Family<ShadowRunLabels, Counter>> = LazyLock::new(Default::default);
 static SHADOW_SLOW_RUNS: LazyLock<Family<DatasetLabel, Counter>> = LazyLock::new(Default::default);
+static SHADOW_LEGACY_SLOW_RUNS: LazyLock<Family<DatasetLabel, Counter>> = LazyLock::new(Default::default);
 static SHADOW_RUN_DURATION: LazyLock<Family<ShadowEngineLabels, Histogram>> =
     LazyLock::new(|| Family::new_with_constructor(|| Histogram::new(buckets(0.0001, 24))));
 static SHADOW_PERCENT: LazyLock<Gauge<f64, std::sync::atomic::AtomicU64>> = 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<ShadowSlower>
+) {
     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",

crates/hotblocks/src/query/executor.rs

@@ -47,21 +47,22 @@ type Job = Box<dyn FnOnce() + Send>;
 /// 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<Job>,
     _workers: Vec<rayon::ThreadPool>
 }
 
 impl TaskQueue {
-    fn new(threads: usize) -> anyhow::Result<Self> {
+    /// `threads` workers, named `<name>-<index>`.
+    pub(crate) fn new(threads: usize, name: &'static str) -> anyhow::Result<Self> {
         let (sender, receiver) = mpsc::channel::<Job>();
         let receiver = Arc::new(Mutex::new(receiver));
 
         let workers = (0..threads)
             .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<Self> {
-        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,

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;

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<MetadataCatalogs>,
     /// 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<LogWindow>
 }
 
 impl Shadow {
     pub fn new(max_in_flight: usize, settings: SettingsRef, catalogs: Arc<MetadataCatalogs>) -> anyhow::Result<Self> {
-        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> {
+    TaskQueue::new(max_in_flight.max(1), "shadow").context("failed to start the shadow threads")
+}
+
+fn slower(settings: &ShadowSettings, legacy: Duration, new: Duration) -> Option<ShadowSlower> {
+    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<dyn std::any::Any + Send>) -> &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");
+    }
+}