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

sva_engine/render/
drive.rs
