pictor_runtime/
stream_metrics.rs

1//! Token streaming metrics: TTFT, inter-token latency (TBT), throughput.
2//!
3//! These metrics are essential for production LLM serving SLAs.
4//!
5//! - **TTFT** — Time To First Token: latency from request start to the first
6//!   generated token.
7//! - **TBT** — Time Between Tokens: inter-token latency during the decode phase.
8//! - **E2E latency** — total request latency from start to generation complete.
9//!
10//! # Usage
11//!
12//! ```rust
13//! use pictor_runtime::stream_metrics::RequestStreamMetrics;
14//!
15//! let mut m = RequestStreamMetrics::new_with_prompt_tokens(128);
16//! // ... first token arrives ...
17//! m.record_first_token();
18//! // ... subsequent tokens ...
19//! m.record_token();
20//! m.record_token();
21//! m.finish();
22//!
23//! let snap = m.snapshot();
24//! println!("{}", snap.summary());
25//! ```
26
27use std::time::{Duration, Instant};
28
29// ── RequestStreamMetrics ─────────────────────────────────────────────────────
30
31/// Per-request streaming metrics collector.
32///
33/// Call [`record_first_token`](RequestStreamMetrics::record_first_token) on the
34/// first generated token, [`record_token`](RequestStreamMetrics::record_token)
35/// for each subsequent token, and
36/// [`finish`](RequestStreamMetrics::finish) when generation is complete.
37pub struct RequestStreamMetrics {
38    request_start: Instant,
39    first_token_time: Option<Instant>,
40    last_token_time: Option<Instant>,
41    /// Wall-clock gaps between consecutive tokens (TBT samples).
42    inter_token_gaps: Vec<Duration>,
43    token_count: usize,
44    prompt_tokens: usize,
45}
46
47impl RequestStreamMetrics {
48    /// Create a new metrics collector with zero prompt tokens.
49    pub fn new() -> Self {
50        Self {
51            request_start: Instant::now(),
52            first_token_time: None,
53            last_token_time: None,
54            inter_token_gaps: Vec::new(),
55            token_count: 0,
56            prompt_tokens: 0,
57        }
58    }
59
60    /// Create a new metrics collector that records the prompt token count.
61    pub fn new_with_prompt_tokens(prompt_tokens: usize) -> Self {
62        Self {
63            prompt_tokens,
64            ..Self::new()
65        }
66    }
67
68    /// Record the arrival of the **first** generated token.
69    ///
70    /// Calling this multiple times is safe — only the first call is recorded.
71    pub fn record_first_token(&mut self) {
72        if self.first_token_time.is_none() {
73            let now = Instant::now();
74            self.first_token_time = Some(now);
75            self.last_token_time = Some(now);
76            self.token_count = 1;
77        }
78    }
79
80    /// Record the arrival of a **subsequent** generated token (not the first).
81    ///
82    /// If [`record_first_token`](Self::record_first_token) has not been called
83    /// yet, this call is treated as the first token.
84    pub fn record_token(&mut self) {
85        let now = Instant::now();
86        if self.first_token_time.is_none() {
87            // Treat as first token.
88            self.first_token_time = Some(now);
89            self.last_token_time = Some(now);
90            self.token_count = 1;
91            return;
92        }
93        if let Some(prev) = self.last_token_time {
94            self.inter_token_gaps.push(now.duration_since(prev));
95        }
96        self.last_token_time = Some(now);
97        self.token_count += 1;
98    }
99
100    /// Mark the end of the generation pass.
101    ///
102    /// This records a final timestamp used for end-to-end latency.  It is safe
103    /// to call even before any tokens have been recorded.
104    pub fn finish(&mut self) {
105        // If no last_token_time has been set we still want an e2e measurement.
106        if self.last_token_time.is_none() {
107            self.last_token_time = Some(Instant::now());
108        }
109    }
110
111    /// Time to first token (TTFT).
112    ///
113    /// Returns `None` if [`record_first_token`](Self::record_first_token) has
114    /// not been called.
115    pub fn ttft(&self) -> Option<Duration> {
116        let first = self.first_token_time?;
117        Some(first.duration_since(self.request_start))
118    }
119
120    /// Median inter-token latency.
121    ///
122    /// Returns `None` when fewer than two tokens have been recorded (i.e. there
123    /// are no TBT samples).
124    pub fn median_tbt(&self) -> Option<Duration> {
125        percentile_duration(&self.inter_token_gaps, 50)
126    }
127
128    /// P99 inter-token latency.
129    ///
130    /// Returns `None` when there are no TBT samples.
131    pub fn p99_tbt(&self) -> Option<Duration> {
132        percentile_duration(&self.inter_token_gaps, 99)
133    }
134
135    /// Mean (arithmetic average) inter-token latency.
136    ///
137    /// Returns `None` when there are no TBT samples.
138    pub fn mean_tbt(&self) -> Option<Duration> {
139        if self.inter_token_gaps.is_empty() {
140            return None;
141        }
142        let total_nanos: u128 = self.inter_token_gaps.iter().map(|d| d.as_nanos()).sum();
143        let mean_nanos = total_nanos / self.inter_token_gaps.len() as u128;
144        Some(Duration::from_nanos(mean_nanos as u64))
145    }
146
147    /// Generation throughput in tokens per second.
148    ///
149    /// Computed over the decode window (from first token to last token) so that
150    /// TTFT does not distort the throughput figure.
151    ///
152    /// Returns `None` if the decode window cannot be determined.
153    pub fn tokens_per_second(&self) -> Option<f64> {
154        if self.token_count < 2 {
155            // Need at least two tokens to define a decode window.
156            return None;
157        }
158        let first = self.first_token_time?;
159        let last = self.last_token_time?;
160        let elapsed = last.duration_since(first);
161        let elapsed_secs = elapsed.as_secs_f64();
162        if elapsed_secs <= 0.0 {
163            return None;
164        }
165        // Denominator is (token_count - 1) inter-token intervals.
166        Some((self.token_count - 1) as f64 / elapsed_secs)
167    }
168
169    /// End-to-end request latency (from request start to generation finish).
170    ///
171    /// Returns `None` if [`finish`](Self::finish) has not been called.
172    pub fn e2e_latency(&self) -> Option<Duration> {
173        let last = self.last_token_time?;
174        Some(last.duration_since(self.request_start))
175    }
176
177    /// Total number of completion tokens generated so far.
178    pub fn completion_tokens(&self) -> usize {
179        self.token_count
180    }
181
182    /// Take a point-in-time snapshot of the current metrics.
183    pub fn snapshot(&self) -> StreamMetricsSnapshot {
184        StreamMetricsSnapshot {
185            ttft_ms: self.ttft().map(duration_to_ms),
186            mean_tbt_ms: self.mean_tbt().map(duration_to_ms),
187            p99_tbt_ms: self.p99_tbt().map(duration_to_ms),
188            tokens_per_second: self.tokens_per_second(),
189            e2e_latency_ms: self.e2e_latency().map(duration_to_ms),
190            completion_tokens: self.token_count,
191            prompt_tokens: self.prompt_tokens,
192        }
193    }
194}
195
196impl Default for RequestStreamMetrics {
197    fn default() -> Self {
198        Self::new()
199    }
200}
201
202// ── StreamMetricsSnapshot ────────────────────────────────────────────────────
203
204/// A point-in-time snapshot of streaming metrics for a single request.
205#[derive(Debug, Clone)]
206pub struct StreamMetricsSnapshot {
207    /// Time to first token in milliseconds, if available.
208    pub ttft_ms: Option<f64>,
209    /// Mean inter-token latency in milliseconds, if available.
210    pub mean_tbt_ms: Option<f64>,
211    /// P99 inter-token latency in milliseconds, if available.
212    pub p99_tbt_ms: Option<f64>,
213    /// Tokens per second during the decode phase, if available.
214    pub tokens_per_second: Option<f64>,
215    /// End-to-end latency in milliseconds, if available.
216    pub e2e_latency_ms: Option<f64>,
217    /// Number of completion tokens generated.
218    pub completion_tokens: usize,
219    /// Number of prompt tokens (prefill).
220    pub prompt_tokens: usize,
221}
222
223impl StreamMetricsSnapshot {
224    /// Format as a one-line human-readable summary.
225    pub fn summary(&self) -> String {
226        let ttft = opt_ms_str(self.ttft_ms, "TTFT");
227        let tbt = opt_ms_str(self.mean_tbt_ms, "mean TBT");
228        let tps = self
229            .tokens_per_second
230            .map(|v| format!("TPS={v:.1}"))
231            .unwrap_or_else(|| "TPS=n/a".to_owned());
232        let e2e = opt_ms_str(self.e2e_latency_ms, "E2E");
233        let tokens = format!("tokens={}/{}", self.completion_tokens, self.prompt_tokens);
234        format!("{ttft} | {tbt} | {tps} | {e2e} | {tokens}")
235    }
236}
237
238// ── StreamingMetricsAggregator ───────────────────────────────────────────────
239
240/// Aggregates [`StreamMetricsSnapshot`]s across multiple requests.
241///
242/// Useful for computing fleet-wide statistics in a serving system.
243pub struct StreamingMetricsAggregator {
244    snapshots: Vec<StreamMetricsSnapshot>,
245}
246
247impl StreamingMetricsAggregator {
248    /// Create an empty aggregator.
249    pub fn new() -> Self {
250        Self {
251            snapshots: Vec::new(),
252        }
253    }
254
255    /// Add a snapshot from a completed request.
256    pub fn record(&mut self, snapshot: StreamMetricsSnapshot) {
257        self.snapshots.push(snapshot);
258    }
259
260    /// Number of requests recorded.
261    pub fn num_requests(&self) -> usize {
262        self.snapshots.len()
263    }
264
265    /// Average TTFT across all requests that have a TTFT measurement.
266    pub fn avg_ttft_ms(&self) -> Option<f64> {
267        avg_opt_field(self.snapshots.iter().map(|s| s.ttft_ms))
268    }
269
270    /// Average tokens-per-second across all requests that have a TPS measurement.
271    pub fn avg_tokens_per_second(&self) -> Option<f64> {
272        avg_opt_field(self.snapshots.iter().map(|s| s.tokens_per_second))
273    }
274
275    /// P99 end-to-end latency across all recorded requests.
276    pub fn p99_e2e_ms(&self) -> Option<f64> {
277        let mut values: Vec<f64> = self
278            .snapshots
279            .iter()
280            .filter_map(|s| s.e2e_latency_ms)
281            .collect();
282        if values.is_empty() {
283            return None;
284        }
285        values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
286        let idx = percentile_index(values.len(), 99);
287        Some(values[idx])
288    }
289
290    /// Total number of completion tokens across all recorded requests.
291    pub fn total_completion_tokens(&self) -> usize {
292        self.snapshots.iter().map(|s| s.completion_tokens).sum()
293    }
294
295    /// Generate a multi-line human-readable report.
296    pub fn report(&self) -> String {
297        let mut out = String::new();
298        out.push_str(&format!("Requests recorded : {}\n", self.num_requests()));
299        match self.avg_ttft_ms() {
300            Some(v) => out.push_str(&format!("Avg TTFT          : {v:.2} ms\n")),
301            None => out.push_str("Avg TTFT          : n/a\n"),
302        }
303        match self.avg_tokens_per_second() {
304            Some(v) => out.push_str(&format!("Avg TPS           : {v:.2} tok/s\n")),
305            None => out.push_str("Avg TPS           : n/a\n"),
306        }
307        match self.p99_e2e_ms() {
308            Some(v) => out.push_str(&format!("P99 E2E latency   : {v:.2} ms\n")),
309            None => out.push_str("P99 E2E latency   : n/a\n"),
310        }
311        out.push_str(&format!(
312            "Total tokens      : {}\n",
313            self.total_completion_tokens()
314        ));
315        out
316    }
317}
318
319impl Default for StreamingMetricsAggregator {
320    fn default() -> Self {
321        Self::new()
322    }
323}
324
325// ── Internal helpers ─────────────────────────────────────────────────────────
326
327/// Convert a [`Duration`] to floating-point milliseconds.
328fn duration_to_ms(d: Duration) -> f64 {
329    d.as_secs_f64() * 1_000.0
330}
331
332/// Compute the p-th percentile (0–100) of a slice of [`Duration`]s.
333///
334/// The slice is copied and sorted internally so the original order is preserved.
335/// Returns `None` when the slice is empty.
336fn percentile_duration(samples: &[Duration], p: u8) -> Option<Duration> {
337    if samples.is_empty() {
338        return None;
339    }
340    let mut sorted: Vec<Duration> = samples.to_vec();
341    sorted.sort_unstable();
342    let idx = percentile_index(sorted.len(), p);
343    Some(sorted[idx])
344}
345
346/// Map a percentile (0–100) to a concrete 0-based index into a sorted slice of
347/// length `n`.
348fn percentile_index(n: usize, p: u8) -> usize {
349    if n == 0 {
350        return 0;
351    }
352    // Ceiling index: ceil((p / 100) * n) - 1, clamped.
353    let idx = ((p as usize) * n).div_ceil(100);
354    idx.saturating_sub(1).min(n - 1)
355}
356
357/// Compute the arithmetic mean of an iterator of `Option<f64>`, ignoring `None`s.
358fn avg_opt_field(iter: impl Iterator<Item = Option<f64>>) -> Option<f64> {
359    let values: Vec<f64> = iter.flatten().collect();
360    if values.is_empty() {
361        return None;
362    }
363    Some(values.iter().sum::<f64>() / values.len() as f64)
364}
365
366/// Format an optional millisecond value as `"label=X.Yms"` or `"label=n/a"`.
367fn opt_ms_str(v: Option<f64>, label: &str) -> String {
368    match v {
369        Some(ms) => format!("{label}={ms:.2}ms"),
370        None => format!("{label}=n/a"),
371    }
372}
373
374// ── Unit tests ───────────────────────────────────────────────────────────────
375
376#[cfg(test)]
377mod tests {
378    use super::*;
379
380    #[test]
381    fn new_metrics_have_no_ttft() {
382        let m = RequestStreamMetrics::new();
383        assert!(m.ttft().is_none());
384    }
385
386    #[test]
387    fn record_first_token_sets_ttft() {
388        let mut m = RequestStreamMetrics::new();
389        m.record_first_token();
390        assert!(m.ttft().is_some());
391    }
392
393    #[test]
394    fn record_multiple_tokens_counts_correctly() {
395        let mut m = RequestStreamMetrics::new();
396        m.record_first_token();
397        m.record_token();
398        m.record_token();
399        assert_eq!(m.completion_tokens(), 3);
400    }
401
402    #[test]
403    fn mean_tbt_available_after_three_tokens() {
404        let mut m = RequestStreamMetrics::new();
405        m.record_first_token();
406        // Spin briefly so there is a measurable gap.
407        std::thread::sleep(Duration::from_micros(100));
408        m.record_token();
409        std::thread::sleep(Duration::from_micros(100));
410        m.record_token();
411        assert!(m.mean_tbt().is_some());
412    }
413
414    #[test]
415    fn tokens_per_second_positive_after_tokens() {
416        let mut m = RequestStreamMetrics::new();
417        m.record_first_token();
418        std::thread::sleep(Duration::from_micros(200));
419        m.record_token();
420        let tps = m.tokens_per_second();
421        assert!(tps.is_some(), "tps should be Some after 2 tokens");
422        assert!(tps.expect("checked") > 0.0, "tps must be positive");
423    }
424
425    #[test]
426    fn e2e_latency_available_after_finish() {
427        let mut m = RequestStreamMetrics::new();
428        m.record_first_token();
429        m.record_token();
430        m.finish();
431        assert!(m.e2e_latency().is_some());
432    }
433
434    #[test]
435    fn snapshot_summary_is_nonempty() {
436        let mut m = RequestStreamMetrics::new_with_prompt_tokens(64);
437        m.record_first_token();
438        m.record_token();
439        m.finish();
440        let snap = m.snapshot();
441        assert!(!snap.summary().is_empty());
442    }
443
444    #[test]
445    fn aggregator_empty_returns_none_avg_ttft() {
446        let agg = StreamingMetricsAggregator::new();
447        assert!(agg.avg_ttft_ms().is_none());
448    }
449
450    #[test]
451    fn aggregator_single_snapshot_avg_ttft_equals_snapshot() {
452        let mut agg = StreamingMetricsAggregator::new();
453        let snap = StreamMetricsSnapshot {
454            ttft_ms: Some(42.0),
455            mean_tbt_ms: None,
456            p99_tbt_ms: None,
457            tokens_per_second: None,
458            e2e_latency_ms: Some(100.0),
459            completion_tokens: 10,
460            prompt_tokens: 5,
461        };
462        agg.record(snap);
463        let avg = agg.avg_ttft_ms().expect("should have avg");
464        assert!((avg - 42.0).abs() < 1e-9);
465    }
466
467    #[test]
468    fn aggregator_multiple_snapshots_averages_correctly() {
469        let mut agg = StreamingMetricsAggregator::new();
470        for ttft in [10.0_f64, 20.0, 30.0] {
471            agg.record(StreamMetricsSnapshot {
472                ttft_ms: Some(ttft),
473                mean_tbt_ms: None,
474                p99_tbt_ms: None,
475                tokens_per_second: None,
476                e2e_latency_ms: Some(ttft * 2.0),
477                completion_tokens: 5,
478                prompt_tokens: 2,
479            });
480        }
481        let avg = agg.avg_ttft_ms().expect("should have avg");
482        assert!((avg - 20.0).abs() < 1e-9, "expected avg=20.0, got {avg}");
483    }
484
485    #[test]
486    fn aggregator_p99_e2e_returns_some_after_records() {
487        let mut agg = StreamingMetricsAggregator::new();
488        for ms in [100.0_f64, 200.0, 300.0, 400.0, 500.0] {
489            agg.record(StreamMetricsSnapshot {
490                ttft_ms: None,
491                mean_tbt_ms: None,
492                p99_tbt_ms: None,
493                tokens_per_second: None,
494                e2e_latency_ms: Some(ms),
495                completion_tokens: 1,
496                prompt_tokens: 1,
497            });
498        }
499        assert!(agg.p99_e2e_ms().is_some());
500    }
501
502    #[test]
503    fn aggregator_total_tokens_sums_correctly() {
504        let mut agg = StreamingMetricsAggregator::new();
505        for tokens in [10usize, 20, 30] {
506            agg.record(StreamMetricsSnapshot {
507                ttft_ms: None,
508                mean_tbt_ms: None,
509                p99_tbt_ms: None,
510                tokens_per_second: None,
511                e2e_latency_ms: None,
512                completion_tokens: tokens,
513                prompt_tokens: 0,
514            });
515        }
516        assert_eq!(agg.total_completion_tokens(), 60);
517    }
518
519    #[test]
520    fn aggregator_report_is_nonempty() {
521        let mut agg = StreamingMetricsAggregator::new();
522        agg.record(StreamMetricsSnapshot {
523            ttft_ms: Some(50.0),
524            mean_tbt_ms: Some(10.0),
525            p99_tbt_ms: Some(20.0),
526            tokens_per_second: Some(100.0),
527            e2e_latency_ms: Some(1000.0),
528            completion_tokens: 50,
529            prompt_tokens: 128,
530        });
531        assert!(!agg.report().is_empty());
532    }
533}
pictor_runtime/stream_metrics.rs

pictor_runtime/
stream_metrics.rs
