diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs index 81570ec1ddf..1daf840269b 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::{ @@ -775,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 @@ -1591,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( @@ -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/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/operator/dynamic/sharded_accumulator.rs b/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs index 3b065cffdfc..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, @@ -390,9 +380,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/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..8b27ced271b 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 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 { + 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/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 ddb5406b76b..5004042c704 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. /// @@ -351,6 +352,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 +375,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(), @@ -400,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()) @@ -409,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()); } @@ -458,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 } @@ -487,21 +502,9 @@ 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(); + let owner = self.owner_worker; if slot.compaction_status == CompactionStatus::InProgress { // We finished merging all batches in the slot as part of compaction. @@ -526,6 +529,30 @@ 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 + // 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.count_loose_batches().0; + let total = self.count_all_batches(); + Span::new(LEVEL_NAMES[level]) + .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} 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(); } /// Returns a copy of the data that the caller can use to construct a @@ -581,10 +608,48 @@ 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 a [`BackpressureWait`] leaves behind once its future has been awaited. +/// +/// 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, + initial_total: usize, +} + #[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 { @@ -690,7 +755,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!( @@ -752,53 +817,93 @@ 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(); - 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_batches = loose.0; + initial_total = state.count_all_batches(); } - // Wait for the batch count to drop below the threshold. - let final_batches = loop { + // Wait for the loose batch count to drop below the threshold. + 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.count_loose_batches().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, + }); + } + + /// 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 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: loose.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, owner) = { + let mut state = self.state.lock().unwrap(); + state.spine_stats.backpressure_wait += elapsed; + ( + state.count_loose_batches().0, + state.count_all_batches(), + state.owner_worker, + ) + }; + 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!( + "{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(); } - 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); @@ -1224,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(); } } @@ -2028,7 +2133,7 @@ where if self .merger .add_batch(batch, false) - .batch_count() + .count_loose_batches() .should_apply_backpressure() { self.merger.backpressure_wait().await; @@ -2051,7 +2156,7 @@ where self.dirty = true; self.merger .add_batch(batch, false) - .batch_count() + .count_loose_batches() .should_apply_backpressure() } else { false @@ -2347,8 +2452,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(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(crate) fn record_backpressure_wait(&self, report: BackpressureWaitReport) { + self.merger.record_backpressure_wait(report) } } 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(); +}