feldera · ryzhyk · Aug 14, 2026 · Aug 12, 2026 · Aug 13, 2026 · Aug 13, 2026
diff --git 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<dyn StreamMetadata>));
 
 /// 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<C, B>(
@@ -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<dyn Executor<RootCircuit>>,
     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<BootstrapInfo>,
 
     /// 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
@@ -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
@@ -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
@@ -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
@@ -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<OB>(
         &self,

diff --git 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
@@ -101,7 +101,6 @@ where
                         let exchange_id: ExchangeId = runtime.sequence_next().try_into().unwrap();
                         let exchange = ShardedAccumulator::<B>::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<usize>,
         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<usize>,
         clients: Arc<ExchangeClients>,
         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
@@ -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,
                 ];