hotpath 0.17.0

One profiler for CPU, time, memory, 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
124
125

use crossbeam_channel::Sender;
use std::sync::{Arc, Mutex, Weak};

pub(crate) const BATCH_SIZE: usize = 64;
pub(crate) const FLUSH_INTERVAL_MS: u64 = 50;
pub(crate) const FLUSH_INTERVAL_NS: u64 = FLUSH_INTERVAL_MS * 1_000_000;

pub(crate) trait BatchedMeasurement: Sized + Send {
    fn elapsed_since_start_ns(&self) -> u64;
    fn fetch_sender() -> Option<Sender<Vec<Self>>>;

    /// Lifecycle events that must reach the worker before the data events they
    /// gate (e.g. `Created`) force an immediate flush so per-thread batching
    /// can't deliver a data event ahead of the entry that establishes its slot.
    fn is_flush_boundary(&self) -> bool {
        false
    }
}

pub(crate) struct MeasurementBatch<M: BatchedMeasurement> {
    measurements: Vec<M>,
    last_flush_elapsed_ns: u64,
    sender: Option<Sender<Vec<M>>>,
}

impl<M: BatchedMeasurement> MeasurementBatch<M> {
    pub(crate) fn new() -> Self {
        Self {
            measurements: Vec::with_capacity(BATCH_SIZE),
            last_flush_elapsed_ns: 0,
            sender: None,
        }
    }

    pub(crate) fn add(&mut self, measurement: M) {
        if self.sender.is_none() {
            self.sender = M::fetch_sender();
        }

        if self.sender.is_none() {
            return;
        }

        let elapsed_since_start_ns = measurement.elapsed_since_start_ns();
        let is_boundary = measurement.is_flush_boundary();
        self.measurements.push(measurement);

        let should_flush = is_boundary
            || self.measurements.len() >= BATCH_SIZE
            || elapsed_since_start_ns.saturating_sub(self.last_flush_elapsed_ns)
                >= FLUSH_INTERVAL_NS;

        if should_flush {
            self.flush();
        }
    }

    pub(crate) fn flush(&mut self) {
        if self.measurements.is_empty() {
            return;
        }

        let sender = self.sender.as_ref().expect("Sender must exist");
        if let Some(last) = self.measurements.last() {
            self.last_flush_elapsed_ns = last.elapsed_since_start_ns();
        }
        let batch = std::mem::replace(&mut self.measurements, Vec::with_capacity(BATCH_SIZE));
        let _ = sender.send(batch);
    }
}

impl<M: BatchedMeasurement> Drop for MeasurementBatch<M> {
    fn drop(&mut self) {
        self.flush();
    }
}

/// Registry of every live per-thread [`MeasurementBatch`] for a given event type.
///
/// Producing threads keep their batch in thread-local storage for a lock-light
/// hot path, but those batches are unreachable from other threads. On shutdown
/// the producing threads may still be alive (e.g. parked async runtime workers)
/// with buffered events that have not reached the worker. The registry holds a
/// `Weak` to each batch so [`BatchRegistry::flush_all`] can drain them all from
/// the shutting-down thread before the worker stops.
pub(crate) struct BatchRegistry<M: BatchedMeasurement> {
    batches: Mutex<Vec<Weak<Mutex<MeasurementBatch<M>>>>>,
}

impl<M: BatchedMeasurement> BatchRegistry<M> {
    pub(crate) const fn new() -> Self {
        Self {
            batches: Mutex::new(Vec::new()),
        }
    }

    fn register(&self, batch: &Arc<Mutex<MeasurementBatch<M>>>) {
        if let Ok(mut batches) = self.batches.lock() {
            batches.push(Arc::downgrade(batch));
        }
    }

    pub(crate) fn flush_all(&self) {
        if let Ok(mut batches) = self.batches.lock() {
            batches.retain(|weak| match weak.upgrade() {
                Some(batch) => {
                    if let Ok(mut batch) = batch.lock() {
                        batch.flush();
                    }
                    true
                }
                None => false,
            });
        }
    }
}

/// Creates a fresh per-thread batch and registers it for shutdown draining.
pub(crate) fn register_thread_batch<M: BatchedMeasurement>(
    registry: &'static BatchRegistry<M>,
) -> Arc<Mutex<MeasurementBatch<M>>> {
    let batch = Arc::new(Mutex::new(MeasurementBatch::new()));
    registry.register(&batch);
    batch
}