mm1-common 0.7.21

M/M/1! Queueing, do you speak it?!
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

use metrics::{Counter, Histogram};

#[macro_export]
macro_rules! make_metrics {
    ($name:literal $(, $k:literal => $v:expr )* $(,)?) => {{
        const _: &str = $name;
        const POLL_COUNT: &str = concat!($name, "_poll_count");
        const DONE_COUNT: &str = concat!($name, "_done_count");
        const BUSY_TIME: &str = concat!($name, "_busy_sec");
        const WAIT_TIME: &str = concat!($name, "_wait_sec");
        const WALL_TIME: &str = concat!($name, "_wall_sec");

        let poll_count = ::metrics::counter!(POLL_COUNT $(, $k => $v )*);
        let done_count = ::metrics::counter!(DONE_COUNT $(, $k => $v )*);
        let busy_time = ::metrics::histogram!(BUSY_TIME $(, $k => $v )*);
        let wait_time = ::metrics::histogram!(WAIT_TIME $(, $k => $v )*);
        let wall_time = ::metrics::histogram!(WALL_TIME $(, $k => $v )*);

        $crate::metrics::Metrics {
            poll_count,
            done_count,
            busy_time,
            wait_time,
            wall_time,
        }
    }};
}

pub trait MeasuredFutureExt: Future + Sized {
    fn measured(self, metrics: Metrics) -> MeasuredFuture<Self> {
        MeasuredFuture::new(self, metrics)
    }
}
impl<F> MeasuredFutureExt for F where F: Future + Sized {}

#[pin_project::pin_project]
#[derive(derive_more::Debug)]
pub struct MeasuredFuture<F> {
    #[debug(skip)]
    #[pin]
    inner: F,

    metrics:      Metrics,
    measurements: Measurements,
}

#[derive(Debug)]
pub struct Metrics {
    pub poll_count: Counter,
    pub done_count: Counter,
    pub busy_time:  Histogram,
    pub wait_time:  Histogram,
    pub wall_time:  Histogram,
}

#[derive(Default, Debug)]
struct Measurements {
    first_poll: Option<Instant>,
    last_poll:  Option<Instant>,
    busy_time:  Duration,
    wait_time:  Duration,
}

impl<F> MeasuredFuture<F> {
    pub fn new(inner: F, metrics: Metrics) -> Self {
        Self {
            inner,
            metrics,
            measurements: Default::default(),
        }
    }
}

impl<F> Future for MeasuredFuture<F>
where
    F: Future,
{
    type Output = F::Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();

        this.metrics.poll_count.increment(1);

        let t_start = Instant::now();
        if let Some(last_poll) = this.measurements.last_poll {
            this.measurements.wait_time += t_start.duration_since(last_poll);
        }
        let first_poll = this.measurements.first_poll.get_or_insert(t_start);

        let poll = this.inner.poll(cx);

        let t_end = Instant::now();
        this.measurements.busy_time += t_end.duration_since(t_start);

        if poll.is_ready() {
            this.metrics
                .wall_time
                .record(t_end.duration_since(*first_poll));
            this.metrics.busy_time.record(this.measurements.busy_time);
            this.metrics.wait_time.record(this.measurements.wait_time);

            this.metrics.done_count.increment(1);
        }

        poll
    }
}