reifydb_profiler/
percentile.rs1use std::mem;
5
6use reifydb_value::value::duration::Duration;
7use serde::{Deserialize, Serialize};
8use tdigest::TDigest;
9
10const MAX_CENTROIDS: usize = 100;
11const FLUSH_THRESHOLD: usize = 64;
12
13#[derive(Clone, Debug, Serialize, Deserialize)]
14pub struct PercentileHistogram {
15 digest: TDigest,
16 pending: Vec<f64>,
17}
18
19impl Default for PercentileHistogram {
20 fn default() -> Self {
21 Self::new()
22 }
23}
24
25impl PercentileHistogram {
26 pub fn new() -> Self {
27 Self {
28 digest: TDigest::new_with_size(MAX_CENTROIDS),
29 pending: Vec::new(),
30 }
31 }
32
33 pub fn observe(&mut self, value_us: u32) {
34 self.pending.push(value_us as f64);
35 if self.pending.len() >= FLUSH_THRESHOLD {
36 self.flush();
37 }
38 }
39
40 pub fn merge(&mut self, other: &Self) {
41 let mut combined: Vec<f64> = Vec::with_capacity(self.pending.len() + other.pending.len());
42 combined.append(&mut self.pending);
43 combined.extend(other.pending.iter().copied());
44 let merged = TDigest::merge_digests(vec![self.digest.clone(), other.digest.clone()]);
45 self.digest = if combined.is_empty() {
46 merged
47 } else {
48 merged.merge_unsorted(combined)
49 };
50 }
51
52 pub fn total_count(&self) -> u64 {
53 (self.digest.count() as u64).saturating_add(self.pending.len() as u64)
54 }
55
56 pub fn is_empty(&self) -> bool {
57 self.total_count() == 0
58 }
59
60 pub fn percentile(&self, p: f64) -> u32 {
61 if self.total_count() == 0 {
62 return 0;
63 }
64 let p = p.clamp(0.0, 1.0);
65 let digest_for_read = if self.pending.is_empty() {
66 self.digest.clone()
67 } else {
68 self.digest.clone().merge_unsorted(self.pending.clone())
69 };
70 let estimate = digest_for_read.estimate_quantile(p);
71 estimate.round().max(0.0).min(u32::MAX as f64) as u32
72 }
73
74 pub fn percentiles(&self) -> Percentiles {
75 if self.total_count() == 0 {
76 return Percentiles::default();
77 }
78 let digest_for_read = if self.pending.is_empty() {
79 self.digest.clone()
80 } else {
81 self.digest.clone().merge_unsorted(self.pending.clone())
82 };
83 let read = |p: f64| digest_for_read.estimate_quantile(p).round().max(0.0).min(u32::MAX as f64) as u32;
84 Percentiles {
85 p50: read(0.50),
86 p60: read(0.60),
87 p70: read(0.70),
88 p75: read(0.75),
89 p80: read(0.80),
90 p85: read(0.85),
91 p90: read(0.90),
92 p95: read(0.95),
93 p98: read(0.98),
94 p99: read(0.99),
95 }
96 }
97
98 pub fn percentiles_duration(&self) -> ProfilerPercentiles {
99 let raw = self.percentiles();
100 ProfilerPercentiles {
101 p50: Duration::from_micros_infallible(raw.p50 as u64),
102 p60: Duration::from_micros_infallible(raw.p60 as u64),
103 p70: Duration::from_micros_infallible(raw.p70 as u64),
104 p75: Duration::from_micros_infallible(raw.p75 as u64),
105 p80: Duration::from_micros_infallible(raw.p80 as u64),
106 p85: Duration::from_micros_infallible(raw.p85 as u64),
107 p90: Duration::from_micros_infallible(raw.p90 as u64),
108 p95: Duration::from_micros_infallible(raw.p95 as u64),
109 p98: Duration::from_micros_infallible(raw.p98 as u64),
110 p99: Duration::from_micros_infallible(raw.p99 as u64),
111 }
112 }
113
114 fn flush(&mut self) {
115 if self.pending.is_empty() {
116 return;
117 }
118 let values = mem::take(&mut self.pending);
119 self.digest = self.digest.clone().merge_unsorted(values);
120 }
121}
122
123#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize)]
124pub struct Percentiles {
125 pub p50: u32,
126 pub p60: u32,
127 pub p70: u32,
128 pub p75: u32,
129 pub p80: u32,
130 pub p85: u32,
131 pub p90: u32,
132 pub p95: u32,
133 pub p98: u32,
134 pub p99: u32,
135}
136
137#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize)]
138pub struct ProfilerPercentiles {
139 pub p50: Duration,
140 pub p60: Duration,
141 pub p70: Duration,
142 pub p75: Duration,
143 pub p80: Duration,
144 pub p85: Duration,
145 pub p90: Duration,
146 pub p95: Duration,
147 pub p98: Duration,
148 pub p99: Duration,
149}
150
151#[cfg(test)]
152mod tests {
153 use super::*;
154
155 #[test]
156 fn empty_histogram_returns_zero_for_every_percentile() {
157 let h = PercentileHistogram::new();
158 assert!(h.is_empty());
159 assert_eq!(h.percentile(0.50), 0);
160 assert_eq!(h.percentile(0.99), 0);
161 }
162
163 #[test]
164 fn observe_increments_count() {
165 let mut h = PercentileHistogram::new();
166 h.observe(10);
167 h.observe(20);
168 h.observe(30);
169 assert_eq!(h.total_count(), 3);
170 }
171
172 #[test]
173 fn percentile_brackets_observed_range() {
174 let mut h = PercentileHistogram::new();
175 for v in 1u32..=1000 {
176 h.observe(v);
177 }
178 let p50 = h.percentile(0.50);
179 let p99 = h.percentile(0.99);
180 assert!((400..=600).contains(&p50), "p50={p50} should bracket the median ~500");
181 assert!((900..=1010).contains(&p99), "p99={p99} should bracket ~990");
182 }
183
184 #[test]
185 fn percentile_does_not_exceed_max_observed() {
186 let mut h = PercentileHistogram::new();
187 for v in [51u32, 80, 120, 200, 419] {
188 h.observe(v);
189 }
190 let max_observed = 419u32;
191 let p99 = h.percentile(0.99);
192 assert!(p99 <= max_observed, "p99={p99} exceeded observed max={max_observed}");
193 }
194
195 #[test]
196 fn merge_combines_counts() {
197 let mut a = PercentileHistogram::new();
198 a.observe(10);
199 a.observe(20);
200 let mut b = PercentileHistogram::new();
201 b.observe(30);
202 b.observe(40);
203 a.merge(&b);
204 assert_eq!(a.total_count(), 4);
205 }
206
207 #[test]
208 fn requested_percentiles_are_monotonic() {
209 let mut h = PercentileHistogram::new();
210 for v in [1u32, 5, 10, 20, 50, 100, 200, 500, 1000, 5000] {
211 for _ in 0..50 {
212 h.observe(v);
213 }
214 }
215 let p = h.percentiles();
216 assert!(p.p50 <= p.p60);
217 assert!(p.p60 <= p.p70);
218 assert!(p.p70 <= p.p75);
219 assert!(p.p75 <= p.p80);
220 assert!(p.p80 <= p.p85);
221 assert!(p.p85 <= p.p90);
222 assert!(p.p90 <= p.p95);
223 assert!(p.p95 <= p.p98);
224 assert!(p.p98 <= p.p99);
225 }
226
227 #[test]
228 fn percentile_clamps_p_to_valid_range() {
229 let mut h = PercentileHistogram::new();
230 h.observe(100);
231 assert!(h.percentile(2.0) > 0);
232 let _ = h.percentile(-1.0);
234 }
235}