dspk_view/pyramid/
builder.rs

1//! Fills a [`Pyramid`] in the background, once, for every reader of it.
2//!
3//! One thread per pyramid reads the recording a chunk at a time, nearest the window being looked
4//! at first and then outward (a new focus re-orders what is left), and reports each chunk done.
5//! Views only draw from the pyramid, so two views of one recording never read the same samples
6//! twice, and a zoomed-out view fills in as the reading advances. With the focus at sample 0 it
7//! is a plain front-to-back build (as for a file pyramid).
8
9use std::ops::Range;
10use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
11use std::sync::Arc;
12
13use dspk_core::RecordingSource;
14
15use super::Pyramid;
16
17/// Most values (channels × samples) one read holds.
18const BLOCK_VALUES: usize = 1 << 22;
19/// Seconds of recording per chunk (between progress reports and focus checks).
20const CHUNK_SEC: f64 = 2.0;
21/// No focus requested.
22const NONE: u64 = u64::MAX;
23
24/// Progress of a pyramid: the samples just built, and how many are built of all.
25#[derive(Debug, Clone, PartialEq)]
26pub struct Progress {
27    pub filled: Range<u64>,
28    pub done: u64,
29    pub total: u64,
30}
31
32pub type OnProgress = Arc<dyn Fn(Progress) + Send + Sync>;
33
34/// The background filler of one pyramid.
35pub struct PyramidBuilder {
36    focus: Arc<AtomicU64>,
37    started: AtomicBool,
38}
39
40impl Default for PyramidBuilder {
41    fn default() -> Self {
42        Self { focus: Arc::new(AtomicU64::new(NONE)), started: AtomicBool::new(false) }
43    }
44}
45
46impl PyramidBuilder {
47    /// Starts filling (once), nearest sample `focus` first; later calls only move the focus.
48    pub fn run(&self, source: Arc<dyn RecordingSource>, pyramid: Arc<Pyramid>, focus: u64, cancel: Arc<AtomicBool>, on_progress: OnProgress) {
49        self.focus.store(focus, Ordering::Relaxed);
50        if self.started.swap(true, Ordering::Relaxed) {
51            return;
52        }
53        let focus = self.focus.clone();
54        let spawned = std::thread::Builder::new().name("minmax-pyramid".into()).spawn(move || fill(source.as_ref(), &pyramid, &focus, &cancel, &*on_progress));
55        if let Err(e) = spawned {
56            tracing::warn!("could not start building the min/max pyramid: {e}");
57        }
58    }
59}
60
61/// Chunks of `total` samples, nearest `focus` first, alternating after / before it.
62fn order(total: u64, chunk: u64, focus: u64) -> Vec<Range<u64>> {
63    let n = total.div_ceil(chunk);
64    if n == 0 {
65        return Vec::new();
66    }
67    let at = (focus / chunk).min(n - 1);
68    let mut index = vec![at];
69    for d in 1..n {
70        index.extend([at + d].into_iter().filter(|&i| i < n));
71        index.extend(at.checked_sub(d));
72    }
73    index.into_iter().map(|i| i * chunk..((i + 1) * chunk).min(total)).collect()
74}
75
76fn fill(source: &dyn RecordingSource, pyramid: &Pyramid, focus: &AtomicU64, cancel: &AtomicBool, on_progress: &dyn Fn(Progress)) {
77    let info = source.info();
78    let total = info.samples;
79    let chunk = ((info.sample_rate_hz() * CHUNK_SEC) as u64).max(1);
80    let done_of = |plan: &[Range<u64>]| plan.iter().filter(|r| pyramid.covers(r.start, r.end)).map(|r| r.end - r.start).sum::<u64>();
81    let mut current = focus.load(Ordering::Relaxed);
82    let mut plan = order(total, chunk, if current == NONE { 0 } else { current });
83    loop {
84        if cancel.load(Ordering::Relaxed) {
85            return;
86        }
87        // A new focus: what is left, nearest it first
88        let f = focus.load(Ordering::Relaxed);
89        if f != current && f != NONE {
90            current = f;
91            plan = order(total, chunk, f);
92        }
93        let Some(next) = plan.iter().find(|r| !pyramid.covers(r.start, r.end)).cloned() else { return };
94        let moved = || cancel.load(Ordering::Relaxed);
95        match pyramid.fill(source, next.start, next.end, BLOCK_VALUES, moved) {
96            Ok(true) => on_progress(Progress { filled: next, done: done_of(&plan), total }),
97            Ok(false) => return,
98            Err(e) => {
99                tracing::warn!("building the min/max pyramid stopped: {e}");
100                return;
101            }
102        }
103    }
104}
105
106#[cfg(test)]
107mod tests {
108    use super::*;
109    use std::sync::Mutex;
110    use std::time::Duration;
111
112    #[test]
113    fn test_order_starts_at_the_focus_and_goes_outward() {
114        let starts: Vec<u64> = order(100, 10, 45).iter().map(|r| r.start).collect();
115        assert_eq!(starts, vec![40, 50, 30, 60, 20, 70, 10, 80, 0, 90]);
116        assert_eq!(order(25, 10, 0).last().unwrap(), &(20..25), "the last chunk is short");
117        assert_eq!(order(100, 10, 1000).first().unwrap().start, 90, "focus past the end: the end");
118    }
119
120    #[test]
121    fn test_fills_everything_once_and_reports_progress() {
122        // 4 channels, 9 s at 1 kHz
123        let data: Vec<f32> = (0..4 * 9000).map(|i| ((i * 31) % 97) as f32).collect();
124        let source: Arc<dyn RecordingSource> = Arc::new(dspk_core::MemoryRecording::new("t", data, 4, 1000.0).unwrap());
125        let pyramid = Arc::new(Pyramid::in_memory(source.as_ref(), 16).unwrap());
126        let reports = Arc::new(Mutex::new(Vec::new()));
127        let r = reports.clone();
128        let s = PyramidBuilder::default();
129        s.run(source.clone(), pyramid.clone(), 5000, Arc::new(AtomicBool::new(false)), Arc::new(move |p| r.lock().unwrap().push(p)));
130        for _ in 0..500 {
131            if reports.lock().unwrap().last().is_some_and(|p: &Progress| p.done == p.total) {
132                break;
133            }
134            std::thread::sleep(Duration::from_millis(10));
135        }
136        let reports = reports.lock().unwrap();
137        assert!(pyramid.covers(0, 9000));
138        assert_eq!(reports.first().unwrap().filled, 4000..6000, "the focus first");
139        assert_eq!(reports.last().unwrap().done, 9000);
140        assert_eq!(reports.len(), 5, "each chunk once");
141    }
142}
dspk_view/pyramid/builder.rs

dspk_view/pyramid/
builder.rs
