pito_jobs/
live.rs

1use std::collections::BTreeMap;
2use std::path::{Path, PathBuf};
3use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
4use std::sync::{Arc, Mutex};
5use std::thread::{self, JoinHandle};
6use std::time::Duration;
7
8use serde::{Deserialize, Serialize};
9
10use super::env::{MACOS, data_dir, jobs_dir};
11use super::host::{host_name, pid_start};
12use super::util::{self, Clock, atomic_write, filled};
13
14const CAP: usize = 2048;
15const ONE_US: u64 = 1_000_000;
16const FIVE_US: u64 = 5_000_000;
17const FILE_LIMIT: usize = 4096;
18
19#[derive(Clone, Copy, Debug, PartialEq, Eq)]
20pub enum Surface {
21    Tui,
22    Desk,
23    Service,
24}
25
26impl Surface {
27    fn as_str(self) -> &'static str {
28        match self {
29            Surface::Tui => "tui",
30            Surface::Desk => "desk",
31            Surface::Service => "service",
32        }
33    }
34
35    fn target_fps(self) -> f64 {
36        match self {
37            Surface::Tui => 120.0,
38            Surface::Desk => 60.0,
39            Surface::Service => 0.0,
40        }
41    }
42}
43
44#[derive(Clone, Copy, Debug, PartialEq, Eq)]
45pub enum Pacing {
46    Continuous,
47    OnChange,
48}
49
50impl Pacing {
51    fn as_str(self) -> &'static str {
52        match self {
53            Pacing::Continuous => "continuous",
54            Pacing::OnChange => "on_change",
55        }
56    }
57}
58
59#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
60#[serde(default)]
61pub struct Span {
62    pub p50: f64,
63    #[serde(default)]
64    pub p95: f64,
65    pub p99: f64,
66    pub max: f64,
67}
68
69#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
70pub struct Beat {
71    #[serde(default)]
72    pub v: u32,
73    #[serde(default)]
74    pub id: String,
75    #[serde(default)]
76    pub host: String,
77    #[serde(default)]
78    pub pid: u32,
79    #[serde(default)]
80    pub pid_start: u64,
81    #[serde(default)]
82    pub app: String,
83    #[serde(default)]
84    pub version: String,
85    #[serde(default)]
86    pub surface: String,
87    #[serde(default)]
88    pub ride: bool,
89    #[serde(default)]
90    pub fps: f64,
91    #[serde(default)]
92    pub frame_ms: Option<Span>,
93    #[serde(default)]
94    pub gpu_ms: Option<Span>,
95    #[serde(default)]
96    pub interval_ms: Option<Span>,
97    #[serde(default)]
98    pub missed: u64,
99    #[serde(default)]
100    pub target_fps: f64,
101    #[serde(default)]
102    pub pacing: String,
103    #[serde(default)]
104    pub idle: bool,
105    #[serde(default)]
106    pub detail: Option<String>,
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub size: Option<[u32; 2]>,
109    #[serde(default, skip_serializing_if = "Option::is_none")]
110    pub renderer: Option<String>,
111    #[serde(default, skip_serializing_if = "Option::is_none")]
112    pub gpu_pass_ms: Option<BTreeMap<String, f64>>,
113    #[serde(default)]
114    pub ride_set: Option<String>,
115    #[serde(default)]
116    pub started: u64,
117    #[serde(default)]
118    pub updated: u64,
119}
120
121pub fn live(app: &str, version: &str, surface: Surface, pacing: Pacing) -> Handle {
122    open(
123        app,
124        version,
125        surface,
126        pacing,
127        Env::from_env(),
128        Clock::system(),
129        true,
130    )
131}
132
133pub fn apps_dir() -> Option<PathBuf> {
134    Env::from_env().dir
135}
136
137pub fn read_apps(path: impl AsRef<Path>) -> Vec<Beat> {
138    let Ok(entries) = std::fs::read_dir(path) else {
139        return Vec::new();
140    };
141    let mut found = Vec::new();
142    for entry in entries.flatten() {
143        let name = entry.file_name();
144        let name = name.to_string_lossy();
145        if name.starts_with('.') {
146            continue;
147        }
148        let Ok(meta) = entry.metadata() else {
149            continue;
150        };
151        if !meta.is_file() {
152            continue;
153        }
154        let Some(bytes) = crate::util::read_capped(&entry.path()) else {
155            continue;
156        };
157        let Ok(beat) = serde_json::from_slice::<Beat>(&bytes) else {
158            continue;
159        };
160        if beat.app.is_empty() {
161            continue;
162        }
163        found.push((name.into_owned(), beat));
164    }
165    found.sort_by(|left, right| left.0.cmp(&right.0));
166    found.into_iter().map(|(_, beat)| beat).collect()
167}
168
169#[must_use = "dropping the handle removes the heartbeat file"]
170pub struct Handle {
171    inner: Option<Arc<Inner>>,
172    stop: Option<Arc<AtomicBool>>,
173    thread: Option<JoinHandle<()>>,
174}
175
176impl Handle {
177    pub fn frame(&self, work: Duration) {
178        self.push(false, work, None);
179    }
180
181    pub fn frame_with(&self, cpu: Duration, gpu: Option<Duration>) {
182        self.push(false, cpu, gpu);
183    }
184
185    pub fn frame_idle(&self) {
186        self.push(true, Duration::ZERO, None);
187    }
188
189    pub fn detail(&self, text: &str) {
190        let Some(inner) = &self.inner else {
191            return;
192        };
193        let line = one_line(text, 240);
194        let mut slot = inner.detail.lock().unwrap_or_else(|err| err.into_inner());
195        *slot = if line.is_empty() { None } else { Some(line) };
196    }
197
198    pub fn size(&self, w: u32, h: u32) {
199        if let Some(inner) = &self.inner {
200            *inner.size.lock().unwrap_or_else(|err| err.into_inner()) = Some([w, h]);
201        }
202    }
203
204    pub fn renderer(&self, name: &str) {
205        if let Some(inner) = &self.inner {
206            let name = short_line(name, 16);
207            *inner.renderer.lock().unwrap_or_else(|err| err.into_inner()) =
208                (!name.is_empty()).then_some(name);
209        }
210    }
211
212    pub fn passes(&self, passes: &[(&str, f64)]) {
213        if let Some(inner) = &self.inner {
214            let mut values = BTreeMap::new();
215            for &(name, ms) in passes {
216                if values.len() >= 16 {
217                    break;
218                }
219                if !ms.is_finite() || ms < 0.0 {
220                    continue;
221                }
222                let name = short_line(name, 24);
223                if !name.is_empty() {
224                    values.insert(name, ms);
225                }
226            }
227            *inner.passes.lock().unwrap_or_else(|err| err.into_inner()) = Some(values);
228        }
229    }
230
231    fn push(&self, idle: bool, cpu: Duration, gpu: Option<Duration>) {
232        let Some(inner) = &self.inner else {
233            return;
234        };
235        let seq = inner.next.fetch_add(1, Ordering::Relaxed);
236        let slot = &inner.slots[(seq as usize) % CAP];
237        slot.at.store(inner.clock.mono_us(), Ordering::Relaxed);
238        slot.cpu.store(as_ns(cpu), Ordering::Relaxed);
239        slot.gpu
240            .store(gpu.map(as_ns).unwrap_or(u64::MAX), Ordering::Relaxed);
241        slot.flags.store(u64::from(idle), Ordering::Relaxed);
242        slot.seq.store(seq + 1, Ordering::Release);
243    }
244}
245
246impl Drop for Handle {
247    fn drop(&mut self) {
248        if let Some(stop) = &self.stop {
249            stop.store(true, Ordering::Release);
250        }
251        if let Some(thread) = self.thread.take() {
252            thread.thread().unpark();
253            let _ = thread.join();
254        }
255        if let Some(inner) = &self.inner {
256            inner.remove_file();
257        }
258    }
259}
260
261struct Slot {
262    seq: AtomicU64,
263    at: AtomicU64,
264    cpu: AtomicU64,
265    gpu: AtomicU64,
266    flags: AtomicU64,
267}
268
269impl Slot {
270    fn new() -> Slot {
271        Slot {
272            seq: AtomicU64::new(0),
273            at: AtomicU64::new(0),
274            cpu: AtomicU64::new(0),
275            gpu: AtomicU64::new(0),
276            flags: AtomicU64::new(0),
277        }
278    }
279}
280
281struct Mark {
282    at_us: u64,
283    cpu_ns: u64,
284    gpu_ns: Option<u64>,
285    idle: bool,
286}
287
288struct Accum {
289    cursor: u64,
290    marks: Vec<Mark>,
291}
292
293struct Measured {
294    fps: f64,
295    idle: bool,
296    frame: Option<Span>,
297    gpu: Option<Span>,
298    interval: Option<Span>,
299    missed: u64,
300}
301
302struct Inner {
303    clock: Clock,
304    slots: Vec<Slot>,
305    next: AtomicU64,
306    accum: Mutex<Accum>,
307    detail: Mutex<Option<String>>,
308    size: Mutex<Option<[u32; 2]>>,
309    renderer: Mutex<Option<String>>,
310    passes: Mutex<Option<BTreeMap<String, f64>>>,
311    wrote: AtomicBool,
312    last_us: AtomicU64,
313    dir: PathBuf,
314    file: String,
315    app: String,
316    version: String,
317    surface: Surface,
318    pacing: Pacing,
319    target_fps: f64,
320    id: String,
321    host: String,
322    pid: u32,
323    pid_start: u64,
324    ride: bool,
325    ride_set: Option<String>,
326    started_ms: u64,
327}
328
329impl Inner {
330    fn write_if_due(&self) {
331        let now = self.clock.mono_us();
332        self.collect(now);
333        if self.wrote.load(Ordering::Relaxed) {
334            let last = self.last_us.load(Ordering::Relaxed);
335            if now.saturating_sub(last) < ONE_US {
336                return;
337            }
338        }
339        let measured = self.measured(now);
340        let detail = self
341            .detail
342            .lock()
343            .unwrap_or_else(|err| err.into_inner())
344            .clone();
345        let size = *self.size.lock().unwrap_or_else(|err| err.into_inner());
346        let renderer = self
347            .renderer
348            .lock()
349            .unwrap_or_else(|err| err.into_inner())
350            .clone();
351        let passes = self
352            .passes
353            .lock()
354            .unwrap_or_else(|err| err.into_inner())
355            .clone();
356        let beat = self.beat(&measured, detail, size, renderer, passes);
357        let bytes = encode(&beat);
358        if bytes.is_empty() {
359            return;
360        }
361        if atomic_write(&self.dir, &self.file, &bytes).is_ok() {
362            self.last_us.store(now, Ordering::Relaxed);
363            self.wrote.store(true, Ordering::Relaxed);
364        }
365    }
366
367    fn collect(&self, now: u64) {
368        let mut accum = self.accum.lock().unwrap_or_else(|err| err.into_inner());
369        drain(&self.slots, &mut accum);
370        prune(&mut accum.marks, now);
371    }
372
373    fn measured(&self, now: u64) -> Measured {
374        let accum = self.accum.lock().unwrap_or_else(|err| err.into_inner());
375        let target = match self.pacing {
376            Pacing::Continuous => self.target_fps,
377            Pacing::OnChange => 0.0,
378        };
379        measure(&accum.marks, now, target)
380    }
381
382    fn beat(
383        &self,
384        measured: &Measured,
385        detail: Option<String>,
386        size: Option<[u32; 2]>,
387        renderer: Option<String>,
388        gpu_pass_ms: Option<BTreeMap<String, f64>>,
389    ) -> Beat {
390        Beat {
391            v: 1,
392            id: self.id.clone(),
393            host: self.host.clone(),
394            pid: self.pid,
395            pid_start: self.pid_start,
396            app: self.app.clone(),
397            version: self.version.clone(),
398            surface: self.surface.as_str().to_string(),
399            ride: self.ride,
400            fps: measured.fps,
401            frame_ms: measured.frame.clone(),
402            gpu_ms: measured.gpu.clone(),
403            interval_ms: measured.interval.clone(),
404            missed: measured.missed,
405            target_fps: self.target_fps,
406            pacing: self.pacing.as_str().to_string(),
407            idle: measured.idle,
408            detail,
409            size,
410            renderer,
411            gpu_pass_ms,
412            ride_set: self.ride_set.clone(),
413            started: self.started_ms,
414            updated: self.clock.wall_ms(),
415        }
416    }
417
418    fn remove_file(&self) {
419        let _ = std::fs::remove_file(self.dir.join(&self.file));
420    }
421}
422
423fn open(
424    app: &str,
425    version: &str,
426    surface: Surface,
427    pacing: Pacing,
428    env: Env,
429    clock: Clock,
430    spawn: bool,
431) -> Handle {
432    let Some(dir) = env.dir else {
433        return Handle {
434            inner: None,
435            stop: None,
436            thread: None,
437        };
438    };
439    if env.off || !util::WRITES {
440        return Handle {
441            inner: None,
442            stop: None,
443            thread: None,
444        };
445    }
446    let started_ms = clock.wall_ms();
447    let pid = util::process_id();
448    let host = host_name();
449    let file = format!("{pid}-{started_ms}.json");
450    let inner = Arc::new(Inner {
451        clock,
452        slots: (0..CAP).map(|_| Slot::new()).collect(),
453        next: AtomicU64::new(0),
454        accum: Mutex::new(Accum {
455            cursor: 0,
456            marks: Vec::new(),
457        }),
458        detail: Mutex::new(None),
459        size: Mutex::new(None),
460        renderer: Mutex::new(None),
461        passes: Mutex::new(None),
462        wrote: AtomicBool::new(false),
463        last_us: AtomicU64::new(0),
464        dir,
465        file,
466        app: one_line(app, 128),
467        version: one_line(version, 128),
468        surface,
469        pacing,
470        target_fps: surface.target_fps(),
471        id: format!("{host}:{pid}:{started_ms}"),
472        host,
473        pid,
474        pid_start: pid_start(),
475        ride: env.ride,
476        ride_set: env.ride_set.map(|set| one_line(&set, 128)),
477        started_ms,
478    });
479    if !spawn {
480        return Handle {
481            inner: Some(inner),
482            stop: None,
483            thread: None,
484        };
485    }
486    let stop = Arc::new(AtomicBool::new(false));
487    let thread = {
488        let inner = Arc::clone(&inner);
489        let stop = Arc::clone(&stop);
490        thread::Builder::new()
491            .name("pito-live".to_string())
492            .spawn(move || worker(inner, stop))
493            .ok()
494    };
495    Handle {
496        inner: Some(inner),
497        stop: Some(stop),
498        thread,
499    }
500}
501
502fn worker(inner: Arc<Inner>, stop: Arc<AtomicBool>) {
503    be_nice();
504    while !stop.load(Ordering::Acquire) {
505        inner.write_if_due();
506        if stop.load(Ordering::Acquire) {
507            break;
508        }
509        thread::park_timeout(Duration::from_millis(200));
510    }
511    inner.remove_file();
512}
513
514#[cfg(target_os = "linux")]
515fn be_nice() {
516    unsafe {
517        let tid = libc::gettid();
518        libc::setpriority(libc::PRIO_PROCESS, tid as libc::id_t, 19);
519        libc::syscall(libc::SYS_ioprio_set, 1, 0, 3 << 13);
520    }
521}
522
523#[cfg(not(target_os = "linux"))]
524fn be_nice() {}
525
526fn drain(slots: &[Slot], accum: &mut Accum) {
527    for _ in 0..CAP.saturating_mul(2) {
528        let slot = &slots[(accum.cursor as usize) % CAP];
529        let seq = slot.seq.load(Ordering::Acquire);
530        let expect = accum.cursor + 1;
531        if seq == expect {
532            let at = slot.at.load(Ordering::Relaxed);
533            let cpu = slot.cpu.load(Ordering::Relaxed);
534            let gpu = slot.gpu.load(Ordering::Relaxed);
535            let flags = slot.flags.load(Ordering::Relaxed);
536            if slot.seq.load(Ordering::Acquire) != expect {
537                break;
538            }
539            accum.marks.push(Mark {
540                at_us: at,
541                cpu_ns: cpu,
542                gpu_ns: (gpu != u64::MAX).then_some(gpu),
543                idle: flags == 1,
544            });
545            accum.cursor += 1;
546            continue;
547        }
548        if seq > expect {
549            accum.cursor = seq - 1;
550            continue;
551        }
552        break;
553    }
554}
555
556fn prune(marks: &mut Vec<Mark>, now: u64) {
557    if let Some(index) = marks
558        .iter()
559        .position(|mark| now.saturating_sub(mark.at_us) <= FIVE_US)
560    {
561        let drop_n = index.saturating_sub(1);
562        if drop_n > 0 {
563            marks.drain(..drop_n);
564        }
565    } else if marks.len() > 1 {
566        let last = marks.pop().unwrap();
567        marks.clear();
568        marks.push(last);
569    }
570    if marks.len() > 8192 {
571        let drop_n = marks.len() - 8192;
572        marks.drain(..drop_n);
573    }
574}
575
576fn measure(marks: &[Mark], now: u64, target_fps: f64) -> Measured {
577    let mut frames = Vec::new();
578    let mut gpus = Vec::new();
579    let mut gaps = Vec::new();
580    let mut fps = 0u64;
581    let mut missed = 0u64;
582    let mut prev: Option<u64> = None;
583    let limit = if target_fps > 0.0 {
584        1.5 * (ONE_US as f64 / target_fps)
585    } else {
586        0.0
587    };
588    for mark in marks {
589        if mark.idle {
590            continue;
591        }
592        let age = now.saturating_sub(mark.at_us);
593        let in_five = age <= FIVE_US;
594        if age <= ONE_US {
595            fps += 1;
596        }
597        if let Some(before) = prev
598            && in_five
599        {
600            let gap = mark.at_us.saturating_sub(before);
601            gaps.push(gap as f64 / 1000.0);
602            if limit > 0.0 && gap as f64 > limit {
603                missed += 1;
604            }
605        }
606        prev = Some(mark.at_us);
607        if in_five {
608            frames.push(mark.cpu_ns as f64 / 1_000_000.0);
609            if let Some(gpu) = mark.gpu_ns {
610                gpus.push(gpu as f64 / 1_000_000.0);
611            }
612        }
613    }
614    Measured {
615        fps: fps as f64,
616        idle: fps == 0,
617        frame: span_of(&frames),
618        gpu: span_of(&gpus),
619        interval: span_of(&gaps),
620        missed,
621    }
622}
623
624fn span_of(values: &[f64]) -> Option<Span> {
625    if values.is_empty() {
626        return None;
627    }
628    let mut sorted = values.to_vec();
629    sorted.sort_by(|left, right| left.partial_cmp(right).unwrap_or(std::cmp::Ordering::Equal));
630    Some(Span {
631        p50: round1(percentile(&sorted, 0.50)),
632        p95: round1(percentile(&sorted, 0.95)),
633        p99: round1(percentile(&sorted, 0.99)),
634        max: round1(sorted[sorted.len() - 1]),
635    })
636}
637
638fn percentile(sorted: &[f64], p: f64) -> f64 {
639    let n = sorted.len();
640    let idx = ((p * n as f64).ceil() as usize)
641        .saturating_sub(1)
642        .min(n - 1);
643    sorted[idx]
644}
645
646fn round1(value: f64) -> f64 {
647    (value * 10.0).round() / 10.0
648}
649
650fn encode(beat: &Beat) -> Vec<u8> {
651    let Ok(bytes) = serde_json::to_vec(beat) else {
652        return Vec::new();
653    };
654    if bytes.len() <= FILE_LIMIT {
655        return bytes;
656    }
657    let mut copy = beat.clone();
658    copy.detail = None;
659    match serde_json::to_vec(&copy) {
660        Ok(bytes) if bytes.len() <= FILE_LIMIT => bytes,
661        _ => Vec::new(),
662    }
663}
664
665fn as_ns(duration: Duration) -> u64 {
666    u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX)
667}
668
669fn one_line(text: &str, max: usize) -> String {
670    let text = text.split(['\n', '\r']).next().unwrap_or("");
671    let mut end = text.len().min(max);
672    while end > 0 && !text.is_char_boundary(end) {
673        end -= 1;
674    }
675    text[..end].to_string()
676}
677
678fn short_line(text: &str, max: usize) -> String {
679    text.chars()
680        .take_while(|ch| *ch != '\n' && *ch != '\r')
681        .filter(|ch| !ch.is_control())
682        .take(max)
683        .collect()
684}
685
686#[derive(Clone, Debug)]
687struct Env {
688    dir: Option<PathBuf>,
689    off: bool,
690    ride: bool,
691    ride_set: Option<String>,
692}
693
694impl Env {
695    fn from_env() -> Env {
696        Self::from_fn(|key| std::env::var(key).ok(), |path| path.is_dir())
697    }
698
699    fn from_fn(mut get: impl FnMut(&str) -> Option<String>, exists: impl Fn(&Path) -> bool) -> Env {
700        let asked_off = get("PITO_JOBS")
701            .as_deref()
702            .is_some_and(|value| value.trim() == "off");
703        let data = data_dir(&mut get, MACOS);
704        let installed = filled(get("PITO_JOBS_DIR")).is_some()
705            || data.is_some_and(|data| exists(&data.join("pito").join("apps")));
706        let off = asked_off || !installed;
707        let ride_app = filled(get("RIDE_APP"));
708        let ride_set = filled(get("RIDE_SET"));
709        let dir = jobs_dir(&mut get, MACOS).map(|dir| dir.join("apps"));
710        Env {
711            dir,
712            off,
713            ride: ride_app.is_some(),
714            ride_set,
715        }
716    }
717}
718
719#[cfg(test)]
720impl Handle {
721    fn drive(&self) {
722        if let Some(inner) = &self.inner {
723            inner.write_if_due();
724        }
725    }
726}
727
728#[cfg(test)]
729mod tests {
730    use super::*;
731
732    #[test]
733    fn read_apps_skips_a_file_over_four_kilobytes_and_keeps_one_at_the_cap() {
734        let dir = scratch("apps-cap");
735        let kept = capped_beat(crate::util::READ_CAP);
736        let mut over = kept.clone();
737        over.push(b' ');
738        std::fs::write(dir.join("kept.json"), &kept).unwrap();
739        std::fs::write(dir.join("over.json"), &over).unwrap();
740        let beats = read_apps(&dir);
741        assert_eq!(beats.len(), 1);
742        assert_eq!(beats[0].app, "paper");
743        assert_eq!(kept.len(), crate::util::READ_CAP);
744    }
745
746    fn capped_beat(len: usize) -> Vec<u8> {
747        let prefix = br#"{"app":"paper","id":""#;
748        let suffix = br#""}"#;
749        let pad = len - prefix.len() - suffix.len();
750        let mut bytes = Vec::with_capacity(len);
751        bytes.extend_from_slice(prefix);
752        bytes.extend(std::iter::repeat_n(b'a', pad));
753        bytes.extend_from_slice(suffix);
754        assert_eq!(bytes.len(), len);
755        bytes
756    }
757
758    fn scratch(name: &str) -> PathBuf {
759        let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
760            .join("tmp")
761            .join(name);
762        let _ = std::fs::remove_dir_all(&path);
763        std::fs::create_dir_all(&path).unwrap();
764        path
765    }
766
767    fn env_at(dir: &Path) -> Env {
768        Env {
769            dir: Some(dir.to_path_buf()),
770            off: false,
771            ride: false,
772            ride_set: None,
773        }
774    }
775
776    fn mark(at_ms: u64, cpu_ms: f64, gpu_ms: Option<f64>, idle: bool) -> Mark {
777        Mark {
778            at_us: at_ms * 1000,
779            cpu_ns: (cpu_ms * 1_000_000.0).round() as u64,
780            gpu_ns: gpu_ms.map(|ms| (ms * 1_000_000.0).round() as u64),
781            idle,
782        }
783    }
784
785    #[test]
786    fn percentiles_use_the_nearest_rank_over_the_sorted_window() {
787        let values: Vec<f64> = (1..=100).map(|n| n as f64).collect();
788        assert_eq!(percentile(&values, 0.50), 50.0);
789        assert_eq!(percentile(&values, 0.95), 95.0);
790        assert_eq!(percentile(&values, 0.99), 99.0);
791        assert_eq!(percentile(&values, 0.50), span_of(&values).unwrap().p50);
792        assert_eq!(span_of(&values).unwrap().p95, 95.0);
793        let one = span_of(&[4.25]).unwrap();
794        assert_eq!(one.p50, 4.3);
795        assert_eq!(one.p95, 4.3);
796        assert_eq!(one.p99, 4.3);
797        assert_eq!(one.max, 4.3);
798        assert!(span_of(&[]).is_none());
799    }
800
801    #[test]
802    fn fps_is_the_last_second_and_percentiles_the_last_five() {
803        let mut marks = Vec::new();
804        for i in 0..100 {
805            let at = 5_000 + i * 50;
806            let gpu = (i % 2 == 0).then_some(i as f64 + 0.5);
807            marks.push(mark(at, (i + 1) as f64, gpu, false));
808        }
809        marks.push(mark(9_900, 0.1, None, true));
810        let now = 10_000_000u64;
811        let got = measure(&marks, now, 120.0);
812        assert_eq!(got.fps, 20.0);
813        assert!(!got.idle);
814        let frame = got.frame.unwrap();
815        assert_eq!(frame.p50, 50.0);
816        assert_eq!(frame.p95, 95.0);
817        assert_eq!(frame.p99, 99.0);
818        assert_eq!(frame.max, 100.0);
819        let gpu = got.gpu.unwrap();
820        assert_eq!(gpu.max, 98.5);
821        assert_eq!(gpu.p95, 94.5);
822        let interval = got.interval.unwrap();
823        assert_eq!(interval.p50, 50.0);
824        assert_eq!(interval.p95, 50.0);
825        assert_eq!(interval.max, 50.0);
826        assert_eq!(got.missed, 99);
827    }
828
829    #[test]
830    fn idle_ticks_do_not_count_and_a_short_gap_is_not_missed() {
831        let marks = vec![
832            mark(0, 1.0, None, false),
833            mark(8, 1.0, None, true),
834            mark(16, 2.0, Some(3.0), false),
835        ];
836        let got = measure(&marks, 20_000, 60.0);
837        assert_eq!(got.fps, 2.0);
838        assert!(!got.idle);
839        assert_eq!(got.interval.unwrap().p50, 16.0);
840        assert_eq!(got.missed, 0);
841        assert_eq!(got.gpu.unwrap().p99, 3.0);
842        let quiet = measure(&[mark(0, 1.0, None, true)], 2_000_000, 120.0);
843        assert_eq!(quiet.fps, 0.0);
844        assert!(quiet.idle);
845        assert!(quiet.frame.is_none());
846        assert_eq!(quiet.missed, 0);
847        let no_target = measure(&marks, 20_000, 0.0);
848        assert_eq!(no_target.missed, 0);
849    }
850
851    #[test]
852    fn an_old_anchor_still_opens_the_first_interval() {
853        let marks = vec![mark(0, 9.0, None, false), mark(6_000, 1.0, None, false)];
854        let got = measure(&marks, 6_000_000, 120.0);
855        assert_eq!(got.fps, 1.0);
856        assert_eq!(got.frame.unwrap().p50, 1.0);
857        assert_eq!(got.interval.unwrap().max, 6000.0);
858        assert_eq!(got.missed, 1);
859    }
860
861    #[test]
862    fn the_folder_follows_the_environment_and_off_clears_nothing_about_it() {
863        let off = Env::from_fn(
864            |key| match key {
865                "PITO_JOBS" => Some(" off ".to_string()),
866                "PITO_JOBS_DIR" => Some("/custom".to_string()),
867                "RIDE_APP" => Some(" paper ".to_string()),
868                "RIDE_SET" => Some(" desk-a ".to_string()),
869                _ => None,
870            },
871            |_| true,
872        );
873        assert!(off.off);
874        assert_eq!(off.dir.unwrap(), PathBuf::from("/custom/apps"));
875        assert!(off.ride);
876        assert_eq!(off.ride_set.as_deref(), Some("desk-a"));
877        let xdg = Env::from_fn(
878            |key| match key {
879                "XDG_RUNTIME_DIR" => Some("/run/user/1000".to_string()),
880                "HOME" => Some("/u".to_string()),
881                "RIDE_APP" => Some("  ".to_string()),
882                _ => None,
883            },
884            |path| {
885                let data = if MACOS {
886                    "/u/Library/Application Support"
887                } else {
888                    "/u/.local/share"
889                };
890                path == Path::new(data).join("pito").join("apps")
891            },
892        );
893        assert!(!xdg.off);
894        assert!(!xdg.ride);
895        assert_eq!(
896            xdg.dir.unwrap(),
897            PathBuf::from("/run/user/1000/pito-jobs/apps")
898        );
899        let missing = Env::from_fn(|_| None, |_| false);
900        assert!(missing.dir.is_none());
901    }
902
903    #[test]
904    fn a_buyers_machine_gets_an_inert_handle() {
905        let runtime = |key: &str| match key {
906            "XDG_RUNTIME_DIR" => Some("/run/user/1000".to_string()),
907            "HOME" => Some("/buyer".to_string()),
908            _ => None,
909        };
910        assert!(Env::from_fn(runtime, |_| false).off);
911        let data_home = Env::from_fn(
912            |key| match key {
913                "XDG_RUNTIME_DIR" => Some("/run/user/1000".to_string()),
914                "XDG_DATA_HOME" => Some("/data".to_string()),
915                _ => None,
916            },
917            |path| path == Path::new("/data/pito/apps"),
918        );
919        assert!(!data_home.off);
920        let pinned = Env::from_fn(
921            |key| match key {
922                "PITO_JOBS_DIR" => Some("/tmp-under-test".to_string()),
923                _ => None,
924            },
925            |_| false,
926        );
927        assert!(!pinned.off);
928    }
929
930    #[test]
931    fn writes_wait_a_second_and_off_writes_nothing() {
932        let dir = scratch("live-throttle");
933        let (clock, us) = Clock::manual(10_000_000);
934        let handle = open(
935            "paper",
936            "0.12.0",
937            Surface::Tui,
938            Pacing::OnChange,
939            env_at(&dir),
940            clock,
941            false,
942        );
943        handle.frame(Duration::from_millis(2));
944        handle.drive();
945        let first = read_apps(&dir);
946        assert_eq!(first.len(), 1);
947        assert_eq!(first[0].fps, 1.0);
948        assert_eq!(first[0].updated, 10_000);
949        us.store(10_000_000 + 999_000, Ordering::Relaxed);
950        handle.frame(Duration::from_millis(2));
951        handle.drive();
952        assert_eq!(read_apps(&dir)[0].fps, 1.0);
953        us.store(10_000_000 + 1_000_000, Ordering::Relaxed);
954        handle.drive();
955        let second = read_apps(&dir);
956        assert_eq!(second[0].fps, 2.0);
957        assert_eq!(second[0].updated, 11_000);
958        drop(handle);
959
960        let quiet = scratch("live-off");
961        let (clock, _) = Clock::manual(1_000);
962        let mut cfg = env_at(&quiet);
963        cfg.off = true;
964        let handle = open(
965            "paper",
966            "0.1.0",
967            Surface::Desk,
968            Pacing::Continuous,
969            cfg,
970            clock,
971            false,
972        );
973        handle.frame(Duration::from_millis(1));
974        handle.frame_idle();
975        handle.detail("nope");
976        handle.size(3840, 2160);
977        handle.renderer("live");
978        handle.passes(&[("shadow", 1.4)]);
979        handle.drive();
980        assert!(read_apps(&quiet).is_empty());
981    }
982
983    #[test]
984    fn the_file_round_trips_ride_detail_and_drop_removes_it() {
985        let dir = scratch("live-round");
986        let (clock, us) = Clock::manual(2_000_000);
987        let mut cfg = env_at(&dir);
988        cfg.ride = true;
989        cfg.ride_set = Some("desk-a".to_string());
990        let handle = open(
991            "studio",
992            "0.4.0",
993            Surface::Desk,
994            Pacing::Continuous,
995            cfg,
996            clock,
997            false,
998        );
999        for i in 0..5 {
1000            us.store(2_000_000 + i * 16_000, Ordering::Relaxed);
1001            handle.frame_with(
1002                Duration::from_micros(1_500),
1003                Some(Duration::from_micros(4_000)),
1004            );
1005        }
1006        handle.frame_idle();
1007        handle.detail("converging 128/512 spp\nsecond");
1008        handle.size(3840, 2160);
1009        handle.renderer("live\nignored");
1010        handle.passes(&[("shadow", 1.4), ("opaque", 3.9)]);
1011        handle.drive();
1012        let beat = read_apps(&dir).pop().unwrap();
1013        assert_eq!(beat.v, 1);
1014        assert_eq!(beat.app, "studio");
1015        assert_eq!(beat.version, "0.4.0");
1016        assert_eq!(beat.surface, "desk");
1017        assert_eq!(beat.pacing, "continuous");
1018        assert!(beat.ride);
1019        assert_eq!(beat.ride_set.as_deref(), Some("desk-a"));
1020        assert_eq!(beat.detail.as_deref(), Some("converging 128/512 spp"));
1021        assert_eq!(beat.size, Some([3840, 2160]));
1022        assert_eq!(beat.renderer.as_deref(), Some("live"));
1023        assert_eq!(beat.gpu_pass_ms.as_ref().unwrap()["shadow"], 1.4);
1024        assert_eq!(beat.fps, 5.0);
1025        assert!(!beat.idle);
1026        assert_eq!(beat.target_fps, 60.0);
1027        assert_eq!(beat.frame_ms.unwrap().p50, 1.5);
1028        assert_eq!(beat.gpu_ms.unwrap().p99, 4.0);
1029        assert_eq!(beat.interval_ms.unwrap().p50, 16.0);
1030        assert_eq!(beat.missed, 0);
1031        assert!(beat.id.contains(&beat.pid.to_string()));
1032        let raw = std::fs::read_dir(&dir)
1033            .unwrap()
1034            .flatten()
1035            .find(|entry| entry.file_name().to_string_lossy().ends_with(".json"))
1036            .unwrap();
1037        assert!(
1038            raw.file_name()
1039                .to_string_lossy()
1040                .starts_with(&format!("{}-", beat.pid))
1041        );
1042        drop(handle);
1043        assert!(read_apps(&dir).is_empty());
1044    }
1045
1046    #[test]
1047    fn passes_drop_invalid_values_and_cap_names_and_count() {
1048        let dir = scratch("live-passes");
1049        let (clock, us) = Clock::manual(3_000_000);
1050        let handle = open(
1051            "studio",
1052            "0.4.0",
1053            Surface::Desk,
1054            Pacing::Continuous,
1055            env_at(&dir),
1056            clock,
1057            false,
1058        );
1059        let names: Vec<_> = (0..17).map(|i| format!("pass{i:02}")).collect();
1060        let mut input: Vec<_> = names.iter().map(|name| (name.as_str(), 1.0)).collect();
1061        let long = "x".repeat(40);
1062        input.insert(0, (long.as_str(), 2.0));
1063        input.insert(1, ("nan", f64::NAN));
1064        input.insert(2, ("negative", -1.0));
1065        handle.passes(&input);
1066        handle.drive();
1067        let beat = read_apps(&dir).pop().unwrap();
1068        let passes = beat.gpu_pass_ms.unwrap();
1069        assert_eq!(passes.len(), 16);
1070        assert_eq!(passes[&"x".repeat(24)], 2.0);
1071        assert!(!passes.contains_key("nan"));
1072        assert!(!passes.contains_key("negative"));
1073        assert!(passes.contains_key("pass14"));
1074        assert!(!passes.contains_key("pass15"));
1075        handle.renderer(&"é".repeat(20));
1076        handle.passes(&[(&"é".repeat(30), 0.0)]);
1077        us.store(4_000_000, Ordering::Relaxed);
1078        handle.drive();
1079        let beat = read_apps(&dir).pop().unwrap();
1080        assert_eq!(beat.renderer.unwrap().chars().count(), 16);
1081        assert_eq!(
1082            beat.gpu_pass_ms
1083                .unwrap()
1084                .keys()
1085                .next()
1086                .unwrap()
1087                .chars()
1088                .count(),
1089            24
1090        );
1091        drop(handle);
1092    }
1093
1094    #[test]
1095    fn a_wall_clock_step_back_does_not_stop_the_heartbeat() {
1096        let dir = scratch("live-wall-step");
1097        let (clock, wall, mono) = Clock::split(10_000_000_000);
1098        let handle = open(
1099            "paper",
1100            "0.12.0",
1101            Surface::Tui,
1102            Pacing::OnChange,
1103            env_at(&dir),
1104            clock,
1105            false,
1106        );
1107        handle.frame(Duration::from_millis(2));
1108        handle.drive();
1109        assert_eq!(read_apps(&dir)[0].updated, 10_000_000);
1110        wall.store(5_000_000_000, Ordering::Relaxed);
1111        mono.store(10_000_000_000 + 1_000_000, Ordering::Relaxed);
1112        handle.frame(Duration::from_millis(2));
1113        handle.drive();
1114        let beat = &read_apps(&dir)[0];
1115        assert_eq!(beat.updated, 5_000_000);
1116        assert_eq!(beat.fps, 2.0);
1117        assert_eq!(beat.started, 10_000_000);
1118        drop(handle);
1119    }
1120
1121    #[test]
1122    fn every_maximum_heartbeat_field_fits_the_file_limit() {
1123        let dir = scratch("live-limit");
1124        let (clock, _) = Clock::manual(u64::MAX - 1_000_000);
1125        let mut cfg = env_at(&dir);
1126        cfg.ride = true;
1127        cfg.ride_set = Some("s".repeat(128));
1128        let handle = open(
1129            &"a".repeat(128),
1130            &"v".repeat(128),
1131            Surface::Desk,
1132            Pacing::Continuous,
1133            cfg,
1134            clock,
1135            false,
1136        );
1137        handle.detail(&"d".repeat(240));
1138        handle.size(u32::MAX, u32::MAX);
1139        handle.renderer(&"r".repeat(16));
1140        let names: Vec<_> = (0..16)
1141            .map(|i| format!("{i:02}{}", "p".repeat(22)))
1142            .collect();
1143        let passes: Vec<_> = names.iter().map(|name| (name.as_str(), f64::MAX)).collect();
1144        handle.passes(&passes);
1145        handle.drive();
1146        let path = dir.join(&handle.inner.as_ref().unwrap().file);
1147        let bytes = std::fs::read(path).unwrap();
1148        assert!(bytes.len() <= FILE_LIMIT, "{} bytes", bytes.len());
1149        let mut beat: Beat = serde_json::from_slice(&bytes).unwrap();
1150        assert_eq!(beat.detail.as_ref().unwrap().len(), 240);
1151        assert_eq!(beat.gpu_pass_ms.as_ref().unwrap().len(), 16);
1152        beat.host = "h".repeat(64);
1153        beat.id = format!("{}:{}:{}", beat.host, u32::MAX, u64::MAX);
1154        beat.pid = u32::MAX;
1155        beat.pid_start = u64::MAX;
1156        beat.fps = f64::MAX;
1157        let span = Span {
1158            p50: f64::MAX,
1159            p95: f64::MAX,
1160            p99: f64::MAX,
1161            max: f64::MAX,
1162        };
1163        beat.frame_ms = Some(span.clone());
1164        beat.gpu_ms = Some(span.clone());
1165        beat.interval_ms = Some(span);
1166        beat.missed = u64::MAX;
1167        beat.target_fps = f64::MAX;
1168        beat.started = u64::MAX;
1169        beat.updated = u64::MAX;
1170        let full = encode(&beat);
1171        assert!(!full.is_empty());
1172        assert!(full.len() <= FILE_LIMIT, "{} bytes", full.len());
1173        let decoded: Beat = serde_json::from_slice(&full).unwrap();
1174        assert!(decoded.detail.is_some());
1175        assert_eq!(decoded.gpu_pass_ms.unwrap().len(), 16);
1176        drop(handle);
1177    }
1178
1179    #[test]
1180    fn old_beat_without_new_fields_still_parses() {
1181        let old = r#"{"v":1,"app":"studio","frame_ms":{"p50":1.9,"p99":6.8,"max":9.1},"gpu_ms":{"p50":7.8,"p99":8.3,"max":8.9},"interval_ms":{"p50":8.3,"p99":9.4,"max":17.2}}"#;
1182        let beat: Beat = serde_json::from_str(old).unwrap();
1183        assert_eq!(beat.frame_ms.unwrap().p95, 0.0);
1184        assert_eq!(beat.gpu_ms.unwrap().p95, 0.0);
1185        assert_eq!(beat.interval_ms.unwrap().p95, 0.0);
1186        assert_eq!(beat.size, None);
1187        assert_eq!(beat.renderer, None);
1188        assert_eq!(beat.gpu_pass_ms, None);
1189    }
1190
1191    #[test]
1192    fn an_on_change_surface_misses_nothing_while_it_idles() {
1193        let dir = scratch("live-on-change");
1194        let (clock, us) = Clock::manual(10_000_000);
1195        let handle = open(
1196            "paper",
1197            "0.12.0",
1198            Surface::Tui,
1199            Pacing::OnChange,
1200            env_at(&dir),
1201            clock,
1202            false,
1203        );
1204        for _ in 0..4 {
1205            us.fetch_add(250_000, Ordering::Relaxed);
1206            handle.frame(Duration::from_micros(2_000));
1207        }
1208        handle.drive();
1209        let beat = read_apps(&dir).pop().unwrap();
1210        assert_eq!(beat.pacing, "on_change");
1211        assert_eq!(beat.fps, 4.0);
1212        assert_eq!(beat.missed, 0);
1213        drop(handle);
1214    }
1215
1216    #[test]
1217    fn a_service_writes_without_frames_and_garbage_is_skipped() {
1218        let dir = scratch("live-service");
1219        let (clock, _) = Clock::manual(4_000_000);
1220        let handle = open(
1221            "dispatch",
1222            "1.0.0",
1223            Surface::Service,
1224            Pacing::OnChange,
1225            env_at(&dir),
1226            clock,
1227            false,
1228        );
1229        handle.drive();
1230        let beat = read_apps(&dir).pop().unwrap();
1231        assert_eq!(beat.app, "dispatch");
1232        assert_eq!(beat.surface, "service");
1233        assert_eq!(beat.fps, 0.0);
1234        assert!(beat.idle);
1235        assert!(beat.frame_ms.is_none());
1236        assert_eq!(beat.target_fps, 0.0);
1237        assert_eq!(beat.pacing, "on_change");
1238        std::fs::write(dir.join(".hidden.json"), b"{\"app\":\"nope\"}").unwrap();
1239        std::fs::write(dir.join("notes.txt"), b"hello").unwrap();
1240        std::fs::write(dir.join("extra.json"), b"{\"extra\":true,\"pid\":3}").unwrap();
1241        let beats = read_apps(&dir);
1242        assert_eq!(beats.len(), 1);
1243        assert_eq!(beats[0].app, "dispatch");
1244        drop(handle);
1245    }
1246
1247    #[test]
1248    fn the_writer_thread_publishes_and_drop_removes_the_file() {
1249        let dir = scratch("live-thread");
1250        let handle = open(
1251            "paper",
1252            "0.1.0",
1253            Surface::Tui,
1254            Pacing::OnChange,
1255            env_at(&dir),
1256            Clock::system(),
1257            true,
1258        );
1259        handle.frame(Duration::from_millis(1));
1260        let mut appeared = false;
1261        for _ in 0..40 {
1262            if !read_apps(&dir).is_empty() {
1263                appeared = true;
1264                break;
1265            }
1266            thread::sleep(Duration::from_millis(50));
1267        }
1268        assert!(appeared, "heartbeat file did not appear");
1269        drop(handle);
1270        assert!(read_apps(&dir).is_empty());
1271    }
1272}
pito_jobs/live.rs

pito_jobs/
live.rs
