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::{Memory, 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
21pub(super) struct Driver {
22    pub(super) value_graph: ValueGraph,
23    pub(super) start: i64,
24    pub(super) at: i64,
25    last: i64,
26    block: usize,
27    threads: NonZeroUsize,
28    until: Option<Until>,
29    frame: usize,
30    stop: Option<i64>,
31    end: Option<i64>,
32    /// Root samples held behind `at`, for `until` to read its open frame.
33    keep: i64,
34    /// Where the samples heard without a break since start from: a skip starts them anew.
35    heard: i64,
36    output: bool,
37    most_bytes: usize,
38    pub(super) work: Work,
39    pub(super) memory: Memory,
40    pub(super) recording: Recording,
41}
42
43/// The `envelope` reading's own frame at the render's rate, where one is asked.
44pub(super) fn frame(config: &RenderConfig, rate: u32) -> usize {
45    let secs = config
46        .asks
47        .iter()
48        .find_map(|ask| match ask.representation {
49            Representation::Envelope { frame_secs } => Some(frame_secs),
50            _ => None,
51        })
52        .flatten()
53        .unwrap_or(DEFAULT_FRAME_SECS);
54    ((secs * f64::from(rate)).round() as usize).max(1)
55}
56
57impl Driver {
58    pub(super) fn new(
59        value_graph: ValueGraph,
60        range: Extent,
61        block: usize,
62        config: &RenderConfig,
63        (memory, recording): (Memory, Recording),
64    ) -> Driver {
65        let frame = frame(config, config.rate);
66        Driver {
67            value_graph,
68            start: range.start,
69            at: range.start,
70            heard: range.start,
71            output: !super::dropped(config),
72            last: range.end,
73            block,
74            threads: config.threads,
75            keep: config.until.as_ref().map_or(0, |_| frame as i64),
76            until: config.until.clone(),
77            frame,
78            stop: None,
79            end: None,
80            most_bytes: 0,
81            work: Work::default(),
82            memory,
83            recording,
84        }
85    }
86
87    /// Ends at `last`, unless `until` already stopped it.
88    pub(super) fn bound(&mut self, last: i64) {
89        self.last = last;
90        if self.stop.is_none() {
91            self.end = None;
92        }
93    }
94
95    pub(super) fn last(&self) -> i64 {
96        self.last
97    }
98
99    pub(super) fn next(&self) -> Extent {
100        let to = self
101            .next_to(self.block)
102            .min(blocks_end(self.at, self.threads));
103        Extent::new(self.at, to)
104    }
105
106    fn next_to(&self, n: usize) -> i64 {
107        self.last.min(self.at.saturating_add(n as i64)).max(self.at)
108    }
109
110    /// `n` samples from where it stands, cut where its range ends; `None` from there on. A
111    /// pull may drop the last one's samples, so each is taken first.
112    pub(super) fn read(&mut self, n: usize) -> Result<Option<Buffer>, EngineError> {
113        let (from, to) = (self.at, self.next_to(n));
114        if !self.pulled(n)? {
115            return Ok(None);
116        }
117        let mut out = self.played(from);
118        while self.at < to {
119            let at = self.at;
120            if !self.pulled((to - at) as usize)? {
121                break;
122            }
123            let more = self.played(at);
124            for (held, more) in out.planes.iter_mut().zip(more.planes) {
125                held.extend(more);
126            }
127        }
128        Ok(Some(out))
129    }
130
131    fn played(&self, from: i64) -> Buffer {
132        let end = self.end.map_or(self.at, |end| end.clamp(from, self.at));
133        self.value_graph
134            .samples(self.value_graph.root, Extent::new(from, end))
135    }
136
137    /// Stands at `to`, computing nothing before it; the stateful values it started silent,
138    /// each asked from before its run reaches.
139    pub(super) fn skip(&mut self, to: i64) -> Result<Vec<usize>, EngineError> {
140        if self.end.is_some_and(|end| self.at >= end) {
141            return Ok(Vec::new());
142        }
143        let to = to.min(self.last);
144        let silenced = self.value_graph.skipped(Extent::new(to, self.last))?;
145        self.recording.reach(to);
146        (self.at, self.heard) = (to, to);
147        let future = (to < self.last).then(|| Extent::new(to, self.last));
148        self.value_graph
149            .release(future, Extent::NOWHERE, self.since());
150        if self.last <= to {
151            self.end = Some(to);
152        }
153        Ok(silenced)
154    }
155
156    pub(super) fn pull(&mut self) -> Result<bool, EngineError> {
157        self.pulled(self.block)
158    }
159
160    /// Up to `n` samples on, within a grid block per thread; false once it ended.
161    pub(super) fn pulled(&mut self, n: usize) -> Result<bool, EngineError> {
162        let from = self.at;
163        if self.end.is_some_and(|end| from >= end) {
164            return Ok(false);
165        }
166        let to = self.next_to(n).min(blocks_end(from, self.threads));
167        self.recording.reach(from);
168        let asked_range = Extent::new(from, to);
169        if from == self.start {
170            let history = self.value_graph.history(
171                asked_range,
172                self.block as i64,
173                (&self.memory, &mut self.recording),
174                self.threads,
175            )?;
176            self.counted(&history);
177        }
178        let pulled = self.value_graph.pull(
179            asked_range,
180            (&self.memory, &mut self.recording),
181            self.threads,
182        )?;
183        self.counted(&pulled);
184        self.work.samples += (to - from) as u64;
185        self.at = to;
186        self.settle(from, to);
187        let future = (to < self.last).then(|| Extent::new(to, self.last));
188        let keep = Extent::new(from.min(to.saturating_sub(self.keep)), to);
189        self.value_graph.release(future, keep, self.since());
190        Ok(true)
191    }
192
193    fn since(&self) -> Option<i64> {
194        self.output.then_some(self.start)
195    }
196
197    fn counted(&mut self, pulled: &Pulled) {
198        self.work.computed_samples += pulled.computed_samples;
199        self.most_bytes = self.most_bytes.max(pulled.most_bytes);
200    }
201
202    pub(super) fn most_bytes(&self) -> usize {
203        self.most_bytes
204    }
205
206    /// The one place `until` is checked. A level is known once its frame is whole, so only the
207    /// frames the last block left open are read again; a frame a skip cut into is never known.
208    fn settle(&mut self, from: i64, to: i64) {
209        if self.end.is_some() {
210            return;
211        }
212        if self.last <= to {
213            self.end = Some(to);
214        }
215        let Some(until) = &self.until else {
216            return;
217        };
218        let frame = self.frame as i64;
219        let open = self.start + (from - 1 - self.start).div_euclid(frame) * frame;
220        let base = open.max(self.heard);
221        let heard = self
222            .value_graph
223            .samples(self.value_graph.root, Extent::new(base, to));
224        let planes: Vec<&[f64]> = heard.planes.iter().map(Vec::as_slice).collect();
225        let rate = self.value_graph.values[self.value_graph.root].grid.rate;
226        let known = Known::new(planes, base, self.start, self.frame, rate, to == self.last);
227        if let Some(at) = until.first(&known, base, to) {
228            self.stop = Some(at);
229            self.end = Some(at.max(from));
230        }
231    }
232
233    pub(super) fn end(&self) -> Option<i64> {
234        self.end
235    }
236
237    pub(super) fn stop(&self) -> Option<i64> {
238        self.stop
239    }
240}
sva_engine/render/drive.rs

sva_engine/render/
drive.rs
