sva_engine/render/
drive.rs

1// Concern: pulls or skips a value graph a grid block per thread at a time until its range ends or `until` holds | Non-concern: what a value computes | IO: (ValueGraph, range) -> blocks
2
3use std::num::NonZeroUsize;
4
5use sva_samples::{Buffer, Extent};
6
7use super::RenderConfig;
8use super::until::{Known, Until};
9use super::value_graph::{Pulled, ValueGraph, blocks_end};
10use crate::cache::{Gauge, Memory, Progress, Recording};
11use crate::error::EngineError;
12use crate::query::{DEFAULT_FRAME_SECS, Representation};
13
14/// Nothing memory answered counts as computed.
15#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
16pub struct Work {
17    pub samples: u64,
18    pub computed_samples: u64,
19}
20
21/// Hears where a call's work stands inside a pull; with none, no cached sample is counted.
22pub(super) type Hears<'h> = Option<&'h dyn Fn(Progress)>;
23
24pub(super) struct Driver {
25    pub(super) value_graph: ValueGraph,
26    pub(super) start: i64,
27    pub(super) at: i64,
28    last: i64,
29    block: usize,
30    threads: NonZeroUsize,
31    until: Option<Until>,
32    frame: usize,
33    stop: Option<i64>,
34    end: Option<i64>,
35    /// Root samples held behind `at`, for `until` to read its open frame.
36    keep: i64,
37    /// Where the samples heard without a break since start from: a skip starts them anew.
38    heard: i64,
39    output: bool,
40    most_bytes: usize,
41    samples: u64,
42    gauge: Gauge,
43    pub(super) memory: Memory,
44    pub(super) recording: Recording,
45}
46
47/// The `envelope` reading's own frame at the render's rate, where one is asked.
48pub(super) fn frame(config: &RenderConfig, rate: u32) -> usize {
49    let secs = config
50        .asks
51        .iter()
52        .find_map(|ask| match ask.representation {
53            Representation::Envelope { frame_secs } => Some(frame_secs),
54            _ => None,
55        })
56        .flatten()
57        .unwrap_or(DEFAULT_FRAME_SECS);
58    ((secs * f64::from(rate)).round() as usize).max(1)
59}
60
61impl Driver {
62    pub(super) fn new(
63        value_graph: ValueGraph,
64        range: Extent,
65        block: usize,
66        config: &RenderConfig,
67        (memory, recording): (Memory, Recording),
68    ) -> Driver {
69        let frame = frame(config, config.rate);
70        Driver {
71            value_graph,
72            start: range.start,
73            at: range.start,
74            heard: range.start,
75            output: !super::dropped(config),
76            last: range.end,
77            block,
78            threads: config.threads,
79            keep: config.until.as_ref().map_or(0, |_| frame as i64),
80            until: config.until.clone(),
81            frame,
82            stop: None,
83            end: None,
84            most_bytes: 0,
85            samples: 0,
86            gauge: Gauge::default(),
87            memory,
88            recording,
89        }
90    }
91
92    /// Ends at `last`, unless `until` already stopped it.
93    pub(super) fn bound(&mut self, last: i64) {
94        self.last = last;
95        if self.stop.is_none() {
96            self.end = None;
97        }
98    }
99
100    pub(super) fn last(&self) -> i64 {
101        self.last
102    }
103
104    pub(super) fn next(&self) -> Extent {
105        let to = self
106            .next_to(self.block)
107            .min(blocks_end(self.at, self.threads));
108        Extent::new(self.at, to)
109    }
110
111    fn next_to(&self, n: usize) -> i64 {
112        self.last.min(self.at.saturating_add(n as i64)).max(self.at)
113    }
114
115    /// `n` samples from where it stands, cut where its range ends; `None` from there on. A
116    /// pull may drop the last one's samples, so each is taken first.
117    pub(super) fn read(&mut self, n: usize, heard: Hears) -> Result<Option<Buffer>, EngineError> {
118        let (from, to) = (self.at, self.next_to(n));
119        if !self.pulled(n, heard)? {
120            return Ok(None);
121        }
122        let mut out = self.played(from);
123        while self.at < to {
124            let at = self.at;
125            if !self.pulled((to - at) as usize, heard)? {
126                break;
127            }
128            let more = self.played(at);
129            for (held, more) in out.planes.iter_mut().zip(more.planes) {
130                held.extend(more);
131            }
132        }
133        Ok(Some(out))
134    }
135
136    pub(super) fn played(&self, from: i64) -> Buffer {
137        let end = self.end.map_or(self.at, |end| end.clamp(from, self.at));
138        self.value_graph
139            .samples(self.value_graph.root, Extent::new(from, end))
140    }
141
142    /// Stands at `to`, computing nothing before it; the stateful values it started silent,
143    /// each asked from before its run reaches.
144    pub(super) fn skip(&mut self, to: i64) -> Result<Vec<usize>, EngineError> {
145        if self.end.is_some_and(|end| self.at >= end) {
146            return Ok(Vec::new());
147        }
148        let to = to.min(self.last);
149        let silenced = self.value_graph.skipped(Extent::new(to, self.last))?;
150        self.recording.reach(to);
151        (self.at, self.heard) = (to, to);
152        let future = (to < self.last).then(|| Extent::new(to, self.last));
153        self.value_graph
154            .release(future, Extent::NOWHERE, self.since());
155        if self.last <= to {
156            self.end = Some(to);
157        }
158        Ok(silenced)
159    }
160
161    pub(super) fn pull(&mut self, heard: Hears) -> Result<bool, EngineError> {
162        self.pulled(self.block, heard)
163    }
164
165    /// The work done, and all its range comes to as its values now stand: `None` where the
166    /// range has no end.
167    pub(super) fn progress(&self) -> Progress {
168        let work = self.gauge;
169        let end = self.end.unwrap_or(self.last);
170        let left = match self.at < end {
171            false => Some(Gauge::default()),
172            true if end == i64::MAX => None,
173            true => {
174                let rest = Extent::new(self.at, end);
175                let held = (&self.memory, &self.recording);
176                match self.at == self.start {
177                    true => self.value_graph.ahead((self.next(), rest), held),
178                    false => self.value_graph.remaining(rest, held),
179                }
180            }
181        };
182        Progress {
183            work,
184            planned_samples: left.map(|left| work.done() + left.done()),
185        }
186    }
187
188    pub(super) fn work(&self) -> Work {
189        Work {
190            samples: self.samples,
191            computed_samples: self.gauge.computed_samples,
192        }
193    }
194
195    /// Up to `n` samples on, within a grid block per thread; false once it ended.
196    pub(super) fn pulled(&mut self, n: usize, heard: Hears) -> Result<bool, EngineError> {
197        let from = self.at;
198        if self.end.is_some_and(|end| from >= end) {
199            return Ok(false);
200        }
201        let to = self.next_to(n).min(blocks_end(from, self.threads));
202        self.recording.reach(from);
203        let asked_range = Extent::new(from, to);
204        let planned_samples = heard.and_then(|_| self.progress().planned_samples);
205        let hears = |before: Gauge| {
206            move |work: Gauge| {
207                let work = before.plus(work);
208                let progress = Progress {
209                    work,
210                    planned_samples,
211                };
212                heard.into_iter().for_each(|heard| heard(progress));
213            }
214        };
215        if from == self.start {
216            let hears = hears(self.gauge);
217            let history = self.value_graph.history(
218                (asked_range, self.last),
219                self.block as i64,
220                (&self.memory, &mut self.recording),
221                (self.threads, heard.map(|_| &hears as &dyn Fn(Gauge))),
222            )?;
223            self.counted(&history);
224        }
225        let hears = hears(self.gauge);
226        let pulled = self.value_graph.pull(
227            asked_range,
228            (&self.memory, &mut self.recording),
229            (self.threads, heard.map(|_| &hears as &dyn Fn(Gauge))),
230        )?;
231        self.counted(&pulled);
232        self.samples += (to - from) as u64;
233        self.at = to;
234        self.settle(from, to);
235        let future = (to < self.last).then(|| Extent::new(to, self.last));
236        let keep = Extent::new(from.min(to.saturating_sub(self.keep)), to);
237        self.value_graph.release(future, keep, self.since());
238        Ok(true)
239    }
240
241    fn since(&self) -> Option<i64> {
242        self.output.then_some(self.start)
243    }
244
245    fn counted(&mut self, pulled: &Pulled) {
246        self.gauge = self.gauge.plus(pulled.work);
247        self.most_bytes = self.most_bytes.max(pulled.most_bytes);
248    }
249
250    pub(super) fn most_bytes(&self) -> usize {
251        self.most_bytes
252    }
253
254    /// The one place `until` is checked. A level is known once its frame is whole, so only the
255    /// frames the last block left open are read again; a frame a skip cut into is never known.
256    fn settle(&mut self, from: i64, to: i64) {
257        if self.end.is_some() {
258            return;
259        }
260        if self.last <= to {
261            self.end = Some(to);
262        }
263        let Some(until) = &self.until else {
264            return;
265        };
266        let frame = self.frame as i64;
267        let open = self.start + (from - 1 - self.start).div_euclid(frame) * frame;
268        let base = open.max(self.heard);
269        let heard = self
270            .value_graph
271            .samples(self.value_graph.root, Extent::new(base, to));
272        let planes: Vec<&[f64]> = heard.planes.iter().map(Vec::as_slice).collect();
273        let rate = self.value_graph.values[self.value_graph.root].grid.rate;
274        let known = Known::new(planes, base, self.start, self.frame, rate, to == self.last);
275        if let Some(at) = until.first(&known, base, to) {
276            self.stop = Some(at);
277            self.end = Some(at.max(from));
278        }
279    }
280
281    pub(super) fn end(&self) -> Option<i64> {
282        self.end
283    }
284
285    pub(super) fn stop(&self) -> Option<i64> {
286        self.stop
287    }
288}
sva_engine/render/drive.rs

sva_engine/render/
drive.rs
