tenant.openagents/omega

1use std::thread::ThreadId;
2use std::time::{Duration, Instant};
3
4use collections::HashMap;
5use hdrhistogram::Histogram;
6use itertools::Itertools;
7
8use crate::STARTUP_TIME;
9
10/// Microseconds since app start
11type MicroSeconds = u64;
12
13// TODO(yara) some crazy ideas:
14// - track most recent action?
15// - Action that this task was spawned from?
16// - flag that enables tracking more for a specific task
17// - task backtrace? (who spawned who etc)
18
19#[derive(Debug, Clone, serde::Serialize)]
20struct HangReport {
21    location: String,
22    hang_density: f64,
23    mean_hang_duration: MicroSeconds,
24
25    /// since app start
26    slowest_start: MicroSeconds,
27    /// since app start
28    slowest_end: MicroSeconds,
29
30    /// 50% of bad polls where faster then this value
31    /// This also approximate the median
32    p50: MicroSeconds,
33    /// 75% of bad polls where faster then this value
34    p75: MicroSeconds,
35    /// 95% of bad polls where faster then this value
36    p95: MicroSeconds,
37}
38
39#[derive(Debug, Clone, serde::Serialize)]
40struct TelemetryReport {
41    foreground: Vec<HangReport>,
42    background: Vec<HangReport>,
43    actions: Vec<HangReport>,
44}
45
46struct Item<T: Hang> {
47    total_hanged: Duration,
48    slowest_poll: T,
49    /// saturates if more then 256 measurements end up in the same bin
50    histogram: Histogram<u8>,
51}
52
53impl<T: Hang> Item<T> {
54    fn hang_density(&self, period: Duration) -> f64 {
55        self.total_hanged.div_duration_f64(period)
56    }
57    fn into_report(&self, location: &T::Descriptor, period: Duration) -> HangReport {
58        let now = Instant::now();
59        let startup = STARTUP_TIME.get().unwrap_or(&now);
60        HangReport {
61            location: T::format_location(location),
62            hang_density: self.hang_density(period),
63            mean_hang_duration: self.histogram.mean() as u64,
64            slowest_start: self
65                .slowest_poll
66                .start()
67                .duration_since(*startup)
68                .as_micros() as u64,
69            slowest_end: self.slowest_poll.end().duration_since(*startup).as_micros() as u64,
70            p50: self.histogram.value_at_quantile(0.5),
71            p75: self.histogram.value_at_quantile(0.75),
72            p95: self.histogram.value_at_quantile(0.95),
73        }
74    }
75}
76
77struct Hangs<T: Hang> {
78    last_reset: Instant,
79    hangs: HashMap<T::Descriptor, Item<T>>,
80}
81
82trait Hang: Clone {
83    type Descriptor: std::hash::Hash + PartialEq + Eq;
84    fn poll_duration(&self) -> Duration;
85    fn descriptor(&self) -> Self::Descriptor;
86    fn format_location(location: &Self::Descriptor) -> String;
87    fn start(&self) -> Instant;
88    fn end(&self) -> Instant;
89}
90
91impl Hang for gpui::TaskTiming {
92    type Descriptor = std::panic::Location<'static>;
93
94    fn poll_duration(&self) -> Duration {
95        gpui::TaskTiming::poll_duration(self)
96    }
97    fn descriptor(&self) -> Self::Descriptor {
98        *self.location
99    }
100    fn format_location(location: &Self::Descriptor) -> String {
101        format!(
102            "{}:{}:{}",
103            location.file(),
104            location.line(),
105            location.column()
106        )
107    }
108    fn start(&self) -> Instant {
109        self.start
110    }
111    fn end(&self) -> Instant {
112        self.end.0
113    }
114}
115
116impl Hang for gpui::ActionTiming {
117    type Descriptor = &'static str;
118
119    fn poll_duration(&self) -> Duration {
120        self.duration()
121    }
122    fn descriptor(&self) -> Self::Descriptor {
123        self.name
124    }
125    fn format_location(location: &Self::Descriptor) -> String {
126        location.to_string()
127    }
128    fn start(&self) -> Instant {
129        self.start
130    }
131    fn end(&self) -> Instant {
132        self.end
133    }
134}
135
136impl<T: Hang> Hangs<T> {
137    fn new() -> Self {
138        Self {
139            last_reset: Instant::now(),
140            hangs: HashMap::default(),
141        }
142    }
143    fn add(&mut self, new: T, min_recorded_us: u64) {
144        const MICROSECONDS_MINUTE: u64 = 60 * 1000 * 1000;
145
146        if self.hangs.len() > 1000 {
147            log::warn!("Too many hanging tasks to track, can not add new");
148            return;
149        }
150
151        self.hangs
152            .entry(new.descriptor())
153            .and_modify(|item| {
154                item.total_hanged += new.poll_duration();
155                item.histogram
156                    .saturating_record(new.poll_duration().as_micros() as u64);
157                if new.poll_duration() > item.slowest_poll.poll_duration() {
158                    item.slowest_poll = new.clone();
159                }
160            })
161            .or_insert({
162                Item {
163                    total_hanged: new.poll_duration(),
164                    slowest_poll: new,
165                    histogram: Histogram::new_with_bounds(min_recorded_us, MICROSECONDS_MINUTE, 3)
166                        .expect("function parameters are constants and correct"),
167                }
168            });
169    }
170
171    fn report_and_reset(&mut self) -> Vec<HangReport> {
172        let period = self.last_reset.elapsed();
173        self.last_reset = Instant::now();
174
175        let lowest_density_to_report = self
176            .hangs
177            .values()
178            .map(|item| item.total_hanged.div_duration_f64(period))
179            .k_largest_relaxed_by(5, f64::total_cmp)
180            .nth(5)
181            .unwrap_or(0.0);
182
183        let report = self
184            .hangs
185            .drain()
186            .filter(|(_, item)| item.hang_density(period) >= lowest_density_to_report)
187            .map(|(location, item)| item.into_report(&location, period))
188            .collect();
189
190        report
191    }
192
193    fn is_empty(&self) -> bool {
194        self.hangs.is_empty()
195    }
196}
197
198pub struct Reporter {
199    record_slower_then: Duration,
200    foreground_thread: ThreadId,
201    last_send: Instant,
202
203    foreground: Hangs<gpui::TaskTiming>,
204    background: Hangs<gpui::TaskTiming>,
205    actions: Hangs<gpui::ActionTiming>,
206}
207
208impl Reporter {
209    pub fn new(foreground_thread: ThreadId) -> Self {
210        Self {
211            record_slower_then: Duration::from_millis(1),
212            foreground_thread,
213            last_send: Instant::now(),
214            foreground: Hangs::new(),
215            background: Hangs::new(),
216            actions: Hangs::new(),
217        }
218    }
219
220    pub fn update(
221        &mut self,
222        task_stats: &[gpui::ThreadTaskStatistics],
223        action_stats: &gpui::ActionStatistics,
224    ) {
225        self.process_foreground(task_stats);
226        self.process_background(task_stats);
227        self.process_actions(action_stats);
228    }
229
230    pub fn send_periodically(&mut self) {
231        // this should be a long period otherwise things like
232        // hang density get
233        if self.last_send.elapsed() > Duration::from_mins(30) {
234            self.send()
235        }
236    }
237
238    pub fn send(&mut self) {
239        self.last_send = Instant::now();
240        if self.nothing_to_report() {
241            return;
242        }
243        let report = TelemetryReport {
244            foreground: self.foreground.report_and_reset(),
245            background: self.background.report_and_reset(),
246            actions: self.actions.report_and_reset(),
247        };
248
249        telemetry::event!("Hang Report", report);
250    }
251
252    fn process_foreground(&mut self, task_stats: &[gpui::ThreadTaskStatistics]) {
253        let foreground_thread = self.foreground_thread;
254        let Some(foreground) = task_stats.iter().find(|t| t.thread_id == foreground_thread) else {
255            // during startup foreground thread might not have statistics yet
256            return;
257        };
258
259        for hang in foreground
260            .stats
261            .longest_poll_times
262            .into_iter()
263            .filter(|task| task.poll_duration() > self.record_slower_then)
264        {
265            self.foreground
266                .add(hang, self.record_slower_then.as_micros() as u64);
267        }
268    }
269    fn process_background(&mut self, task_stats: &[gpui::ThreadTaskStatistics]) {
270        let foreground_thread = self.foreground_thread;
271        let background = task_stats
272            .iter()
273            .filter(|t| t.thread_id != foreground_thread);
274
275        for worker in background {
276            for hang in worker
277                .stats
278                .longest_poll_times
279                .into_iter()
280                .filter(|task| task.poll_duration() > self.record_slower_then)
281            {
282                self.background
283                    .add(hang, self.record_slower_then.as_micros() as u64);
284            }
285        }
286    }
287
288    fn process_actions(&mut self, action_stats: &gpui::ActionStatistics) {
289        for hang in action_stats
290            .longest_runtimes(false)
291            .filter(|action| action.runtime() > self.record_slower_then)
292        {
293            self.actions
294                .add(hang, self.record_slower_then.as_micros() as u64);
295        }
296    }
297
298    fn nothing_to_report(&self) -> bool {
299        self.actions.is_empty() && self.foreground.is_empty() && self.background.is_empty()
300    }
301}
302