hotpath 0.22.0

One profiler for CPU, time, memory, SQL, and async code - quickly find and debug performance bottlenecks.
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
113
114
115
116
117
118
119
120
121
122
123

//! Instrumented wrapper for [`tokio::sync::Mutex`].

use tokio::sync::Mutex as TokioMutex;

use crate::instant::Instant;
use crate::mutexes::{
    cancel_wait_stamp, elapsed_nanos, register_mutex, send_mutex_event, wait_stamp,
    InstrumentMutex, MutexEvent,
};

/// Instrumented drop-in replacement for [`tokio::sync::Mutex`].
///
/// Not constructed directly - use the [`mutex!`](crate::mutex) macro.
pub struct Mutex<T> {
    inner: TokioMutex<T>,
    id: u32,
}

#[cfg_attr(feature = "hotpath-meta", hotpath_meta::measure_all)]
impl<T> Mutex<T> {
    /// Drop-in constructor for the `hotpath::wrap` prefix migration. Captures the
    /// caller location as the registered source.
    #[track_caller]
    #[deprecated(note = "construct via the hotpath::mutex! macro instead of new()")]
    pub fn new(value: T) -> Self {
        let loc = std::panic::Location::caller();
        let source: &'static str =
            Box::leak(format!("{}:{}", loc.file(), loc.line()).into_boxed_str());
        Self::__new_instrumented(TokioMutex::new(value), source, None)
    }

    #[doc(hidden)]
    pub fn __new_instrumented(
        inner: TokioMutex<T>,
        source: &'static str,
        label: Option<String>,
    ) -> Self {
        let id = register_mutex::<T>(source, label);
        Self { inner, id }
    }

    pub async fn lock(&self) -> MutexGuard<'_, T> {
        // Stamp before acquisition to measure wait time; the guard then measures acquire time.
        let wait_start = wait_stamp();
        let inner = self.inner.lock().await;
        self.guard(inner, wait_start.map(elapsed_nanos))
    }

    pub fn try_lock(&self) -> Result<MutexGuard<'_, T>, tokio::sync::TryLockError> {
        let wait_start = wait_stamp();
        let inner = self.inner.try_lock();
        if inner.is_err() {
            cancel_wait_stamp();
        }
        inner.map(|inner| self.guard(inner, wait_start.map(elapsed_nanos)))
    }

    pub fn into_inner(self) -> T {
        self.inner.into_inner()
    }

    pub fn get_mut(&mut self) -> &mut T {
        self.inner.get_mut()
    }

    fn guard<'a>(
        &'a self,
        inner: tokio::sync::MutexGuard<'a, T>,
        wait_nanos: Option<u64>,
    ) -> MutexGuard<'a, T> {
        MutexGuard {
            inner: Some(inner),
            start: wait_nanos.map(|_| Instant::now()),
            wait_nanos,
            id: self.id,
        }
    }
}

/// Guard returned by [`Mutex::lock`]. Emits wait and acquire durations on drop.
#[must_use = "if unused the Mutex will immediately unlock"]
pub struct MutexGuard<'a, T> {
    inner: Option<tokio::sync::MutexGuard<'a, T>>,
    start: Option<Instant>,
    wait_nanos: Option<u64>,
    id: u32,
}

impl<T> std::ops::Deref for MutexGuard<'_, T> {
    type Target = T;
    fn deref(&self) -> &T {
        self.inner.as_ref().expect("guard held until drop")
    }
}

impl<T> std::ops::DerefMut for MutexGuard<'_, T> {
    fn deref_mut(&mut self) -> &mut T {
        self.inner.as_mut().expect("guard held until drop")
    }
}

impl<T> Drop for MutexGuard<'_, T> {
    fn drop(&mut self) {
        // Release the real lock before stamping/sending so the held duration
        // excludes the event-send cost and the lock frees as early as possible.
        drop(self.inner.take());
        let acquire_nanos = self
            .start
            .map(|start| Instant::now().duration_since(start).as_nanos() as u64);
        send_mutex_event(MutexEvent::Released {
            id: self.id,
            wait_nanos: self.wait_nanos,
            acquire_nanos,
        });
    }
}

impl<T> InstrumentMutex for TokioMutex<T> {
    type Output = Mutex<T>;
    fn instrument(self, source: &'static str, label: Option<String>) -> Self::Output {
        Mutex::__new_instrumented(self, source, label)
    }
}