Measure processor output backpressure

Author: zzylol

asap-precompute-rs/src/bin/asap_stream_bench.rs

@@ -109,6 +109,9 @@ struct Stats {
     completed_windows: u64,
     generator_paused: bool,
     otlp_payload_bytes_sent: u64,
+    processor_wall_ns: u64,
+    output_send_wall_ns: u64,
+    output_messages: u64,
     window_latency_ms: BTreeMap<u64, u64>,
 }
 struct Shared {
@@ -228,6 +231,7 @@ impl processor::Processor<OtapPdata> for StreamProcessor {
         effects: &mut processor::EffectHandler<OtapPdata>,
     ) -> Result<(), Error> {
         if let Message::PData(mut pdata) = message {
+            let process_started = Instant::now();
             let outputs = if let Some(normalizer) = &mut self.normalizer {
                 let count = pdata.num_items();
                 {
@@ -252,9 +256,16 @@ impl processor::Processor<OtapPdata> for StreamProcessor {
                 error,
                 source_detail: String::new(),
             })?;
+            let process_wall_ns = process_started.elapsed().as_nanos() as u64;
+            let output_messages = outputs.len() as u64;
+            let send_started = Instant::now();
             for output in outputs {
                 effects.send_message_with_source_node(output).await?;
             }
+            let mut stats = shared().stats.lock().unwrap();
+            stats.processor_wall_ns += process_wall_ns;
+            stats.output_send_wall_ns += send_started.elapsed().as_nanos() as u64;
+            stats.output_messages += output_messages;
         }
         Ok(())
     }

benchmarks/PROFILING.md

@@ -34,6 +34,14 @@ used by other threads cannot inflate a computation scope. Categories are:
 | `processor_bookkeeping` | Remaining synchronous processor work: benchmark labels/validation, window maps, object destruction, pass-through output allocation |
 | `outside_scopes` | Process CPU minus the above: native OTLP receiver/exporter, runtime, channels, native transport conversion, allocator work outside the processor, control listener and probe overhead |
 
+Every run also records `processor_wall_ns`, `output_send_wall_ns`, and
+`output_messages` per role. These counters distinguish synchronous processor
+work from time awaiting capacity in the processor-to-exporter path. They are
+wall-clock diagnostics and must not be added to the exclusive CPU categories.
+If processor wall time plus output-send wall time approaches the observation
+interval while CPU remains below one core, output backpressure explains the
+gap rather than unused compute capacity.
+
 **The residual is not a measurement of pure OTAP scheduling overhead.** Likewise,
 `pdata_codec` is not all OTAP engine time: it also includes the ASAP adapter's
 per-point allocation and labels. Use the call-stack profiles to inspect owners.

benchmarks/analyze-profile.py

@@ -85,7 +85,7 @@ def family(frames):
 
 def analyze_stacks(root):
     rows = []
-    for path in sorted(root.glob('*mbit/*/*/perf-stacks.txt')):
+    for path in sorted(root.rglob('perf-stacks.txt')):
         parsed = list(samples(path))
         mapping = demangle_names({s for _, frames in parsed for s, _ in frames})
         roles = {int(v): k for k, v in json.loads((path.parent / 'profile-pids.json').read_text()).items()}
@@ -119,7 +119,7 @@ def main():
     parser.add_argument('--control-root', type=Path)
     args = parser.parse_args()
     reports = []
-    for path in sorted(args.root.glob('*mbit/*/*/result.json')):
+    for path in sorted(args.root.rglob('result.json')):
         report = json.loads(path.read_text())
         if 'cpu_profile' not in report:
             raise ValueError(f'{path}: requires --profile-cpu or --account-cpu')
@@ -170,7 +170,8 @@ def main():
     if args.control_root:
         comparison = []
         for row in summary:
-            runs = [json.loads(p.read_text()) for p in args.control_root.glob(f'*mbit/{row["scenario"]}/*/result.json')]
+            runs = [json.loads(p.read_text()) for p in args.control_root.rglob('result.json')
+                    if json.loads(p.read_text()).get('scenario') == row['scenario']]
             if not runs:
                 continue
             inputs = sum(r['cpu_profile']['completed_input_signals'] for r in runs)

benchmarks/run-streaming.py

@@ -333,6 +333,8 @@ def observe(self):
                                "lifetime_peak_rss_mib": end["process_peak_rss_kib"] / 1024,
                                "data_rx_bytes": rx, "data_tx_bytes": tx, "data_rx_mbit_s": rx * 8 / elapsed / 1e6,
                                "data_tx_mbit_s": tx * 8 / elapsed / 1e6}
+            for key in ("processor_wall_ns", "output_send_wall_ns", "output_messages"):
+                resources[role][key] = end["stats"].get(key, 0) - start["stats"].get(key, 0)
             if self.args.profile_cpu:
                 scopes = {key: (value - start["profile_cpu_ns"][key]) / 1e9 for key, value in end["profile_cpu_ns"].items()}
                 resources[role]["scoped_cpu_seconds"] = scopes