lodestone_types/
performance.rs

1//! Bounded process-local stage measurements for development performance work.
2//!
3//! Measurements are reconstructible observations, never correctness or
4//! acknowledgement authority. Each thread owns its counter cell, so hot paths
5//! do not contend on one process-wide atomic. Snapshots walk a bounded registry
6//! of retained cells and never reset cumulative state, including after a worker
7//! thread exits.
8
9use std::cell::RefCell;
10use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
11use std::sync::{Arc, Mutex, OnceLock};
12use std::time::{Duration, Instant};
13
14use serde::Serialize;
15
16const MAX_REGISTERED_THREADS: usize = 128;
17
18/// Stable process-local performance stages used by profile-v2 measurements.
19#[derive(Clone, Copy, Debug, Eq, PartialEq)]
20#[repr(usize)]
21pub enum PerformanceStageV2 {
22    SearchParse = 0,
23    SearchAdmission = 1,
24    SearchMask = 2,
25    SearchExecute = 3,
26    SearchMaterialize = 4,
27    SearchEncode = 5,
28    IngestParseValidate = 6,
29    IngestWalEncode = 7,
30    ObjectFileSync = 8,
31    ObjectDirectorySync = 9,
32    ObjectPutReadback = 10,
33    RaftAppend = 11,
34    CommittedIndexPersistence = 12,
35    TantivyCommit = 13,
36    TantivyReload = 14,
37    ProjectionRebuild = 15,
38    VisibilityProof = 16,
39    IngestResponseMaterialize = 17,
40}
41
42impl PerformanceStageV2 {
43    /// Complete stable stage inventory in serialized order.
44    pub const ALL: [Self; 18] = [
45        Self::SearchParse,
46        Self::SearchAdmission,
47        Self::SearchMask,
48        Self::SearchExecute,
49        Self::SearchMaterialize,
50        Self::SearchEncode,
51        Self::IngestParseValidate,
52        Self::IngestWalEncode,
53        Self::ObjectFileSync,
54        Self::ObjectDirectorySync,
55        Self::ObjectPutReadback,
56        Self::RaftAppend,
57        Self::CommittedIndexPersistence,
58        Self::TantivyCommit,
59        Self::TantivyReload,
60        Self::ProjectionRebuild,
61        Self::VisibilityProof,
62        Self::IngestResponseMaterialize,
63    ];
64
65    /// Stable snake-case label emitted in receipts and reports.
66    #[must_use]
67    pub const fn as_str(self) -> &'static str {
68        match self {
69            Self::SearchParse => "search_parse",
70            Self::SearchAdmission => "search_admission",
71            Self::SearchMask => "search_mask",
72            Self::SearchExecute => "search_execute",
73            Self::SearchMaterialize => "search_materialize",
74            Self::SearchEncode => "search_encode",
75            Self::IngestParseValidate => "ingest_parse_validate",
76            Self::IngestWalEncode => "ingest_wal_encode",
77            Self::ObjectFileSync => "object_file_sync",
78            Self::ObjectDirectorySync => "object_directory_sync",
79            Self::ObjectPutReadback => "object_put_readback",
80            Self::RaftAppend => "raft_append",
81            Self::CommittedIndexPersistence => "committed_index_persistence",
82            Self::TantivyCommit => "tantivy_commit",
83            Self::TantivyReload => "tantivy_reload",
84            Self::ProjectionRebuild => "projection_rebuild",
85            Self::VisibilityProof => "visibility_proof",
86            Self::IngestResponseMaterialize => "ingest_response_materialize",
87        }
88    }
89
90    const fn index(self) -> usize {
91        self as usize
92    }
93}
94
95#[derive(Debug)]
96struct ThreadCounters {
97    observations: [AtomicU64; PerformanceStageV2::ALL.len()],
98    elapsed_nanos: [AtomicU64; PerformanceStageV2::ALL.len()],
99}
100
101impl ThreadCounters {
102    fn new() -> Self {
103        Self {
104            observations: std::array::from_fn(|_| AtomicU64::new(0)),
105            elapsed_nanos: std::array::from_fn(|_| AtomicU64::new(0)),
106        }
107    }
108
109    fn record(&self, stage: PerformanceStageV2, elapsed: Duration) {
110        let index = stage.index();
111        self.observations[index].fetch_add(1, Ordering::Relaxed);
112        let elapsed = u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX);
113        let _ = self.elapsed_nanos[index].fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
114            Some(current.saturating_add(elapsed))
115        });
116    }
117}
118
119#[derive(Debug)]
120struct Registry {
121    threads: Mutex<Vec<Arc<ThreadCounters>>>,
122    overflow: Arc<ThreadCounters>,
123    overflow_assignments: AtomicU64,
124    observation_enabled: AtomicBool,
125}
126
127impl Registry {
128    fn new() -> Self {
129        Self {
130            threads: Mutex::new(Vec::new()),
131            overflow: Arc::new(ThreadCounters::new()),
132            overflow_assignments: AtomicU64::new(0),
133            observation_enabled: AtomicBool::new(true),
134        }
135    }
136
137    fn register_thread(&self) -> Arc<ThreadCounters> {
138        let mut threads = match self.threads.lock() {
139            Ok(threads) => threads,
140            Err(poisoned) => poisoned.into_inner(),
141        };
142        if threads.len() >= MAX_REGISTERED_THREADS {
143            self.overflow_assignments.fetch_add(1, Ordering::Relaxed);
144            return Arc::clone(&self.overflow);
145        }
146        let counters = Arc::new(ThreadCounters::new());
147        threads.push(Arc::clone(&counters));
148        counters
149    }
150
151    fn registered_threads(&self) -> Vec<Arc<ThreadCounters>> {
152        let threads = match self.threads.lock() {
153            Ok(threads) => threads,
154            Err(poisoned) => poisoned.into_inner(),
155        };
156        threads.clone()
157    }
158}
159
160static REGISTRY: OnceLock<Registry> = OnceLock::new();
161
162fn registry() -> &'static Registry {
163    REGISTRY.get_or_init(Registry::new)
164}
165
166thread_local! {
167    static THREAD_COUNTERS: RefCell<Option<Arc<ThreadCounters>>> = const { RefCell::new(None) };
168}
169
170fn thread_counters() -> Arc<ThreadCounters> {
171    THREAD_COUNTERS.with(|slot| {
172        if let Ok(slot) = slot.try_borrow()
173            && let Some(counters) = slot.as_ref()
174        {
175            return Arc::clone(counters);
176        }
177        let counters = registry().register_thread();
178        if let Ok(mut slot) = slot.try_borrow_mut() {
179            *slot = Some(Arc::clone(&counters));
180        }
181        counters
182    })
183}
184
185/// One cumulative reading for a stable stage.
186#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
187pub struct PerformanceStageReadingV2 {
188    pub stage: &'static str,
189    pub observations: u64,
190    pub elapsed_nanos: u64,
191}
192
193/// One bounded, cumulative process-local snapshot.
194#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
195pub struct BoundedStageSnapshotV2 {
196    pub format_version: u32,
197    pub readings: Vec<PerformanceStageReadingV2>,
198    pub registered_threads: u64,
199    pub overflow_assignments: u64,
200}
201
202impl BoundedStageSnapshotV2 {
203    /// Returns a saturating stage-by-stage delta from an earlier snapshot.
204    #[must_use]
205    pub fn since(&self, earlier: &Self) -> Self {
206        let readings = self
207            .readings
208            .iter()
209            .zip(&earlier.readings)
210            .map(|(current, prior)| PerformanceStageReadingV2 {
211                stage: current.stage,
212                observations: current.observations.saturating_sub(prior.observations),
213                elapsed_nanos: current.elapsed_nanos.saturating_sub(prior.elapsed_nanos),
214            })
215            .collect();
216        Self {
217            format_version: 2,
218            readings,
219            registered_threads: self.registered_threads,
220            overflow_assignments: self.overflow_assignments.saturating_sub(earlier.overflow_assignments),
221        }
222    }
223
224    /// Reads one stage by stable identity.
225    #[must_use]
226    pub fn reading(&self, stage: PerformanceStageV2) -> Option<&PerformanceStageReadingV2> {
227        self.readings.get(stage.index())
228    }
229}
230
231/// Takes a non-resetting process-local snapshot from at most 128 thread cells
232/// plus one explicit overflow cell.
233#[must_use]
234pub fn bounded_stage_snapshot_v2() -> BoundedStageSnapshotV2 {
235    let registry = registry();
236    let threads = registry.registered_threads();
237    let mut observations = [0_u64; PerformanceStageV2::ALL.len()];
238    let mut elapsed_nanos = [0_u64; PerformanceStageV2::ALL.len()];
239    for counters in threads.iter().chain(std::iter::once(&registry.overflow)) {
240        for stage in PerformanceStageV2::ALL {
241            let index = stage.index();
242            observations[index] =
243                observations[index].saturating_add(counters.observations[index].load(Ordering::Relaxed));
244            elapsed_nanos[index] =
245                elapsed_nanos[index].saturating_add(counters.elapsed_nanos[index].load(Ordering::Relaxed));
246        }
247    }
248    BoundedStageSnapshotV2 {
249        format_version: 2,
250        readings: PerformanceStageV2::ALL
251            .into_iter()
252            .map(|stage| {
253                let index = stage.index();
254                PerformanceStageReadingV2 {
255                    stage: stage.as_str(),
256                    observations: observations[index],
257                    elapsed_nanos: elapsed_nanos[index],
258                }
259            })
260            .collect(),
261        registered_threads: u64::try_from(threads.len()).unwrap_or(u64::MAX),
262        overflow_assignments: registry.overflow_assignments.load(Ordering::Relaxed),
263    }
264}
265
266/// Records an already measured exclusive duration.
267pub fn record_performance_stage_v2(stage: PerformanceStageV2, elapsed: Duration) {
268    if registry().observation_enabled.load(Ordering::Relaxed) {
269        thread_counters().record(stage, elapsed);
270    }
271}
272
273/// RAII timer for request-sized or I/O-sized stage boundaries.
274pub struct PerformanceObservationV2 {
275    stage: PerformanceStageV2,
276    started: Option<Instant>,
277}
278
279impl PerformanceObservationV2 {
280    #[must_use]
281    pub fn start(stage: PerformanceStageV2) -> Self {
282        let started = registry()
283            .observation_enabled
284            .load(Ordering::Relaxed)
285            .then(Instant::now);
286        Self { stage, started }
287    }
288}
289
290impl Drop for PerformanceObservationV2 {
291    fn drop(&mut self) {
292        if let Some(started) = self.started {
293            record_performance_stage_v2(self.stage, started.elapsed());
294        }
295    }
296}
297
298#[cfg(test)]
299fn set_observation_enabled_for_test(enabled: bool) {
300    registry().observation_enabled.store(enabled, Ordering::SeqCst);
301}
302
303#[cfg(test)]
304mod tests {
305    use std::hint::black_box;
306    use std::sync::MutexGuard;
307
308    use super::*;
309
310    static PERFORMANCE_TEST_LOCK: Mutex<()> = Mutex::new(());
311
312    struct RestoreObservation;
313
314    impl Drop for RestoreObservation {
315        fn drop(&mut self) {
316            set_observation_enabled_for_test(true);
317        }
318    }
319
320    fn request_sized_work() {
321        let mut value = 0x9e37_79b9_7f4a_7c15_u64;
322        for iteration in 0..250_000_u64 {
323            value = value.rotate_left(7) ^ iteration.wrapping_mul(0xd6e8_feb8_6659_fd93);
324            black_box(value);
325        }
326    }
327
328    fn test_lock() -> MutexGuard<'static, ()> {
329        match PERFORMANCE_TEST_LOCK.lock() {
330            Ok(guard) => guard,
331            Err(poisoned) => poisoned.into_inner(),
332        }
333    }
334
335    fn measure_request_window(observed: bool) -> u64 {
336        set_observation_enabled_for_test(observed);
337        let started = Instant::now();
338        for _ in 0..64 {
339            let _stage = PerformanceObservationV2::start(PerformanceStageV2::SearchExecute);
340            request_sized_work();
341        }
342        u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX)
343    }
344
345    fn percentile(values: &mut [u64], numerator: usize, denominator: usize) -> u64 {
346        values.sort_unstable();
347        let index = values
348            .len()
349            .saturating_sub(1)
350            .saturating_mul(numerator)
351            .div_ceil(denominator);
352        values[index.min(values.len().saturating_sub(1))]
353    }
354
355    #[test]
356    fn performance_v2_snapshot_is_bounded_monotonic_and_stage_exact() {
357        let _guard = test_lock();
358        let before = bounded_stage_snapshot_v2();
359        record_performance_stage_v2(PerformanceStageV2::ObjectDirectorySync, Duration::from_nanos(17));
360        let after = bounded_stage_snapshot_v2();
361        let delta = after.since(&before);
362        let reading = delta
363            .reading(PerformanceStageV2::ObjectDirectorySync)
364            .map(|reading| (reading.observations, reading.elapsed_nanos));
365        assert_eq!(reading, Some((1, 17)));
366        assert_eq!(delta.readings.len(), PerformanceStageV2::ALL.len());
367        assert!(after.registered_threads <= MAX_REGISTERED_THREADS as u64);
368    }
369
370    #[test]
371    fn performance_v2_snapshot_retains_completed_worker_observations() {
372        let _guard = test_lock();
373        let before = bounded_stage_snapshot_v2();
374        let worker = std::thread::spawn(|| {
375            record_performance_stage_v2(PerformanceStageV2::VisibilityProof, Duration::from_nanos(31));
376        });
377        assert!(worker.join().is_ok());
378
379        let delta = bounded_stage_snapshot_v2().since(&before);
380        let reading = delta
381            .reading(PerformanceStageV2::VisibilityProof)
382            .map(|reading| (reading.observations, reading.elapsed_nanos));
383        assert_eq!(reading, Some((1, 31)));
384    }
385
386    #[test]
387    fn performance_v2_no_observation_control_stays_within_three_percent() {
388        let _guard = test_lock();
389        let _restore = RestoreObservation;
390        let mut observed = Vec::with_capacity(41);
391        let mut unobserved = Vec::with_capacity(41);
392        request_sized_work();
393        for trial in 0..41 {
394            if trial % 2 == 0 {
395                observed.push(measure_request_window(true));
396                unobserved.push(measure_request_window(false));
397            } else {
398                unobserved.push(measure_request_window(false));
399                observed.push(measure_request_window(true));
400            }
401        }
402        let observed_p50 = percentile(&mut observed.clone(), 50, 100);
403        let observed_p95 = percentile(&mut observed, 95, 100);
404        let unobserved_p50 = percentile(&mut unobserved.clone(), 50, 100);
405        let unobserved_p95 = percentile(&mut unobserved, 95, 100);
406        assert!(
407            observed_p50 <= unobserved_p50.saturating_mul(103).div_ceil(100),
408            "observed_p50={observed_p50} unobserved_p50={unobserved_p50}"
409        );
410        assert!(
411            observed_p95 <= unobserved_p95.saturating_mul(103).div_ceil(100),
412            "observed_p95={observed_p95} unobserved_p95={unobserved_p95}"
413        );
414    }
415}
lodestone_types/performance.rs

lodestone_types/
performance.rs
