From 3fea4cd5b2b70df6a0a06bc712d7d21e339cd990 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 12 Aug 2026 11:29:02 -0700 Subject: [PATCH 1/7] [dbsp] Fixup circuit wait time measurement. circuit_wait_time_seconds has read zero since June. It was fed by the scheduler's WaitStart/WaitEnd events, which fire only when the scheduler has no runnable task, and that requires a DBSP-asynchronous operator. Since 50ff76716 turned ExchangeReceiver into a Rust-async operator, a blocked operator is a pending task rather than an idle scheduler, so the events stop happening: only RebalancingExchangeSender and GatherConsumer still declare themselves asynchronous, and neither appears in an ordinary circuit. Fortunately we have an easy way to measure how long the circuit waits by asking the tokio runtime on each step. Also fill in the step profile's CPU time, which was hardcoded to zero, and report it as circuit_cpu_time_seconds plus circuit_nonblocking_percent, mirroring what operators already report. A step's wall time now decomposes: circuit_runtime_seconds - circuit_cpu_time_seconds on cpu - circuit_wait_time_seconds nothing to run = blocked in the kernel or descheduled Signed-off-by: Leonid Ryzhyk --- crates/dbsp/src/circuit/circuit_builder.rs | 32 ++++- crates/dbsp/src/circuit/dbsp_handle.rs | 2 +- crates/dbsp/src/circuit/metadata.rs | 25 +++- crates/dbsp/src/profile.rs | 27 ++-- crates/dbsp/src/profile/cpu.rs | 131 +++++++++++++++--- crates/dbsp/tests/circuit_wait_time.rs | 147 +++++++++++++++++++++ 6 files changed, 326 insertions(+), 38 deletions(-) create mode 100644 crates/dbsp/tests/circuit_wait_time.rs diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs index 81570ec1ddf..17060b6f831 100644 --- a/crates/dbsp/src/circuit/circuit_builder.rs +++ b/crates/dbsp/src/circuit/circuit_builder.rs @@ -22,6 +22,7 @@ //! The API that this directly exposes runs the circuit in the context of the //! current thread. To instead run the circuit in a collection of worker //! threads, use [`Runtime::init_circuit`]. +use crate::profile::RuntimeIdle; use crate::{ Error as DbspError, Position, Runtime, RuntimeError, circuit::{ @@ -3215,13 +3216,23 @@ impl RootCircuit { // TODO: user LocalRuntime instead of Runtime + LocalSet when // tokio::LocalRuntime is stable. // Local tokio runtime that schedules operators on the current worker thread. - let tokio_runtime = tokio::runtime::Builder::new_current_thread() - .build() - .map_err(|e| { + // + // The park hooks measure the time this runtime has nothing to run, which + // is what the circuit's wait time reports. The runtime parks only while + // it is driving a step, so all of it belongs to some step; the profiler + // reads the accumulator at step boundaries to say which. + let runtime_idle = RuntimeIdle::new(); + let tokio_runtime = { + let (park, unpark) = (runtime_idle.clone(), runtime_idle.clone()); + let mut builder = tokio::runtime::Builder::new_current_thread(); + builder.on_thread_park(move || park.park()); + builder.on_thread_unpark(move || unpark.unpark()); + builder.build().map_err(|e| { DbspError::Scheduler(SchedulerError::TokioError { error: e.to_string(), }) - })?; + })? + }; let mut circuit = RootCircuit::new(); // On failure, explicitly deallocate whatever the constructor built: @@ -3284,6 +3295,7 @@ impl RootCircuit { circuit, executor, tokio_runtime, + runtime_idle, replay_info: None, boundary_streams: None, concurrent_bootstrap_info: None, @@ -7617,6 +7629,10 @@ pub struct CircuitHandle { circuit: RootCircuit, executor: Box>, tokio_runtime: TokioRuntime, + + /// Time `tokio_runtime` spent parked, fed by its park hooks and read by the + /// CPU profiler at step boundaries. + runtime_idle: RuntimeIdle, replay_info: Option, /// The boundary streams of the replay prepared by @@ -7733,6 +7749,14 @@ struct ConcurrentBootstrapInfo { } impl CircuitHandle { + /// Time this circuit's runtime has spent with nothing to run. + /// + /// Handed to the CPU profiler, which turns it into the circuit's wait time + /// per step. + pub fn runtime_idle(&self) -> RuntimeIdle { + self.runtime_idle.clone() + } + /// Start and instantly commit a transaction, waiting for the commit to complete. pub fn transaction(&self) -> Result<(), DbspError> { self.tokio_runtime diff --git a/crates/dbsp/src/circuit/dbsp_handle.rs b/crates/dbsp/src/circuit/dbsp_handle.rs index aa762e386cd..6457095e6a8 100644 --- a/crates/dbsp/src/circuit/dbsp_handle.rs +++ b/crates/dbsp/src/circuit/dbsp_handle.rs @@ -824,7 +824,7 @@ impl Runtime { } } Ok(Command::EnableProfiler) => { - profiler.enable_cpu_profiler(); + profiler.enable_cpu_profiler(circuit.runtime_idle()); // Send response. if status_sender.send(Ok(Response::Unit)).is_err() { return; diff --git a/crates/dbsp/src/circuit/metadata.rs b/crates/dbsp/src/circuit/metadata.rs index e48ef450d1f..888330c32ab 100644 --- a/crates/dbsp/src/circuit/metadata.rs +++ b/crates/dbsp/src/circuit/metadata.rs @@ -168,6 +168,13 @@ pub const CIRCUIT_WAIT_TIME_SECONDS: MetricId = MetricId(Cow::Borrowed("circuit_wait_time_seconds")); pub const STEPS_COUNT: MetricId = MetricId(Cow::Borrowed("steps_count")); pub const CIRCUIT_RUNTIME_SECONDS: MetricId = MetricId(Cow::Borrowed("circuit_runtime_seconds")); + +/// CPU time the worker thread used while evaluating steps. +pub const CIRCUIT_CPU_TIME_SECONDS: MetricId = MetricId(Cow::Borrowed("circuit_cpu_time_seconds")); + +/// Fraction of the circuit's step time spent on the CPU. +pub const CIRCUIT_NONBLOCKING_PERCENT: MetricId = + MetricId(Cow::Borrowed("circuit_nonblocking_percent")); pub const CIRCUIT_IDLE_TIME_SECONDS: MetricId = MetricId(Cow::Borrowed("circuit_idle_time_seconds")); pub const CIRCUIT_RUNTIME_ELAPSED_SECONDS: MetricId = @@ -180,7 +187,7 @@ pub const PREFIX_BATCHES_STATS: MetricId = MetricId(Cow::Borrowed("prefix_batche pub const INPUT_INTEGRAL_RECORDS_COUNT: MetricId = MetricId(Cow::Borrowed("input_integral_records_count")); -pub const CIRCUIT_METRICS: [CircuitMetric; 76] = [ +pub const CIRCUIT_METRICS: [CircuitMetric; 78] = [ // State CircuitMetric { name: USED_MEMORY_BYTES, @@ -471,7 +478,7 @@ pub const CIRCUIT_METRICS: [CircuitMetric; 76] = [ name: CIRCUIT_WAIT_TIME_SECONDS, category: CircuitMetricCategory::Time, advanced: false, - description: "Time the circuit scheduler spent waiting for an operator to become ready.", + description: "Time during a step when the worker's async runtime had nothing to run, for example while waiting for other workers at an exchange or for background work to finish. Excludes time an operator blocks its thread without yielding, which shows up instead as a low 'circuit_nonblocking_percent'.", }, CircuitMetric { name: STEPS_COUNT, @@ -483,7 +490,19 @@ pub const CIRCUIT_METRICS: [CircuitMetric; 76] = [ name: CIRCUIT_RUNTIME_SECONDS, category: CircuitMetricCategory::Time, advanced: false, - description: "Total time spent evaluating the circuit, including operators runtime and circuit wait time ('circuit_wait_time_seconds').", + description: "Total time spent evaluating the circuit: operator runtime, circuit wait time ('circuit_wait_time_seconds'), and time blocked in the kernel or descheduled. See also 'circuit_cpu_time_seconds'.", + }, + CircuitMetric { + name: CIRCUIT_CPU_TIME_SECONDS, + category: CircuitMetricCategory::Time, + advanced: false, + description: "CPU time the worker thread used while evaluating steps. Subtracting this and 'circuit_wait_time_seconds' from 'circuit_runtime_seconds' leaves the time the worker was blocked in the kernel or descheduled.", + }, + CircuitMetric { + name: CIRCUIT_NONBLOCKING_PERCENT, + category: CircuitMetricCategory::Time, + advanced: true, + description: "Fraction of the circuit's step time spent on the CPU, as opposed to waiting for other workers, blocking in the kernel, or being descheduled.", }, CircuitMetric { name: CIRCUIT_IDLE_TIME_SECONDS, diff --git a/crates/dbsp/src/profile.rs b/crates/dbsp/src/profile.rs index 7b149675c07..1554ba8eb19 100644 --- a/crates/dbsp/src/profile.rs +++ b/crates/dbsp/src/profile.rs @@ -6,11 +6,12 @@ use crate::{ GlobalNodeId, circuit_builder::{CircuitBase, Node}, metadata::{ - BACKGROUND_CACHE_OCCUPANCY, CIRCUIT_IDLE_TIME_SECONDS, CIRCUIT_METRICS, - CIRCUIT_RUNTIME_ELAPSED_SECONDS, CIRCUIT_RUNTIME_SECONDS, CIRCUIT_WAIT_TIME_SECONDS, - CircuitMetric, FOREGROUND_CACHE_OCCUPANCY, INVOCATIONS_COUNT, MetaItem, MetricId, - MetricReading, OperatorMeta, RUNTIME_NONBLOCKING_PERCENT, RUNTIME_PERCENT, - RUNTIME_SECONDS, SPINE_STORAGE_SIZE_BYTES, STEPS_COUNT, USED_MEMORY_BYTES, + BACKGROUND_CACHE_OCCUPANCY, CIRCUIT_CPU_TIME_SECONDS, CIRCUIT_IDLE_TIME_SECONDS, + CIRCUIT_METRICS, CIRCUIT_NONBLOCKING_PERCENT, CIRCUIT_RUNTIME_ELAPSED_SECONDS, + CIRCUIT_RUNTIME_SECONDS, CIRCUIT_WAIT_TIME_SECONDS, CircuitMetric, + FOREGROUND_CACHE_OCCUPANCY, INVOCATIONS_COUNT, MetaItem, MetricId, MetricReading, + OperatorMeta, RUNTIME_NONBLOCKING_PERCENT, RUNTIME_PERCENT, RUNTIME_SECONDS, + SPINE_STORAGE_SIZE_BYTES, STEPS_COUNT, USED_MEMORY_BYTES, }, }, monitor::{TraceMonitor, visual_graph::Graph}, @@ -30,7 +31,7 @@ use std::{ use zip::{ZipWriter, write::SimpleFileOptions}; mod cpu; -pub use cpu::CPUProfiler; +pub use cpu::{CPUProfiler, RuntimeIdle}; /// Rudimentary circuit profiler. /// @@ -308,8 +309,13 @@ impl Profiler { } /// Enable CPU profiling. - pub fn enable_cpu_profiler(&self) { - self.cpu_profiler.attach(&self.circuit, "cpu_profiler"); + /// + /// `runtime_idle` comes from the [`CircuitHandle`](crate::circuit::CircuitHandle) + /// whose runtime evaluates this circuit; it is the source of the circuit's + /// wait time. + pub fn enable_cpu_profiler(&self, runtime_idle: RuntimeIdle) { + self.cpu_profiler + .attach(&self.circuit, "cpu_profiler", runtime_idle); } pub fn profile(&self, runtime_elapsed: Duration) -> WorkerProfile { @@ -387,6 +393,11 @@ impl Profiler { CIRCUIT_WAIT_TIME_SECONDS => profile.wait_profile.real_time(), STEPS_COUNT => profile.step_profile.invocations(), CIRCUIT_RUNTIME_SECONDS => profile.step_profile.real_time(), + CIRCUIT_CPU_TIME_SECONDS => profile.step_profile.cpu_time(), + CIRCUIT_NONBLOCKING_PERCENT => MetaItem::Percent { + numerator: profile.step_profile.cpu_time().as_micros() as u64, + denominator: profile.step_profile.real_time().as_micros() as u64, + }, CIRCUIT_IDLE_TIME_SECONDS => profile.idle_profile.real_time(), CIRCUIT_RUNTIME_ELAPSED_SECONDS => runtime_elapsed, ]; diff --git a/crates/dbsp/src/profile/cpu.rs b/crates/dbsp/src/profile/cpu.rs index e240fa0222b..564ae391b7e 100644 --- a/crates/dbsp/src/profile/cpu.rs +++ b/crates/dbsp/src/profile/cpu.rs @@ -6,14 +6,82 @@ // - We currently do not measure the time spent in `clock_start`/`clock_end` // events, which can in theory do non-trivial work. -use crate::circuit::{GlobalNodeId, RootCircuit, trace::SchedulerEvent}; +use crate::circuit::{GlobalNodeId, RootCircuit, ThreadCpuTime, trace::SchedulerEvent}; use hashbrown::HashMap; use std::{ cell::RefCell, rc::Rc, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, time::{Duration, Instant}, }; +/// Time a worker's async runtime spent with nothing to run. +/// +/// The runtime that evaluates a circuit is a current-thread runtime owned by one +/// worker, so the time it spends parked is time that worker had no runnable +/// task: it is waiting for its peers at an exchange, for background work, or for +/// an asynchronous operator to become ready. +/// +/// [`CPUProfiler`] samples this at step boundaries, which is what attributes the +/// idle time to a step and leaves out parks that happen outside one. +/// +/// Timestamps are nanoseconds measured against [`RuntimeIdle::base`] rather than +/// wall-clock time, so this shares a monotonic clock with the step's own +/// duration and the two can be subtracted. +#[derive(Clone, Debug)] +pub struct RuntimeIdle { + base: Instant, + /// When the current park started, or 0 when the runtime is not parked. + park_start: Arc, + /// Total time parked. + total: Arc, +} + +impl Default for RuntimeIdle { + fn default() -> Self { + Self::new() + } +} + +impl RuntimeIdle { + pub fn new() -> Self { + Self { + base: Instant::now(), + park_start: Arc::new(AtomicU64::new(0)), + total: Arc::new(AtomicU64::new(0)), + } + } + + fn now(&self) -> u64 { + self.base.elapsed().as_nanos() as u64 + } + + /// Called by the runtime's `on_thread_park` hook. + pub fn park(&self) { + self.park_start.store(self.now(), Ordering::Release); + } + + /// Called by the runtime's `on_thread_unpark` hook. + /// + /// An unpark without a preceding park adds nothing, which is what happens + /// for the unpark the runtime performs as it starts running. + pub fn unpark(&self) { + let start = self.park_start.swap(0, Ordering::AcqRel); + if start != 0 { + self.total + .fetch_add(self.now().saturating_sub(start), Ordering::Release); + } + } + + /// Total time parked so far. + pub fn total(&self) -> Duration { + Duration::from_nanos(self.total.load(Ordering::Acquire)) + } +} + /// Per-operator CPU profile. #[derive(Clone, Default, Debug)] pub struct OperatorCPUProfile { @@ -69,10 +137,16 @@ pub struct CircuitCPUProfile { #[derive(Default, Debug)] struct CPUProfilerInner { operators: HashMap, - wait_start_times: HashMap, step_start_times: HashMap, step_end_times: HashMap, + /// Thread CPU time when the current step started, per circuit. + step_start_cpu: HashMap, + /// Runtime idle total when the current step started, per circuit. + step_start_idle: HashMap, circuit_profiles: HashMap, + /// Set when the profiler is attached; `None` leaves the wait and CPU + /// figures at zero rather than reporting a number with nothing behind it. + runtime_idle: Option, } impl CPUProfilerInner { @@ -92,17 +166,40 @@ impl CPUProfilerInner { self.step_start_times .insert((*circuit_id).clone(), Instant::now()); + self.step_start_cpu + .insert((*circuit_id).clone(), ThreadCpuTime::now().0); + if let Some(idle) = &self.runtime_idle { + self.step_start_idle + .insert((*circuit_id).clone(), idle.total()); + } } SchedulerEvent::StepEnd { circuit_id } => { if let Some(start_time) = self.step_start_times.remove(*circuit_id) { let duration = Instant::now().duration_since(start_time); + let cpu = self + .step_start_cpu + .remove(*circuit_id) + .map(|start| ThreadCpuTime::now().0.saturating_sub(start)) + .unwrap_or_default(); let circuit_profile = self .circuit_profiles .entry((*circuit_id).clone()) .or_insert_with(Default::default); - circuit_profile - .step_profile - .add_event(duration, Duration::ZERO); + circuit_profile.step_profile.add_event(duration, cpu); + + // Time the runtime spent parked during this step. Measured + // here rather than from the scheduler's `WaitStart`/`WaitEnd` + // events: since operators became Rust-async, a blocked + // operator is a pending task rather than an idle scheduler, + // so those events no longer fire for it. + if let (Some(idle), Some(before)) = ( + self.runtime_idle.as_ref(), + self.step_start_idle.remove(*circuit_id), + ) { + circuit_profile + .wait_profile + .add_event(idle.total().saturating_sub(before), Duration::ZERO); + } }; self.step_end_times .insert((*circuit_id).clone(), Instant::now()); @@ -117,22 +214,9 @@ impl CPUProfilerInner { // println!("{}:{}:{:?}", crate::Runtime::worker_index(), // node.global_id(), duration); } - SchedulerEvent::WaitStart { circuit_id } => { - self.wait_start_times - .insert((*circuit_id).clone(), Instant::now()); - } - SchedulerEvent::WaitEnd { circuit_id } => { - if let Some(start_time) = self.wait_start_times.remove(*circuit_id) { - let duration = Instant::now().duration_since(start_time); - let circuit_profile = self - .circuit_profiles - .entry((*circuit_id).clone()) - .or_insert_with(Default::default); - circuit_profile - .wait_profile - .add_event(duration, Duration::ZERO); - }; - } + // `WaitStart`/`WaitEnd` are deliberately ignored: they report only + // the scheduler having no runnable task, which the runtime's park + // time already covers, and counting both would double count. _ => (), } } @@ -152,7 +236,10 @@ impl CPUProfiler { /// Attach CPU profiler to a circuit. The profiler will start measuring /// circuit's CPU usage. - pub fn attach(&self, circuit: &RootCircuit, handler_name: &str) { + pub fn attach(&self, circuit: &RootCircuit, handler_name: &str, runtime_idle: RuntimeIdle) { + if let Ok(mut this) = self.0.try_borrow_mut() { + this.runtime_idle = Some(runtime_idle); + } let self_clone = self.clone(); circuit.register_scheduler_event_handler(handler_name, move |event| { diff --git a/crates/dbsp/tests/circuit_wait_time.rs b/crates/dbsp/tests/circuit_wait_time.rs new file mode 100644 index 00000000000..ec9447ad845 --- /dev/null +++ b/crates/dbsp/tests/circuit_wait_time.rs @@ -0,0 +1,147 @@ +//! The circuit's wait time must reflect a worker's async runtime having nothing +//! to run. +//! +//! This was silently zero for two months: the scheduler's own wait accounting +//! only fires for DBSP-asynchronous operators, and once operators became +//! Rust-async, a blocked operator became a pending task rather than an idle +//! scheduler. The tests below pin the observable behaviour, both directions: a +//! skewed multi-worker circuit must report waiting, and a circuit with nothing +//! to wait for must not. + +use dbsp::circuit::Circuit; +use dbsp::circuit::metadata::{ + CIRCUIT_CPU_TIME_SECONDS, CIRCUIT_NONBLOCKING_PERCENT, CIRCUIT_RUNTIME_SECONDS, + CIRCUIT_WAIT_TIME_SECONDS, MetaItem, MetricId, +}; +use dbsp::typed_batch::OrdZSet; +use dbsp::utils::Tup2; +use dbsp::{DBSPHandle, Runtime, operator::Generator}; +use std::thread::sleep; +use std::time::Duration; + +/// Sums a duration metric over all workers, in seconds. +fn duration_metric(handle: &mut DBSPHandle, metric: &MetricId) -> f64 { + let profile = handle.retrieve_profile().unwrap(); + profile + .worker_profiles + .iter() + .flat_map(|w| w.attribute_profile(metric).into_values()) + .map(|item| match item { + MetaItem::Duration(d) => d.as_secs_f64(), + other => panic!("expected a duration, got {other:?}"), + }) + .sum() +} + +/// The worst worker's percentage for `metric`, as (numerator, denominator). +fn percent_metric(handle: &mut DBSPHandle, metric: &MetricId) -> Vec<(u64, u64)> { + let profile = handle.retrieve_profile().unwrap(); + profile + .worker_profiles + .iter() + .flat_map(|w| w.attribute_profile(metric).into_values()) + .map(|item| match item { + MetaItem::Percent { + numerator, + denominator, + } => (numerator, denominator), + other => panic!("expected a percent, got {other:?}"), + }) + .collect() +} + +/// Builds a circuit whose source optionally stalls worker 0, so its peers have +/// to wait for it at the exchange that `shard()` inserts. +fn skewed_circuit(workers: usize, stall: Duration, steps: usize) -> DBSPHandle { + let (mut handle, _) = Runtime::init_circuit(workers, move |circuit| { + let source = circuit.add_source(Generator::new(move || { + if Runtime::worker_index() == 0 && !stall.is_zero() { + sleep(stall); + } + let keys: Vec> = (0..64) + .map(|k| Tup2(k * 7 + Runtime::worker_index() as u64, 1i64)) + .collect(); + OrdZSet::from_keys((), keys) + })); + source.shard().integrate().apply(|_| ()); + Ok(()) + }) + .unwrap(); + + handle.enable_cpu_profiler().unwrap(); + handle.start_transaction().unwrap(); + for _ in 0..steps { + handle.step().unwrap(); + } + handle.commit_transaction().unwrap(); + handle +} + +/// A worker stalled every step makes its peers wait, and that wait is reported. +#[test] +fn stalled_worker_is_reported_as_circuit_wait_time() { + const WORKERS: usize = 4; + const STEPS: usize = 20; + const STALL: Duration = Duration::from_millis(5); + + let mut handle = skewed_circuit(WORKERS, STALL, STEPS); + let wait = duration_metric(&mut handle, &CIRCUIT_WAIT_TIME_SECONDS); + + // Three of four workers wait out most of each stall. Half of the total + // stall time is a deliberately loose floor: the point is that the metric + // tracks the stall rather than reading zero. + let stalled = STALL.as_secs_f64() * STEPS as f64; + let floor = stalled * (WORKERS - 1) as f64 * 0.5; + assert!( + wait > floor, + "circuit wait time {wait:.3}s should exceed {floor:.3}s with a \ + {STALL:?} stall on 1 of {WORKERS} workers over {STEPS} steps" + ); + handle.kill().unwrap(); +} + +/// Nothing to wait for means no wait time: a single worker whose operators never +/// block must not accumulate any. Without this, a hook that counted parks +/// outside a step, or counted the runtime's own startup, would pass the test +/// above while reporting nonsense here. +#[test] +fn a_circuit_with_nothing_to_wait_for_reports_no_wait_time() { + let mut handle = skewed_circuit(1, Duration::ZERO, 20); + let wait = duration_metric(&mut handle, &CIRCUIT_WAIT_TIME_SECONDS); + let runtime = duration_metric(&mut handle, &CIRCUIT_RUNTIME_SECONDS); + assert!( + wait < runtime * 0.5, + "a single worker with no exchange waited {wait:.3}s of {runtime:.3}s" + ); + handle.kill().unwrap(); +} + +/// The step budget has to close: CPU time and wait time are both parts of the +/// step's wall time, so neither may exceed it, and a busy circuit must show CPU +/// time rather than leaving it at the zero it used to report. +#[test] +fn step_time_decomposes_into_cpu_and_wait() { + let mut handle = skewed_circuit(4, Duration::from_millis(2), 20); + + let runtime = duration_metric(&mut handle, &CIRCUIT_RUNTIME_SECONDS); + let cpu = duration_metric(&mut handle, &CIRCUIT_CPU_TIME_SECONDS); + let wait = duration_metric(&mut handle, &CIRCUIT_WAIT_TIME_SECONDS); + + assert!(cpu > 0.0, "circuit cpu time should not be zero"); + // A 10% tolerance covers the two clocks: the step is timed with `Instant`, + // the CPU with `CLOCK_THREAD_CPUTIME_ID`, and the parks with a third set of + // reads against a monotonic base. + assert!( + cpu + wait <= runtime * 1.1, + "cpu {cpu:.3}s + wait {wait:.3}s should fit within runtime {runtime:.3}s" + ); + + for (numerator, denominator) in percent_metric(&mut handle, &CIRCUIT_NONBLOCKING_PERCENT) { + assert!(denominator > 0, "nonblocking percent has no denominator"); + assert!( + numerator <= denominator, + "nonblocking percent {numerator}/{denominator} exceeds 100%" + ); + } + handle.kill().unwrap(); +} From f909fdd1f0ca37724053f1fe5b838c74875c2ddf Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Thu, 13 Aug 2026 10:17:54 -0700 Subject: [PATCH 2/7] [dbsp] report the spine's batch count in each merge marker A merge marker said what the merge consumed and produced but not what it left behind? The marker now ends with "spine now holds N loose of M batches". The marker moves after the merged batch is added to the spine, so the counts describe the state the merge leaves rather than the state mid-way through it. Signed-off-by: Leonid Ryzhyk --- crates/dbsp/src/trace/spine_async.rs | 36 ++++++++++++++++------------ 1 file changed, 21 insertions(+), 15 deletions(-) diff --git a/crates/dbsp/src/trace/spine_async.rs b/crates/dbsp/src/trace/spine_async.rs index ddb5406b76b..4200daedafb 100644 --- a/crates/dbsp/src/trace/spine_async.rs +++ b/crates/dbsp/src/trace/spine_async.rs @@ -487,21 +487,8 @@ where let post_len = new_batch.len(); self.spine_stats .report_merge(pre_len, post_len, cache_stats); - Span::new(LEVEL_NAMES[level]) - .with_category("Spine") - .with_start(start) - .with_tooltip(|| { - format!( - "{} in worker {} merged {} batches ({pre_len} -> {post_len}) in {} steps using {:.1} ms real time and {:.1} ms CPU time", - &self.name, - Runtime::worker_index(), - batches.len(), - n_steps, - elapsed.real.as_secs_f64() * 1000.0, - elapsed.cpu.as_secs_f64() * 1000.0 - ) - }) - .record(); + let n_merged_batches = batches.len(); + let merge_name = self.name.clone(); if slot.compaction_status == CompactionStatus::InProgress { // We finished merging all batches in the slot as part of compaction. @@ -526,6 +513,25 @@ where } else if !new_batch.is_empty() { self.add_batch(new_batch, new_level); } + + // Recorded here, rather than where the merge is accounted above, so the + // counts describe the spine as the merge leaves it. `loose` is what + // backpressure measures, and it is what grows when merging cannot keep + // up with the batches each step adds. + let loose = self.batch_count().0; + let total: usize = self.slots.iter().map(Slot::n_batches).sum(); + Span::new(LEVEL_NAMES[level]) + .with_category("Spine") + .with_start(start) + .with_tooltip(|| { + format!( + "{merge_name} in worker {} merged {n_merged_batches} batches ({pre_len} -> {post_len}) in {n_steps} steps using {:.1} ms real time and {:.1} ms CPU time; spine now holds {loose} loose of {total} batches", + Runtime::worker_index(), + elapsed.real.as_secs_f64() * 1000.0, + elapsed.cpu.as_secs_f64() * 1000.0 + ) + }) + .record(); } /// Returns a copy of the data that the caller can use to construct a From 0b12be9e9f8619193304814ff895b209465e7803 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Thu, 13 Aug 2026 10:18:50 -0700 Subject: [PATCH 3/7] [dbsp] account for merge backpressure waits taken through the waiter merge_backpressure_wait_time_seconds read zero on a workload where workers demonstrably waited for merging, and no backpressure-wait marker appeared in its profile. There are two ways to wait for the spine to drain and only one of them was instrumented. `backpressure_wait` polls in a loop, adds the elapsed time to the spine's stats, and records a marker. `backpressure_waiter` just hands out a `Notified` to await, and recorded nothing; the sharded accumulator uses that one (ShardedAccumulatorLocalWaiter), so its waits reached neither the metric nor the profile. They were visible only as unexplained time inside the operator. The waiter now carries what a report needs, its caller reports the wait once awaited, and both paths funnel into a single `record_backpressure_wait`, so a wait counts the same however it was taken. The marker reports the batch counts either side of the wait, loose and total, which says whether waiting achieved anything. Splitting the waiter into a future and a report keeps the `Notified` out of the value the caller holds across the await. Signed-off-by: Leonid Ryzhyk --- .../operator/dynamic/sharded_accumulator.rs | 10 +- crates/dbsp/src/trace/spine_async.rs | 140 ++++++++++++++---- 2 files changed, 120 insertions(+), 30 deletions(-) diff --git a/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs b/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs index 3b065cffdfc..f0d1006f873 100644 --- a/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs +++ b/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs @@ -390,9 +390,15 @@ where .spines .front() .and_then(|entry| entry.spine.backpressure_waiter()); - if let Some(waiter) = waiter { + if let Some((notified, report)) = waiter { local_waiters.push(worker); - waiter.await; + notified.await; + // Report it, so the wait reaches + // `merge_backpressure_wait_time_seconds` and the profile rather + // than only showing up as time spent in this operator. + if let Some(entry) = rxq.lock().unwrap().spines.front() { + entry.spine.record_backpressure_wait(report); + } } } if !local_waiters.is_empty() { diff --git a/crates/dbsp/src/trace/spine_async.rs b/crates/dbsp/src/trace/spine_async.rs index 4200daedafb..867cbed600f 100644 --- a/crates/dbsp/src/trace/spine_async.rs +++ b/crates/dbsp/src/trace/spine_async.rs @@ -515,9 +515,9 @@ where } // Recorded here, rather than where the merge is accounted above, so the - // counts describe the spine as the merge leaves it. `loose` is what - // backpressure measures, and it is what grows when merging cannot keep - // up with the batches each step adds. + // batch counts describe the spine as the merge leaves it: `loose` is + // what backpressure measures, and it is what grows when merging cannot + // keep up with the batches each step adds. let loose = self.batch_count().0; let total: usize = self.slots.iter().map(Slot::n_batches).sum(); Span::new(LEVEL_NAMES[level]) @@ -587,6 +587,44 @@ where } } +/// Handed out by `backpressure_waiter` when the spine holds too many batches. +/// +/// Awaiting `notify` blocks until merging catches up; `report` then describes +/// the wait for the metric and the profile. +struct BackpressureWait { + notify: OwnedNotified, + name: Arc, + start: Instant, + initial_loose: usize, + initial_total: usize, +} + +impl BackpressureWait { + /// Splits into the future to await and the description to report afterwards. + fn split(self) -> (OwnedNotified, BackpressureWaitReport) { + ( + self.notify, + BackpressureWaitReport { + name: self.name, + start: self.start, + initial_loose: self.initial_loose, + initial_total: self.initial_total, + }, + ) + } +} + +/// What [`BackpressureWait`] leaves behind once its future has been awaited. +/// +/// Public because the sharded accumulator awaits the future and reports the +/// wait; see [`Spine::backpressure_waiter`]. +pub struct BackpressureWaitReport { + name: Arc, + start: Instant, + initial_loose: usize, + initial_total: usize, +} + #[derive(Copy, Clone, Debug)] struct BatchCount(usize); @@ -758,7 +796,7 @@ where // Do an initial check of the batch count. If it's already dropped, // exit early, otherwise save the name and the initial number of // batches for later recording. - let (name, initial_batches); + let (name, initial_batches, initial_total); { let state = self.state.lock().unwrap(); let batch_count = state.batch_count(); @@ -767,44 +805,81 @@ where } name = state.name.clone(); initial_batches = batch_count.0; + initial_total = state.slots.iter().map(Slot::n_batches).sum(); } // Wait for the batch count to drop below the threshold. - let final_batches = loop { + loop { let notify = self.no_backpressure.notified(); { - let mut state = self.state.lock().unwrap(); - let batch_count = state.batch_count(); - if batch_count.should_relieve_backpressure() { - state.spine_stats.backpressure_wait += start_time.elapsed(); - break batch_count.0; + let state = self.state.lock().unwrap(); + if state.batch_count().should_relieve_backpressure() { + break; } } notify.await; - }; + } - // Record the wait. - COMPACTION_STALL_TIME_NANOSECONDS - .fetch_add(start_time.elapsed().as_nanos() as u64, Ordering::Relaxed); + self.record_backpressure_wait(BackpressureWaitReport { + name, + start: start_time, + initial_loose: initial_batches, + initial_total: initial_total, + }); + } + + /// Returns something to await when the spine holds too many batches, or + /// `None` when it does not. + /// + /// Callers that await the result should report the wait through + /// [`Self::record_backpressure_wait`], so that it reaches + /// `merge_backpressure_wait_time_seconds` and the profile. Awaiting this + /// without reporting is how a wait becomes invisible. + fn backpressure_waiter(&self) -> Option { + let notify = self.no_backpressure.clone().notified_owned(); + let state = self.state.lock().unwrap(); + let batch_count = state.batch_count(); + batch_count.should_apply_backpressure().then(|| { + let total: usize = state.slots.iter().map(Slot::n_batches).sum(); + BackpressureWait { + notify, + name: state.name.clone(), + start: Instant::now(), + initial_loose: batch_count.0, + initial_total: total, + } + }) + } + + /// Accounts for a wait that [`Self::backpressure_waiter`] handed out and the + /// caller has finished awaiting. + fn record_backpressure_wait(&self, wait: BackpressureWaitReport) { + let elapsed = wait.start.elapsed(); + let (final_loose, final_total) = { + let mut state = self.state.lock().unwrap(); + state.spine_stats.backpressure_wait += elapsed; + ( + state.batch_count().0, + state.slots.iter().map(Slot::n_batches).sum::(), + ) + }; + COMPACTION_STALL_TIME_NANOSECONDS.fetch_add(elapsed.as_nanos() as u64, Ordering::Relaxed); Span::new("backpressure-wait") .with_category("Spine") - .with_start(start_time) + .with_start(wait.start) .with_tooltip(|| { - format!("{name} wait for drop from {initial_batches} to {final_batches} batches") + format!( + "{} in worker {} waited {:.1} ms for merges: loose batches {} -> {final_loose}, total {} -> {final_total}", + wait.name, + Runtime::worker_index(), + elapsed.as_secs_f64() * 1000.0, + wait.initial_loose, + wait.initial_total, + ) }) .record(); } - fn backpressure_waiter(&self) -> Option { - let notify = self.no_backpressure.clone().notified_owned(); - self.state - .lock() - .unwrap() - .batch_count() - .should_apply_backpressure() - .then_some(notify) - } - /// Adds `batches` to the shared merging state and wakes up the merger. fn add_batches(&self, batches: impl IntoIterator, usize)>) { self.state.lock().unwrap().add_batches(batches); @@ -2353,8 +2428,17 @@ where /// Returns an object that can be used to wait for backpressure to be /// relieved. Returns `None` if no backpressure is needed. - pub fn backpressure_waiter(&self) -> Option { - self.merger.backpressure_waiter() + /// + /// Pass the report half to [`Self::record_backpressure_wait`] after + /// awaiting, so the wait shows up in + /// `merge_backpressure_wait_time_seconds` and in profiles. + pub fn backpressure_waiter(&self) -> Option<(OwnedNotified, BackpressureWaitReport)> { + self.merger.backpressure_waiter().map(|wait| wait.split()) + } + + /// Reports a wait obtained from [`Self::backpressure_waiter`]. + pub fn record_backpressure_wait(&self, report: BackpressureWaitReport) { + self.merger.record_backpressure_wait(report) } } From 38d34fd0ceb880a8adf09fde5c57600af203e196 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Thu, 13 Aug 2026 14:19:18 -0700 Subject: [PATCH 4/7] [dbsp] give each receive queue's spine to the worker that drains it Every worker shares one `ShardedAccumulator`, built by whichever of them reaches the constructor first, and that worker's index went to all of the receive queues. Their compaction then ran on the constructing worker's background buffer cache and slab allocator, which a spine's worker index picks, rather than spreading over its peers', and a profile reported all of the operator's merges under that one worker. The wrong worker also made concurrent merges of distinct queues look like repeated merges of a single spine in samply markers: every queue names its spines after the same operator, so the owner is the only field that tells them apart. Construction now takes no worker index at all, so the mistake cannot recur. A spine carries its owner, so the backpressure-wait marker can name it alongside the blocked worker that records the wait. Signed-off-by: Leonid Ryzhyk --- .../operator/dynamic/sharded_accumulator.rs | 14 +----- crates/dbsp/src/trace/spine_async.rs | 44 +++++++++++++------ 2 files changed, 32 insertions(+), 26 deletions(-) diff --git a/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs b/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs index f0d1006f873..ebfaf537fe4 100644 --- a/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs +++ b/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs @@ -101,7 +101,6 @@ where let exchange_id: ExchangeId = runtime.sequence_next().try_into().unwrap(); let exchange = ShardedAccumulator::::with_runtime( &runtime, - Runtime::worker_index(), workers.clone(), exchange_id, factories, @@ -177,7 +176,6 @@ where { fn with_runtime( runtime: &Runtime, - worker_index: usize, workers: Range, exchange_id: ExchangeId, factories: &B::Factories, @@ -194,7 +192,6 @@ where .or_insert_with(|| { ShardedAccumulator::new( runtime, - worker_index, workers, clients, exchange_id, @@ -210,7 +207,6 @@ where /// given `workers` (which is often all the workers in the runtime). fn new( runtime: &Runtime, - worker_index: usize, workers: Range, clients: Arc, exchange_id: ExchangeId, @@ -229,14 +225,8 @@ where clients, rxq: layout .local_workers() - .map(|_| { - Mutex::new(Rxq::new( - runtime, - worker_index, - factories, - npeers, - name.get(), - )) + .map(|receiver| { + Mutex::new(Rxq::new(runtime, receiver, factories, npeers, name.get())) }) .collect(), name, diff --git a/crates/dbsp/src/trace/spine_async.rs b/crates/dbsp/src/trace/spine_async.rs index 867cbed600f..80dd32a4857 100644 --- a/crates/dbsp/src/trace/spine_async.rs +++ b/crates/dbsp/src/trace/spine_async.rs @@ -351,6 +351,12 @@ where #[size_of(skip)] factories: B::Factories, name: Arc, + + /// Worker that owns this spine, fixed when it is constructed. + /// + /// It selects the background buffer cache and slab allocator that this + /// spine's merges use, and it is the worker they are reported under. + owner_worker: usize, #[size_of(skip)] key_filter: Option>, #[size_of(skip)] @@ -368,10 +374,11 @@ impl SharedState where B: Batch, { - pub fn new(factories: &B::Factories, name: Arc) -> Self { + pub fn new(factories: &B::Factories, name: Arc, owner_worker: usize) -> Self { Self { factories: factories.clone(), name, + owner_worker, key_filter: None, value_filter: None, frontier: B::Time::minimum(), @@ -489,6 +496,7 @@ where .report_merge(pre_len, post_len, cache_stats); let n_merged_batches = batches.len(); let merge_name = self.name.clone(); + let owner = self.owner_worker; if slot.compaction_status == CompactionStatus::InProgress { // We finished merging all batches in the slot as part of compaction. @@ -524,11 +532,16 @@ where .with_category("Spine") .with_start(start) .with_tooltip(|| { + // The merge runs on a pooled merger thread whose + // `Runtime::worker_index()` is set from this same owner, so the + // owner is the only worker there is to report here. format!( - "{merge_name} in worker {} merged {n_merged_batches} batches ({pre_len} -> {post_len}) in {n_steps} steps using {:.1} ms real time and {:.1} ms CPU time; spine now holds {loose} loose of {total} batches", - Runtime::worker_index(), - elapsed.real.as_secs_f64() * 1000.0, - elapsed.cpu.as_secs_f64() * 1000.0 + "{merge_name} on behalf of worker {owner} merged {n_merged_batches} batches \ + ({pre_len} -> {post_len} rows) in {n_steps} steps using {real_ms:.1} ms real \ + time and {cpu_ms:.1} ms CPU time; spine now holds {loose} loose of {total} \ + batches", + real_ms = elapsed.real.as_secs_f64() * 1000.0, + cpu_ms = elapsed.cpu.as_secs_f64() * 1000.0, ) }) .record(); @@ -734,7 +747,7 @@ where ) -> Self { let idle = Arc::new(Condvar::new()); let no_backpressure = Arc::new(Notify::new()); - let state = Arc::new(Mutex::new(SharedState::new(factories, name))); + let state = Arc::new(Mutex::new(SharedState::new(factories, name, worker_index))); let max_level0_batch_size_records = max_level0_batch_size_records(); assert!( @@ -824,7 +837,7 @@ where name, start: start_time, initial_loose: initial_batches, - initial_total: initial_total, + initial_total, }); } @@ -855,12 +868,13 @@ where /// caller has finished awaiting. fn record_backpressure_wait(&self, wait: BackpressureWaitReport) { let elapsed = wait.start.elapsed(); - let (final_loose, final_total) = { + let (final_loose, final_total, owner) = { let mut state = self.state.lock().unwrap(); state.spine_stats.backpressure_wait += elapsed; ( state.batch_count().0, state.slots.iter().map(Slot::n_batches).sum::(), + state.owner_worker, ) }; COMPACTION_STALL_TIME_NANOSECONDS.fetch_add(elapsed.as_nanos() as u64, Ordering::Relaxed); @@ -869,12 +883,14 @@ where .with_start(wait.start) .with_tooltip(|| { format!( - "{} in worker {} waited {:.1} ms for merges: loose batches {} -> {final_loose}, total {} -> {final_total}", - wait.name, - Runtime::worker_index(), - elapsed.as_secs_f64() * 1000.0, - wait.initial_loose, - wait.initial_total, + "{name} owned by worker {owner} waited {wait_ms:.1} ms for merges on behalf of \ + worker {merging_worker}: loose batches {initial_loose} -> {final_loose}, \ + total {initial_total} -> {final_total}", + name = wait.name, + wait_ms = elapsed.as_secs_f64() * 1000.0, + merging_worker = Runtime::worker_index(), + initial_loose = wait.initial_loose, + initial_total = wait.initial_total, ) }) .record(); From fabeba978cc724afc4b6c75abad2c41fa0e1d0fe Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Thu, 13 Aug 2026 15:19:07 -0700 Subject: [PATCH 5/7] [dbsp] keep the backpressure wait report out of the public API Its documentation links to the private `BackpressureWait` it is split from, which rustdoc reports because that link resolves only while private items are being documented and breaks in the public docs. The report, and the two `Spine` methods that hand it out and take it back, are used only by the sharded accumulator inside this crate, so `pub(crate)` describes them accurately and the link resolves. Also stop pointing at `RuntimeIdle`'s private `base` field, which rustdoc flags for the same reason. --- crates/dbsp/src/profile/cpu.rs | 4 ++-- crates/dbsp/src/trace/spine_async.rs | 12 ++++++------ 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/crates/dbsp/src/profile/cpu.rs b/crates/dbsp/src/profile/cpu.rs index 564ae391b7e..8b27ced271b 100644 --- a/crates/dbsp/src/profile/cpu.rs +++ b/crates/dbsp/src/profile/cpu.rs @@ -28,8 +28,8 @@ use std::{ /// [`CPUProfiler`] samples this at step boundaries, which is what attributes the /// idle time to a step and leaves out parks that happen outside one. /// -/// Timestamps are nanoseconds measured against [`RuntimeIdle::base`] rather than -/// wall-clock time, so this shares a monotonic clock with the step's own +/// Timestamps are nanoseconds measured against a fixed `base` instant rather +/// than wall-clock time, so this shares a monotonic clock with the step's own /// duration and the two can be subtracted. #[derive(Clone, Debug)] pub struct RuntimeIdle { diff --git a/crates/dbsp/src/trace/spine_async.rs b/crates/dbsp/src/trace/spine_async.rs index 80dd32a4857..a49321dd4fc 100644 --- a/crates/dbsp/src/trace/spine_async.rs +++ b/crates/dbsp/src/trace/spine_async.rs @@ -627,11 +627,11 @@ impl BackpressureWait { } } -/// What [`BackpressureWait`] leaves behind once its future has been awaited. +/// What a [`BackpressureWait`] leaves behind once its future has been awaited. /// -/// Public because the sharded accumulator awaits the future and reports the -/// wait; see [`Spine::backpressure_waiter`]. -pub struct BackpressureWaitReport { +/// Reaches beyond this module because the sharded accumulator awaits the future +/// and reports the wait; see [`Spine::backpressure_waiter`]. +pub(crate) struct BackpressureWaitReport { name: Arc, start: Instant, initial_loose: usize, @@ -2448,12 +2448,12 @@ where /// Pass the report half to [`Self::record_backpressure_wait`] after /// awaiting, so the wait shows up in /// `merge_backpressure_wait_time_seconds` and in profiles. - pub fn backpressure_waiter(&self) -> Option<(OwnedNotified, BackpressureWaitReport)> { + pub(crate) fn backpressure_waiter(&self) -> Option<(OwnedNotified, BackpressureWaitReport)> { self.merger.backpressure_waiter().map(|wait| wait.split()) } /// Reports a wait obtained from [`Self::backpressure_waiter`]. - pub fn record_backpressure_wait(&self, report: BackpressureWaitReport) { + pub(crate) fn record_backpressure_wait(&self, report: BackpressureWaitReport) { self.merger.record_backpressure_wait(report) } } From 6d06891899e07f716acf301e6d524f435141de98 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Thu, 13 Aug 2026 15:19:17 -0700 Subject: [PATCH 6/7] [dbsp] fix stale and ambiguous rustdoc links `cargo doc -p dbsp` reported eleven warnings that predate this branch, and the crate's documentation now builds without any. Four links named items that no longer exist or never did: `StaticScheduler` is gone, `group_transform` is now `dyn_group_transform`, `size_from_level` is now `size_to_level`, and `eof` is `BulkRows::at_eof` rather than a method of the index level whose documentation cited it. Two more spelled out a target that the link's own label already resolved to, and one wrote `std::mem::transmute` with a single colon. The remaining four are prose indexing an array, `weight_times[current_index]` and `MERGE_COUNTS[level]`, which rustdoc reads as a link. Backticks make them code, which is what they are. --- crates/dbsp/src/circuit/circuit_builder.rs | 6 +++--- crates/dbsp/src/circuit/schedule/dynamic_scheduler.rs | 5 ++--- crates/dbsp/src/operator/dynamic/group.rs | 2 +- crates/dbsp/src/operator/dynamic/multijoin/star_join.rs | 6 +++--- crates/dbsp/src/storage/file/reader/bulk_rows.rs | 2 +- crates/dbsp/src/trace/spine_async.rs | 5 +++-- 6 files changed, 13 insertions(+), 13 deletions(-) diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs index 17060b6f831..1daf840269b 100644 --- a/crates/dbsp/src/circuit/circuit_builder.rs +++ b/crates/dbsp/src/circuit/circuit_builder.rs @@ -776,7 +776,7 @@ where { /// Transmute a stream of `D` into a stream of `D2`. /// - /// This is unsafe and dangerous for the same reasons [`std::mem:transmute`] + /// This is unsafe and dangerous for the same reasons [`std::mem::transmute`] /// is dangerous and should be used with care. /// /// # Safety @@ -1592,14 +1592,14 @@ circuit_cache_key!(ReplaySource(StreamId => Box)); /// Register `replay_stream` as a replay source for `stream`. /// -/// Also attaches a [`Recorder`](crate::operator::dynamic::recorder::Recorder) +/// Also attaches a [`Recorder`] /// to `stream`: every stream with a replay source can turn out to be a /// boundary stream of the bootstrapped region during concurrent /// bootstrapping, in which case the deltas applied to it while the bootstrap /// circuit replays the (frozen) integral must be recorded for the /// synchronization transaction. The recorder is disabled (empty, no-op) /// until bootstrap orchestration enables it through the -/// [`RecorderId`](crate::operator::dynamic::recorder::RecorderId) cache +/// [`RecorderId`] cache /// entry. #[track_caller] pub(crate) fn register_replay_stream( diff --git a/crates/dbsp/src/circuit/schedule/dynamic_scheduler.rs b/crates/dbsp/src/circuit/schedule/dynamic_scheduler.rs index 8a0010ca5e9..64f713c25bd 100644 --- a/crates/dbsp/src/circuit/schedule/dynamic_scheduler.rs +++ b/crates/dbsp/src/circuit/schedule/dynamic_scheduler.rs @@ -8,9 +8,8 @@ //! 3. Pick a highest priority node among nodes that satisfy the first two //! conditions. //! -//! Unlike [`StaticScheduler`](`crate::circuit::schedule::StaticScheduler`), -//! the dynamic scheduler blocks when there are no runnable nodes instead of -//! busy waiting. +//! The scheduler blocks when there are no runnable nodes instead of busy +//! waiting. //! //! # Design //! diff --git a/crates/dbsp/src/operator/dynamic/group.rs b/crates/dbsp/src/operator/dynamic/group.rs index cff95f3b63c..c0e67cc6114 100644 --- a/crates/dbsp/src/operator/dynamic/group.rs +++ b/crates/dbsp/src/operator/dynamic/group.rs @@ -311,7 +311,7 @@ where ) } - /// Like [`group_transform`](`Self::group_transform`), but can output any + /// Like [`Self::dyn_group_transform`], but can output any /// indexed Z-set, not just [`OrdIndexedZSet`] fn dyn_group_transform_generic( &self, diff --git a/crates/dbsp/src/operator/dynamic/multijoin/star_join.rs b/crates/dbsp/src/operator/dynamic/multijoin/star_join.rs index d6097f806cc..0c102232c5e 100644 --- a/crates/dbsp/src/operator/dynamic/multijoin/star_join.rs +++ b/crates/dbsp/src/operator/dynamic/multijoin/star_join.rs @@ -387,7 +387,7 @@ where } } - /// Initialize weight_times[current_index] + /// Initialize `weight_times[current_index]` fn init_weight_times(&mut self) { self.weight_times[self.current_index].0.clear(); let (previous_weight, previous_time) = self.previous_weight_time(); @@ -412,8 +412,8 @@ where self.weight_times[self.current_index].1 = 0; } - /// Increment weight_times[current_index].1 by 1. If it reaches the end of the vector, - /// advance trace_cursors[current_index] to the next value. + /// Increment `weight_times[current_index].1` by 1. If it reaches the end of the + /// vector, advance `trace_cursors[current_index]` to the next value. fn advance_weight_times(&mut self) { if self.weight_times[self.current_index].1 == self.weight_times[self.current_index].0.len() - 1 diff --git a/crates/dbsp/src/storage/file/reader/bulk_rows.rs b/crates/dbsp/src/storage/file/reader/bulk_rows.rs index b0c38d64b7f..b0f03688339 100644 --- a/crates/dbsp/src/storage/file/reader/bulk_rows.rs +++ b/crates/dbsp/src/storage/file/reader/bulk_rows.rs @@ -528,7 +528,7 @@ where /// Returns the next [TreeNode] to read in the level below this one, or /// `None` if we've exhausted this level or there are none to read yet. - /// (Use [eof](Self::eof) to distinguish the meanings of `None`.) + /// (Use [`BulkRows::at_eof`] to distinguish the meanings of `None`.) fn child(&self) -> Result, Error> { self.blocks .front() diff --git a/crates/dbsp/src/trace/spine_async.rs b/crates/dbsp/src/trace/spine_async.rs index a49321dd4fc..5b359f561c5 100644 --- a/crates/dbsp/src/trace/spine_async.rs +++ b/crates/dbsp/src/trace/spine_async.rs @@ -198,7 +198,8 @@ impl Display for CompactionStatus { } } -/// A group of batches with similar sizes (as determined by [size_from_level]). +/// A group of batches with similar sizes (as determined by +/// [`Spine::size_to_level`]). #[derive(Clone, SizeOf)] struct Slot where @@ -259,7 +260,7 @@ where B: Batch, { /// If this slot doesn't currently have an ongoing merge, and it does have - /// at least MERGE_COUNTS[level].start() loose batches, picks an upper limit + /// at least `MERGE_COUNTS[level].start()` loose batches, picks an upper limit /// of the loose batches and makes them into merging batches, and returns /// those batches. Otherwise, returns `None` without changing anything. /// From 059d77770fa65ef2851433359009070a55a99fb8 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Fri, 14 Aug 2026 11:36:33 -0700 Subject: [PATCH 7/7] [dbsp] say which batches the spine's counters count `batch_count` counted only the loose batches, not the batches being merged, so every caller read as if it covered the whole spine. It is now `count_loose_batches`, returning a `LooseBatchCount`, and the locals that held its result are named for what they hold: the type carried the same ambiguity as the function, and so did calling a loose count `batch_count`. The total, which really is every batch, was spelled out as a fold over the slots in six places. It is now `count_all_batches`, so the two counts read as the pair they are. --- crates/dbsp/src/trace/spine_async.rs | 53 ++++++++++++++++------------ 1 file changed, 30 insertions(+), 23 deletions(-) diff --git a/crates/dbsp/src/trace/spine_async.rs b/crates/dbsp/src/trace/spine_async.rs index 5b359f561c5..5004042c704 100644 --- a/crates/dbsp/src/trace/spine_async.rs +++ b/crates/dbsp/src/trace/spine_async.rs @@ -408,8 +408,10 @@ where self.slots[level].notify.notify_one(); } - fn batch_count(&self) -> BatchCount { - BatchCount( + /// Number of batches waiting to be merged, which excludes the ones already + /// being merged. This is the count backpressure is based on. + fn count_loose_batches(&self) -> LooseBatchCount { + LooseBatchCount( self.slots .iter() .map(|s| s.loose_batches.len()) @@ -417,13 +419,18 @@ where ) } + /// Number of batches in the spine, loose and merging alike. + fn count_all_batches(&self) -> usize { + self.slots.iter().map(Slot::n_batches).sum() + } + fn get_filters(&self) -> (Option>, Option>) { (self.key_filter.clone(), self.value_filter.clone()) } /// Gets a copy of all of the batches (whether loose or being merged). fn get_batches(&self) -> Vec> { - let mut batches = Vec::with_capacity(self.slots.iter().map(Slot::n_batches).sum()); + let mut batches = Vec::with_capacity(self.count_all_batches()); for slot in &self.slots { batches.extend(slot.all_batches().cloned()); } @@ -466,7 +473,7 @@ where .slots .iter() .all(|s| s.compaction_status == CompactionStatus::None); - let total_batches: usize = self.slots.iter().map(Slot::n_batches).sum(); + let total_batches = self.count_all_batches(); all_requests_processed && !self.is_merging() && total_batches <= 1 } @@ -527,8 +534,8 @@ where // batch counts describe the spine as the merge leaves it: `loose` is // what backpressure measures, and it is what grows when merging cannot // keep up with the batches each step adds. - let loose = self.batch_count().0; - let total: usize = self.slots.iter().map(Slot::n_batches).sum(); + let loose = self.count_loose_batches().0; + let total = self.count_all_batches(); Span::new(LEVEL_NAMES[level]) .with_category("Spine") .with_start(start) @@ -640,9 +647,9 @@ pub(crate) struct BackpressureWaitReport { } #[derive(Copy, Clone, Debug)] -struct BatchCount(usize); +struct LooseBatchCount(usize); -impl BatchCount { +impl LooseBatchCount { const HIGH_THRESHOLD: usize = 128; fn should_apply_backpressure(&self) -> bool { @@ -813,21 +820,21 @@ where let (name, initial_batches, initial_total); { let state = self.state.lock().unwrap(); - let batch_count = state.batch_count(); - if batch_count.should_relieve_backpressure() { + let loose = state.count_loose_batches(); + if loose.should_relieve_backpressure() { return; } name = state.name.clone(); - initial_batches = batch_count.0; - initial_total = state.slots.iter().map(Slot::n_batches).sum(); + initial_batches = loose.0; + initial_total = state.count_all_batches(); } - // Wait for the batch count to drop below the threshold. + // Wait for the loose batch count to drop below the threshold. loop { let notify = self.no_backpressure.notified(); { let state = self.state.lock().unwrap(); - if state.batch_count().should_relieve_backpressure() { + if state.count_loose_batches().should_relieve_backpressure() { break; } } @@ -852,14 +859,14 @@ where fn backpressure_waiter(&self) -> Option { let notify = self.no_backpressure.clone().notified_owned(); let state = self.state.lock().unwrap(); - let batch_count = state.batch_count(); - batch_count.should_apply_backpressure().then(|| { - let total: usize = state.slots.iter().map(Slot::n_batches).sum(); + let loose = state.count_loose_batches(); + loose.should_apply_backpressure().then(|| { + let total = state.count_all_batches(); BackpressureWait { notify, name: state.name.clone(), start: Instant::now(), - initial_loose: batch_count.0, + initial_loose: loose.0, initial_total: total, } }) @@ -873,8 +880,8 @@ where let mut state = self.state.lock().unwrap(); state.spine_stats.backpressure_wait += elapsed; ( - state.batch_count().0, - state.slots.iter().map(Slot::n_batches).sum::(), + state.count_loose_batches().0, + state.count_all_batches(), state.owner_worker, ) }; @@ -1322,7 +1329,7 @@ where self.no_backpressure.notify_waiters(); break; } - if state.batch_count().should_relieve_backpressure() { + if state.count_loose_batches().should_relieve_backpressure() { self.no_backpressure.notify_waiters(); } } @@ -2126,7 +2133,7 @@ where if self .merger .add_batch(batch, false) - .batch_count() + .count_loose_batches() .should_apply_backpressure() { self.merger.backpressure_wait().await; @@ -2149,7 +2156,7 @@ where self.dirty = true; self.merger .add_batch(batch, false) - .batch_count() + .count_loose_batches() .should_apply_backpressure() } else { false