perf/
stats.rs

1use hdrhistogram::Histogram;
2use noq::StreamId;
3use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
4use std::sync::{Arc, Mutex};
5use std::time::{Duration, Instant, SystemTime};
6#[cfg(feature = "json-output")]
7use {std::fs::File, std::io, std::path::Path};
8
9pub struct Stats {
10    /// Test start time
11    start_instant: Instant,
12    /// Test start system time
13    start: SystemTime,
14    /// Durations of uploads
15    upload_duration: Histogram<u64>,
16    /// Durations of downloads
17    download_duration: Histogram<u64>,
18    /// Time from finishing the upload until receiving the first byte of the response
19    fbl: Histogram<u64>,
20    /// Throughput for uploads
21    upload_throughput: Histogram<u64>,
22    /// Throughput for downloads
23    download_throughput: Histogram<u64>,
24    /// The total amount of requests executed
25    requests: usize,
26    /// Stats accumulated over each interval
27    intervals: Vec<Interval>,
28}
29
30impl Default for Stats {
31    fn default() -> Self {
32        Self {
33            start_instant: Instant::now(),
34            start: SystemTime::now(),
35            upload_duration: Histogram::new(3).unwrap(),
36            download_duration: Histogram::new(3).unwrap(),
37            fbl: Histogram::new(3).unwrap(),
38            upload_throughput: Histogram::new(3).unwrap(),
39            download_throughput: Histogram::new(3).unwrap(),
40            requests: 0,
41            intervals: vec![],
42        }
43    }
44}
45
46impl Stats {
47    pub fn on_interval(&mut self, start: Instant, stream_stats: &OpenStreamStats) {
48        let mut interval = Interval::new(start - self.start_instant, self.start_instant.elapsed());
49        let mut guard = stream_stats.0.lock().unwrap();
50
51        guard.retain(|stream_stats| {
52            self.record(stream_stats.clone());
53            interval.record_stream_stats(stream_stats.clone());
54            // Retain if not finished yet
55            !stream_stats.finished.load(Ordering::SeqCst)
56        });
57
58        self.intervals.push(interval);
59    }
60
61    fn record(&mut self, stream_stats: Arc<StreamStats>) {
62        if stream_stats.finished.load(Ordering::SeqCst) {
63            let duration = stream_stats.duration.load(Ordering::SeqCst);
64            let bps = throughput_bytes_per_second(duration, stream_stats.request_size);
65
66            if stream_stats.sender {
67                self.upload_throughput.record(bps as u64).unwrap();
68                self.upload_duration.record(duration).unwrap();
69            } else {
70                self.download_throughput.record(bps as u64).unwrap();
71                self.download_duration.record(duration).unwrap();
72                self.fbl
73                    .record(stream_stats.first_byte_latency.load(Ordering::SeqCst))
74                    .unwrap();
75                self.requests += 1;
76            }
77        }
78    }
79
80    pub fn print(&self) {
81        let dt = self.start_instant.elapsed();
82        let rps = self.requests as f64 / dt.as_secs_f64();
83
84        println!("Overall stats:");
85        println!(
86            "RPS: {:.2} ({} requests in {:4.2?})",
87            rps, self.requests, dt,
88        );
89        println!();
90
91        println!("Stream metrics:\n");
92
93        println!(
94            "      │ Upload Duration │ Download Duration | FBL        | Upload Throughput | Download Throughput"
95        );
96        println!(
97            "──────┼─────────────────┼───────────────────┼────────────┼───────────────────┼────────────────────"
98        );
99
100        let print_metric = |label: &'static str, get_metric: fn(&Histogram<u64>) -> u64| {
101            println!(
102                " {} │ {:>15.2?} │ {:>17.2?} │  {:>9.2?} │ {:12.2} Mb/s │ {:13.2} Mb/s",
103                label,
104                Duration::from_micros(get_metric(&self.upload_duration)),
105                Duration::from_micros(get_metric(&self.download_duration)),
106                Duration::from_micros(get_metric(&self.fbl)),
107                get_metric(&self.upload_throughput) as f64 * 8.0 / 1000.0 / 1000.0,
108                get_metric(&self.download_throughput) as f64 * 8.0 / 1000.0 / 1000.0,
109            );
110        };
111
112        print_metric("AVG ", |hist| hist.mean() as u64);
113        print_metric("P0  ", |hist| hist.value_at_quantile(0.00));
114        print_metric("P10 ", |hist| hist.value_at_quantile(0.10));
115        print_metric("P50 ", |hist| hist.value_at_quantile(0.50));
116        print_metric("P90 ", |hist| hist.value_at_quantile(0.90));
117        print_metric("P100", |hist| hist.value_at_quantile(1.00));
118        println!();
119    }
120
121    #[cfg(feature = "json-output")]
122    pub fn print_json(&self, path: &Path) -> io::Result<()> {
123        if path == Path::new("-") {
124            json::print(self, io::stdout());
125        } else {
126            let file = File::create(path)?;
127            json::print(self, file)
128        }
129        Ok(())
130    }
131}
132
133/// Statistics for the currently open streams
134#[derive(Clone, Default)]
135pub struct OpenStreamStats(Arc<Mutex<Vec<Arc<StreamStats>>>>);
136
137impl OpenStreamStats {
138    pub fn new_sender(&self, stream: &noq::SendStream, upload_size: u64) -> Arc<StreamStats> {
139        let send_stream_stats = StreamStats {
140            id: stream.id(),
141            request_size: upload_size,
142            bytes: Default::default(),
143            sender: true,
144            finished: Default::default(),
145            duration: Default::default(),
146            first_byte_latency: Default::default(),
147        };
148        let send_stream_stats = Arc::new(send_stream_stats);
149        self.push(send_stream_stats.clone());
150        send_stream_stats
151    }
152
153    pub fn new_receiver(&self, stream: &noq::RecvStream, download_size: u64) -> Arc<StreamStats> {
154        let recv_stream_stats = StreamStats {
155            id: stream.id(),
156            request_size: download_size,
157            bytes: Default::default(),
158            sender: false,
159            finished: Default::default(),
160            duration: Default::default(),
161            first_byte_latency: Default::default(),
162        };
163        let recv_stream_stats = Arc::new(recv_stream_stats);
164        self.push(recv_stream_stats.clone());
165        recv_stream_stats
166    }
167
168    fn push(&self, stream_stats: Arc<StreamStats>) {
169        self.0.lock().unwrap().push(stream_stats);
170    }
171}
172
173pub struct StreamStats {
174    id: StreamId,
175    request_size: u64,
176    bytes: AtomicUsize,
177    sender: bool,
178    finished: AtomicBool,
179    duration: AtomicU64,
180    first_byte_latency: AtomicU64,
181}
182
183impl StreamStats {
184    pub fn on_first_byte(&self, latency: Duration) {
185        self.first_byte_latency
186            .store(latency.as_micros() as u64, Ordering::SeqCst);
187    }
188
189    pub fn on_bytes(&self, bytes: usize) {
190        self.bytes.fetch_add(bytes, Ordering::SeqCst);
191    }
192
193    pub fn finish(&self, duration: Duration) {
194        self.duration
195            .store(duration.as_micros() as u64, Ordering::SeqCst);
196        self.finished.store(true, Ordering::SeqCst);
197    }
198}
199
200struct Interval {
201    streams: Vec<StreamIntervalStats>,
202    period: IntervalPeriod,
203}
204
205impl Interval {
206    fn new(start: Duration, end: Duration) -> Self {
207        let period = IntervalPeriod {
208            start: start.as_secs_f64(),
209            end: end.as_secs_f64(),
210            seconds: (end - start).as_secs_f64(),
211        };
212
213        Self {
214            streams: vec![],
215            period,
216        }
217    }
218
219    fn record_stream_stats(&mut self, stream_stats: Arc<StreamStats>) {
220        let bytes = stream_stats.bytes.swap(0, Ordering::SeqCst);
221        self.streams.push(StreamIntervalStats {
222            id: stream_stats.id,
223            bytes,
224            sender: stream_stats.sender,
225        })
226    }
227}
228
229struct IntervalPeriod {
230    start: f64,
231    end: f64,
232    seconds: f64,
233}
234
235struct StreamIntervalStats {
236    id: StreamId,
237    bytes: usize,
238    sender: bool,
239}
240
241fn throughput_bytes_per_second(duration_in_micros: u64, size: u64) -> f64 {
242    (size as f64) / (duration_in_micros as f64 / 1000000.0)
243}
244
245#[cfg(feature = "json-output")]
246mod json {
247    use crate::stats;
248    use crate::stats::{Stats, StreamIntervalStats};
249    use noq::StreamId;
250    use serde::{self, Serialize, Serializer, ser::SerializeStruct};
251    use std::io::Write;
252    use std::time::{SystemTime, UNIX_EPOCH};
253
254    pub(crate) fn print<W: Write>(stats: &Stats, out: W) {
255        let report = Report {
256            start: Start {
257                timestamp: stats.start,
258            },
259            intervals: &stats
260                .intervals
261                .iter()
262                .map(Interval::from_stats_interval)
263                .collect(),
264        };
265
266        serde_json::to_writer(out, &report).unwrap();
267    }
268
269    #[derive(Serialize)]
270    struct Report<'a> {
271        start: Start,
272        intervals: &'a Vec<Interval>,
273        // TODO: add end stats
274    }
275
276    #[derive(Serialize)]
277    struct Start {
278        #[serde(serialize_with = "serialize_timestamp")]
279        timestamp: SystemTime,
280    }
281
282    fn serialize_timestamp<S>(time: &SystemTime, s: S) -> Result<S::Ok, S::Error>
283    where
284        S: Serializer,
285    {
286        use serde::ser::SerializeMap;
287        let mut state = s.serialize_map(Some(1))?;
288        state.serialize_entry(
289            "timesecs",
290            &time.duration_since(UNIX_EPOCH).unwrap().as_secs(),
291        )?;
292        state.end()
293    }
294
295    struct Interval {
296        streams: Vec<Stream>,
297        recv_sum: Sum,
298        send_sum: Sum,
299    }
300
301    impl Interval {
302        fn from_stats_interval(interval: &stats::Interval) -> Self {
303            Self {
304                streams: interval
305                    .streams
306                    .iter()
307                    .map(|stats| Stream::from_stream_interval_stats(stats, &interval.period))
308                    .collect(),
309                recv_sum: Sum::from_stream_interval_stats(
310                    &interval.streams,
311                    &interval.period,
312                    false,
313                ),
314                send_sum: Sum::from_stream_interval_stats(
315                    &interval.streams,
316                    &interval.period,
317                    true,
318                ),
319            }
320        }
321    }
322
323    impl Serialize for Interval {
324        fn serialize<S>(
325            &self,
326            serializer: S,
327        ) -> Result<<S as Serializer>::Ok, <S as Serializer>::Error>
328        where
329            S: Serializer,
330        {
331            let mut state = serializer.serialize_struct("Interval", 2)?;
332            state.serialize_field("streams", &self.streams)?;
333            // iperf3 outputs duplicate "sum" entries when run in bidirectional mode
334            // serde does not support duplicate keys, so only output one of the sums
335            if self.send_sum.bytes > 0 {
336                state.serialize_field("sum", &self.send_sum)?;
337            } else {
338                state.serialize_field("sum", &self.recv_sum)?;
339            }
340            state.end()
341        }
342    }
343
344    #[derive(Serialize)]
345    struct Stream {
346        #[serde(serialize_with = "serialize_stream_id")]
347        id: StreamId,
348        start: f64,
349        end: f64,
350        seconds: f64,
351        bytes: usize,
352        bits_per_second: f64,
353        sender: bool,
354    }
355
356    impl Stream {
357        fn from_stream_interval_stats(
358            stats: &StreamIntervalStats,
359            period: &stats::IntervalPeriod,
360        ) -> Self {
361            let bits_per_second = stats.bytes as f64 * 8.0 / period.seconds;
362
363            Self {
364                id: stats.id,
365                start: period.start,
366                end: period.end,
367                seconds: period.seconds,
368                bytes: stats.bytes,
369                bits_per_second,
370                sender: stats.sender,
371            }
372        }
373    }
374
375    fn serialize_stream_id<S: Serializer>(id: &StreamId, serializer: S) -> Result<S::Ok, S::Error> {
376        serializer.serialize_u64(u64::from(*id))
377    }
378
379    #[derive(Serialize)]
380    struct Sum {
381        start: f64,
382        end: f64,
383        seconds: f64,
384        bytes: usize,
385        bits_per_second: f64,
386        sender: bool,
387    }
388
389    impl Sum {
390        fn from_stream_interval_stats(
391            stats: &[StreamIntervalStats],
392            period: &stats::IntervalPeriod,
393            sender: bool,
394        ) -> Self {
395            let bytes = stats
396                .iter()
397                .filter(|stat| stat.sender == sender)
398                .map(|stat| stat.bytes)
399                .sum();
400            let bits_per_second = bytes as f64 * 8.0 / period.seconds;
401
402            Self {
403                start: period.start,
404                end: period.end,
405                seconds: period.seconds,
406                bytes,
407                bits_per_second,
408                sender,
409            }
410        }
411    }
412}
perf/stats.rs

perf/
stats.rs
