drive_abci/
perf.rs

1//! Lightweight per-block phase timing for debug builds.
2//!
3//! This module and all timing call sites are compiled only with debug assertions
4//! enabled. Standard release builds exclude the instrumentation entirely, even
5//! when `DRIVE_BLOCK_PERF=1` is set.
6//!
7//! In debug builds, enabled only when `DRIVE_BLOCK_PERF=1` is set in the
8//! environment. Phases are accumulated in memory and reported as means every
9//! `DRIVE_BLOCK_PERF_EVERY` blocks (default 500), so the measurement does not
10//! pay for a log line inside the very spans it is measuring.
11//!
12//! This is read straight from the environment rather than through
13//! `PlatformConfig` on purpose: it is a developer switch for replay benchmarks,
14//! it must cost nothing when off, and it should not need a config change to be
15//! flipped on a node under investigation.
16//!
17//! Each [`PhaseTimer`] belongs to a scope, normally the function it lives in,
18//! and every phase is reported as `scope.phase`. Phases are named after the
19//! call they time, so a term in the report can be grepped straight to the code.
20//!
21//! Phases nest where a handler times a call whose body is itself timed:
22//! `finalize_block.finalize_block_proposal` covers all of the
23//! `finalize_block_proposal.*` phases. Add up phases from one level only.
24
25use std::sync::{Mutex, OnceLock, PoisonError};
26use std::time::Instant;
27
28fn enabled() -> bool {
29    static ENABLED: OnceLock<bool> = OnceLock::new();
30    *ENABLED.get_or_init(|| std::env::var("DRIVE_BLOCK_PERF").as_deref() == Ok("1"))
31}
32
33fn report_every() -> u64 {
34    static EVERY: OnceLock<u64> = OnceLock::new();
35    *EVERY.get_or_init(|| {
36        std::env::var("DRIVE_BLOCK_PERF_EVERY")
37            .ok()
38            .and_then(|v| v.parse().ok())
39            .unwrap_or(500)
40    })
41}
42
43/// A phase's identity in the report: the timer's scope and the phase name.
44type PhaseKey = (&'static str, &'static str);
45
46#[derive(Default)]
47struct Totals {
48    blocks: u64,
49    /// (scope, phase, summed microseconds, samples), in first-seen order
50    phases: Vec<(&'static str, &'static str, u64, u64)>,
51}
52
53impl Totals {
54    fn add(&mut self, (scope, phase): PhaseKey, micros: u64) {
55        if let Some(entry) = self
56            .phases
57            .iter_mut()
58            .find(|(s, p, _, _)| *s == scope && *p == phase)
59        {
60            entry.2 += micros;
61            entry.3 += 1;
62        } else {
63            self.phases.push((scope, phase, micros, 1));
64        }
65    }
66
67    /// One `scope.phase=mean/samples` term per phase, space separated. The mean
68    /// is over blocks, not over samples: a phase that only runs on some blocks
69    /// shows its share of the per-block cost, and the sample count shows how
70    /// often it ran. Only called from `end_block`, after a block was counted.
71    fn report_line(&self) -> String {
72        debug_assert!(self.blocks > 0, "report_line before any block was counted");
73        let mut line = String::with_capacity(self.phases.len() * 48);
74        for (scope, phase, sum, samples) in &self.phases {
75            if !line.is_empty() {
76                line.push(' ');
77            }
78            line.push_str(scope);
79            line.push('.');
80            line.push_str(phase);
81            line.push('=');
82            line.push_str(&(*sum / self.blocks.max(1)).to_string());
83            line.push('/');
84            line.push_str(&samples.to_string());
85        }
86        line
87    }
88
89    /// Counts a finished block. Returns the report and resets when the
90    /// reporting interval is reached. An interval of zero reports every block.
91    fn end_block(&mut self, every: u64) -> Option<(u64, String)> {
92        self.blocks += 1;
93        if self.blocks < every {
94            return None;
95        }
96        let report = (self.blocks, self.report_line());
97        self.phases.clear();
98        self.blocks = 0;
99        Some(report)
100    }
101}
102
103fn totals() -> &'static Mutex<Totals> {
104    static TOTALS: OnceLock<Mutex<Totals>> = OnceLock::new();
105    TOTALS.get_or_init(|| Mutex::new(Totals::default()))
106}
107
108/// Times the successive phases of one function during block execution.
109///
110/// A phase is the time between the previous [`end_phase`](Self::end_phase)
111/// (or construction) and this one. Timings are merged into the process-wide
112/// totals when the timer is dropped, under `scope.phase`.
113pub struct PhaseTimer {
114    scope: &'static str,
115    phase_start: Instant,
116    on: bool,
117    buf: Vec<(PhaseKey, u64)>,
118}
119
120impl PhaseTimer {
121    /// Start timing under `scope`, normally the name of the enclosing function.
122    /// Cheap and inert when perf logging is off.
123    pub fn new(scope: &'static str) -> Self {
124        let on = enabled();
125        PhaseTimer {
126            scope,
127            phase_start: Instant::now(),
128            on,
129            buf: if on {
130                Vec::with_capacity(32)
131            } else {
132                Vec::new()
133            },
134        }
135    }
136
137    /// Record the time since the previous phase ended under `phase`.
138    pub fn end_phase(&mut self, phase: &'static str) {
139        self.end_phase_if(true, phase);
140    }
141
142    /// Like [`end_phase`](Self::end_phase), but only records a sample when
143    /// `ran` is true. Use it after work that runs on some blocks only, so the
144    /// sample count in the report is the number of blocks the work actually
145    /// ran on. The next phase starts now either way.
146    pub fn end_phase_if(&mut self, ran: bool, phase: &'static str) {
147        if !self.on {
148            return;
149        }
150        let now = Instant::now();
151        if ran {
152            self.buf.push((
153                (self.scope, phase),
154                now.duration_since(self.phase_start).as_micros() as u64,
155            ));
156        }
157        self.phase_start = now;
158    }
159}
160
161impl Drop for PhaseTimer {
162    fn drop(&mut self) {
163        if !self.on || self.buf.is_empty() {
164            return;
165        }
166        // Telemetry only: a panic elsewhere while the lock was held must not
167        // turn into a second panic here, least of all during unwinding.
168        let mut totals = totals().lock().unwrap_or_else(PoisonError::into_inner);
169        for (key, micros) in self.buf.drain(..) {
170            totals.add(key, micros);
171        }
172    }
173}
174
175/// Called once per finalized block. Emits the means and resets every
176/// `DRIVE_BLOCK_PERF_EVERY` blocks.
177pub fn end_block(height: u64) {
178    if !enabled() {
179        return;
180    }
181    let report = totals()
182        .lock()
183        .unwrap_or_else(PoisonError::into_inner)
184        .end_block(report_every());
185    if let Some((blocks, line)) = report {
186        tracing::info!(
187            block_perf = "agg",
188            height,
189            blocks,
190            phases = line,
191            "block perf"
192        );
193    }
194}
195
196#[cfg(test)]
197mod tests {
198    use super::*;
199    use std::time::Duration;
200
201    #[test]
202    fn phases_keep_first_seen_order_and_sum_samples() {
203        let mut totals = Totals::default();
204        totals.add(("s", "b"), 10);
205        totals.add(("s", "a"), 5);
206        totals.add(("s", "b"), 20);
207        // Same phase name in another scope is a different phase.
208        totals.add(("t", "b"), 1);
209
210        assert_eq!(
211            totals.phases,
212            vec![("s", "b", 30, 2), ("s", "a", 5, 1), ("t", "b", 1, 1)]
213        );
214    }
215
216    #[test]
217    fn report_means_over_blocks_not_over_samples() {
218        let mut totals = Totals::default();
219        // Ran on one block out of four, costing 400 µs that time.
220        totals.add(("run", "rare"), 400);
221        // Ran on every block.
222        for _ in 0..4 {
223            totals.add(("run", "common"), 10);
224        }
225        totals.blocks = 4;
226
227        assert_eq!(totals.report_line(), "run.rare=100/1 run.common=10/4");
228    }
229
230    #[test]
231    fn end_block_reports_and_resets_at_the_interval() {
232        let mut totals = Totals::default();
233        totals.add(("s", "x"), 30);
234        assert_eq!(totals.end_block(3), None);
235        totals.add(("s", "x"), 30);
236        assert_eq!(totals.end_block(3), None);
237        totals.add(("s", "x"), 30);
238
239        assert_eq!(totals.end_block(3), Some((3, "s.x=30/3".to_string())));
240        assert_eq!(totals.blocks, 0);
241        assert!(totals.phases.is_empty());
242    }
243
244    #[test]
245    fn a_phase_that_did_not_run_starts_the_next_phase_without_a_sample() {
246        let mut timer = PhaseTimer {
247            scope: "test",
248            phase_start: Instant::now(),
249            on: true,
250            buf: Vec::new(),
251        };
252        // Put the running phase's start in the past: the skipped phase must
253        // still move the start forward, or the next phase would absorb it.
254        let before = Instant::now();
255        timer.phase_start = before - Duration::from_secs(1);
256        timer.end_phase_if(false, "skipped");
257        assert!(timer.buf.is_empty());
258        assert!(timer.phase_start >= before);
259
260        timer.end_phase_if(true, "ran");
261        assert_eq!(timer.buf.len(), 1);
262        assert_eq!(timer.buf[0].0, ("test", "ran"));
263        assert!(
264            timer.buf[0].1 < 1_000_000,
265            "must not include the skipped second"
266        );
267        // Drop must not merge test phases into the process-wide totals.
268        timer.buf.clear();
269    }
270
271    #[test]
272    fn report_line_is_empty_when_nothing_was_recorded() {
273        let mut totals = Totals::default();
274        assert_eq!(totals.end_block(1), Some((1, String::new())));
275    }
276}
drive_abci/perf.rs

drive_abci/
perf.rs
