sva_engine/render/
stream.rs

1// Concern: opens a target as a stream over memory, editing it and its terms live | Non-concern: pulling its blocks, what an edit carries on | IO: (Graph, target) -> Stream; (Change) -> Changed
2
3use std::cell::RefCell;
4use std::collections::{BTreeMap, BTreeSet};
5use std::sync::Arc;
6
7use sva_ast::{Expr, Graph};
8use sva_formula::{Hash, NodeId};
9use sva_samples::{Buffer, Extent};
10
11use super::drive::{Block, Driver};
12use super::table::Table;
13use super::table::support::Supports;
14use super::terms::{Handle, NOTES, Terms, cut, placed};
15use super::{Ends, RenderConfig, range_over};
16use crate::cache::{Backend, CacheStats, Counters, Memory, Recording, Stored, Tier};
17use crate::error::{Diagnostic, EngineError, Located};
18use crate::flops::Work;
19use crate::recent::Recent;
20use world::{Plan, Walked, Wanted, World};
21
22#[cfg(test)]
23mod rebuilt;
24mod world;
25
26pub const STREAMED: &str = "streamed";
27
28/// The lookups, and the names started silent, a stream keeps of all it made.
29pub const LATEST: usize = 256;
30
31#[derive(Clone, Debug, PartialEq)]
32pub struct StreamConfig {
33    pub block: usize,
34    /// The target's own at open where `None`; a mono one plays in each, a wider is refused.
35    pub channels: Option<usize>,
36    pub render: RenderConfig,
37}
38
39/// A target rendered block by block off one table, reading `@notes`, the sum of the terms
40/// added. A change builds only what it changed or newly reads; the rest stays as it stood.
41pub struct Stream {
42    config: StreamConfig,
43    world: World,
44    driver: Driver,
45    expr: Expr,
46    terms: Terms,
47    width: usize,
48    supports: BTreeMap<Handle, Extent>,
49    /// The first sounding term's end, which `supports` evicts.
50    ending: Option<Option<i64>>,
51    generation: u64,
52    live: bool,
53    dropped: Recent<String>,
54    late: usize,
55    built: Built,
56    demands: usize,
57}
58
59#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
60pub struct Counts {
61    pub dropped: usize,
62    /// Edits that landed past where issued.
63    pub late: usize,
64    pub terms: usize,
65    /// The latest change's, or the open's.
66    pub built: Built,
67    /// The reads that worked out what the root asks of the note sum.
68    pub demands: usize,
69    pub tier: Counters,
70}
71
72/// What one change did anew, not what it carried over.
73#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
74pub struct Built {
75    /// Texts it took in: its expression and each node it newly reads.
76    pub parsed: usize,
77    pub instances: usize,
78    pub visited: usize,
79    pub typed: usize,
80    pub values: usize,
81    /// Values it made taking an old one's state.
82    pub copied: usize,
83    pub lookups: usize,
84}
85
86impl Stream {
87    pub async fn open<B: Backend>(
88        graph: &Graph,
89        target: &Expr,
90        config: StreamConfig,
91        tier: &Tier<B>,
92    ) -> Result<Stream, EngineError> {
93        let render = blocked(&config)?.render.clone();
94        let world = World::new(graph, &render)?;
95        let recording = Recording::over(tier.memory()).latest(LATEST);
96        let table = Table::new(&render.profile);
97        let driver = Driver::new(table, Extent::new(0, 0), config.block, &render, recording);
98        let mut stream = Stream {
99            world,
100            driver,
101            expr: target.clone(),
102            terms: Terms::default(),
103            width: 0,
104            supports: BTreeMap::new(),
105            ending: None,
106            generation: 0,
107            live: false,
108            dropped: Recent::keeping(LATEST),
109            late: 0,
110            built: Built::default(),
111            demands: 0,
112            config,
113        };
114        let opening = Prospect {
115            target: target.clone(),
116            terms: Terms::default(),
117            term: None,
118            from: None,
119            answer: Changed::Edited,
120            landing: None,
121            parsed: 1,
122        };
123        let mut local = Local::default();
124        let round = tier.begin();
125        loop {
126            match stream.attempt(&opening, &mut local, (tier.memory(), round))? {
127                Attempt::Landed(_) => break,
128                Attempt::Asks(keys) => local.look(keys, tier, round).await,
129                Attempt::Reads(wants) => local.read(wants, tier).await,
130                Attempt::Moved => unreachable!("nothing plays a stream before it opens"),
131            }
132        }
133        let mut needs = stream.needs();
134        while !needs.is_empty() {
135            let fetched = tier.fetch(&needs).await;
136            for (key, parts) in fetched.handed {
137                stream.driver.table.took(key, &parts);
138            }
139            needs = fetched.left;
140        }
141        Ok(stream)
142    }
143
144    pub fn graph(&self) -> &Graph {
145        &self.world.graph
146    }
147
148    fn prospect(&self, change: Change) -> Result<Prospect, Changed> {
149        let mut landing = None;
150        let prospect = |terms, term: Option<(Handle, Expr)>, from: Option<Graph>, answer| {
151            let roots = |(handle, term): &(Handle, Expr)| sva_ast::reads_of(&handle.node(), term);
152            Prospect {
153                target: self.expr.clone(),
154                terms,
155                from: from.map(|graph| (graph, term.as_ref().map(roots).unwrap_or_default())),
156                term,
157                answer,
158                landing: None,
159                parsed: 1,
160            }
161        };
162        Ok(match change {
163            Change::Target(graph, target) => {
164                let roots = sva_ast::reads_of(STREAMED, &target);
165                Prospect {
166                    target,
167                    from: Some((graph, roots)),
168                    ..prospect(self.terms.clone(), None, None, Changed::Edited)
169                }
170            }
171            Change::Add(graph, term, at) => {
172                let term = match at {
173                    Placed::Written => term,
174                    Placed::Landing => placed(&term, *landing.insert(self.driver.at)),
175                };
176                let (terms, handle) = self.terms.added();
177                Prospect {
178                    landing,
179                    ..prospect(
180                        terms,
181                        Some((handle, term)),
182                        Some(graph),
183                        Changed::Added(handle),
184                    )
185                }
186            }
187            Change::Replace(handle, graph, term, at) => {
188                let landed = self.terms.landed(handle).ok_or(Changed::Held(false))?;
189                let term = match at {
190                    Placed::Written => term,
191                    Placed::Landing => placed(&term, landed),
192                };
193                let terms = self.terms.clone();
194                prospect(
195                    terms,
196                    Some((handle, term)),
197                    Some(graph),
198                    Changed::Held(true),
199                )
200            }
201            Change::Remove(handle) => {
202                let at = self.driver.at as f64 / f64::from(self.config.render.rate);
203                let terms = self.terms.removed(handle).ok_or(Changed::Held(false))?;
204                let held = self.world.graph.expr(&handle.node());
205                let term = cut(held.expect("a held term's node"), at);
206                Prospect {
207                    parsed: 0,
208                    ..prospect(terms, Some((handle, term)), None, Changed::Held(true))
209                }
210            }
211        })
212    }
213
214    /// One try at `prospect`: planned, typed and built beside what plays, then held, or undone
215    /// where memory must answer first.
216    fn attempt(
217        &mut self,
218        prospect: &Prospect,
219        local: &mut Local,
220        (memory, round): (&Memory, u64),
221    ) -> Result<Attempt, EngineError> {
222        if prospect.landing.is_some_and(|at| at != self.driver.at) {
223            return Ok(Attempt::Moved);
224        }
225        let wanted = Wanted {
226            target: &prospect.target,
227            terms: &prospect.terms,
228            term: prospect.term.as_ref().map(|(h, e)| (*h, e)),
229            from: prospect.from.as_ref().map(|(g, roots)| (g, roots.clone())),
230        };
231        let found = |key: Hash| memory.answer(key, round);
232        let walked = self.world.plan(&wanted, &found)?;
233        let mut plan = match walked {
234            Walked::Asks(keys) => return Ok(Attempt::Asks(keys)),
235            Walked::Planned(plan) => plan,
236        };
237        let built = self.built(&mut plan);
238        let (root, range) = match built {
239            Ok(held) => held,
240            Err(e) => {
241                let freed = self.world.abort();
242                self.driver.table.abort(&freed);
243                return Err(e);
244            }
245        };
246        let window = ahead(self.driver.at.max(range.start), self.config.render.rate);
247        let wants = self.driver.table.needs_made(root, window);
248        let wants: Vec<(Hash, Extent)> = wants
249            .into_iter()
250            .filter(|(key, over)| !local.holds(*key, *over))
251            .collect();
252        if !wants.is_empty() {
253            let freed = self.world.abort();
254            self.driver.table.abort(&freed);
255            return Ok(Attempt::Reads(wants));
256        }
257        Ok(Attempt::Landed(self.land(
258            prospect,
259            (plan, root, range),
260            local,
261        )))
262    }
263
264    /// The plan typed and built: the root's value and its range.
265    fn built(&mut self, plan: &mut Plan) -> Result<(usize, Extent), EngineError> {
266        let typing = &mut self.world.typing;
267        let id = typing
268            .id(STREAMED)
269            .ok_or_else(|| EngineError::UnknownNode(STREAMED.to_string()))?;
270        let prefixes: BTreeMap<NodeId, Arc<Stored>> = plan
271            .prefixes
272            .iter()
273            .filter_map(|(path, stored)| Some((typing.id(path)?, Arc::clone(stored))))
274            .collect();
275        let table = &mut self.driver.table;
276        let root = table.grow(typing, id, &prefixes)?;
277        let plays = table.values[root].width;
278        let mut render = self.config.render.clone();
279        match self.width {
280            0 if self.config.channels == Some(0) => {
281                return Err(refusal("a stream of no channels".to_string()));
282            }
283            0 => widens(plays, self.config.channels.unwrap_or(plays))?,
284            width => {
285                widens(plays, width)?;
286                render.range.start = Some(self.driver.start);
287            }
288        }
289        let profile = &self.config.render.profile;
290        let support = Supports::over(typing, profile, Some(&table.supports)).of(id);
291        let range = range_over((&render, STREAMED), support, Ends::Pulled)?;
292        Ok((root, range))
293    }
294
295    /// The change held, each value it made carrying on what it continues.
296    fn land(
297        &mut self,
298        prospect: &Prospect,
299        (mut plan, root, range): (Plan, usize, Extent),
300        local: &mut Local,
301    ) -> Changed {
302        let freed = self.world.commit(&mut plan);
303        let now = self.driver.at;
304        let carried = self.driver.table.settled(root, &freed, (now, self.live));
305        for (key, parts) in &local.fetched {
306            self.driver.table.took(*key, parts);
307        }
308        for at in &carried.silent {
309            self.dropped
310                .push(self.driver.table.values[*at].name.clone());
311        }
312        match self.width {
313            0 => {
314                let plays = self.driver.table.values[root].width;
315                self.width = self.config.channels.unwrap_or(plays);
316                (self.driver.start, self.driver.at) = (range.start, range.start);
317            }
318            _ => self.late += usize::from(self.driver.at > local.issued),
319        }
320        let last = self.last(range.end);
321        self.driver.bound(last);
322        self.driver.recording.found(std::mem::take(&mut plan.hits));
323        self.expr = prospect.target.clone();
324        self.terms = prospect.terms.clone();
325        if let Changed::Added(handle) = prospect.answer {
326            self.terms.land(handle, self.driver.at);
327        }
328        if let Some((handle, _)) = prospect.term {
329            self.supported(handle);
330        }
331        self.built = Built {
332            parsed: prospect.parsed + plan.adopted,
333            instances: plan.named,
334            visited: plan.visited,
335            typed: self.world.typing.lowered().len(),
336            values: self.driver.table.built,
337            copied: carried.taken,
338            lookups: local.lookups,
339        };
340        self.generation += 1;
341        self.prune();
342        prospect.answer
343    }
344
345    /// `handle`'s support, where it still sounds.
346    fn supported(&mut self, handle: Handle) {
347        self.supports.remove(&handle);
348        self.ending = None;
349        if !self.terms.handles().any(|h| h == handle) {
350            return;
351        }
352        let typing = &self.world.typing;
353        let Some(id) = typing.id(&handle.node()) else {
354            return;
355        };
356        let profile = &self.config.render.profile;
357        let support = Supports::over(typing, profile, Some(&self.driver.table.supports)).of(id);
358        self.supports.insert(handle, support);
359    }
360
361    fn needs(&self) -> Vec<(Hash, Extent)> {
362        let next = ahead(self.driver.at, self.config.render.rate);
363        self.driver.table.needs(next)
364    }
365
366    /// `n` samples from `at`, `None` past the end. An `at` behind is refused; one ahead skips
367    /// there, computing through the span, or, live, as `go_live` says.
368    pub fn read(&mut self, at: i64, n: usize) -> Result<Option<Block>, EngineError> {
369        let now = self.driver.at;
370        if at < now {
371            return Err(refused(
372                "engine.stream_behind",
373                format!("sample {at} is before sample {now}, where the stream stands"),
374                "read from the stream's position or later",
375            ));
376        }
377        if n == 0 {
378            return Err(refused(
379                "engine.empty_read",
380                format!("a read of no samples at sample {at}"),
381                "read one sample or more",
382            ));
383        }
384        match self.live {
385            true if at > now => {
386                for silenced in self.driver.skip(at)? {
387                    self.dropped
388                        .push(self.driver.table.values[silenced].name.clone());
389                }
390            }
391            _ => {
392                let block = self.config.block;
393                while self.driver.at < at {
394                    let step = block.min((at - self.driver.at) as usize);
395                    if !self.driver.pulled(step)? {
396                        return Ok(None);
397                    }
398                    self.prune();
399                }
400            }
401        }
402        let block = self.driver.read(n)?;
403        self.prune();
404        Ok(block.map(|b| b.widened(self.width)))
405    }
406
407    /// An edited node with no state, or one a read skips past, starts silent, never computing
408    /// its past, named in `dropped`; a formula reads on exactly.
409    pub fn go_live(&mut self) {
410        self.live = true;
411        let last = self.last(self.driver.last());
412        self.driver.bound(last);
413    }
414
415    fn last(&self, range_end: i64) -> i64 {
416        match self.live {
417            true => self.config.render.range.end.unwrap_or(i64::MAX),
418            false => range_end,
419        }
420    }
421
422    /// The latest nodes a live edit started silent, of `counts().dropped`.
423    pub fn dropped(&self) -> Vec<&str> {
424        self.dropped.iter().map(String::as_str).collect()
425    }
426
427    pub fn counts(&self) -> Counts {
428        Counts {
429            dropped: self.dropped.made(),
430            late: self.late,
431            terms: self.terms.count(),
432            built: self.built,
433            demands: self.demands,
434            tier: self.driver.recording.since(),
435        }
436    }
437
438    /// Retires every term whose support ended by now and before the first sample of `notes`
439    /// the root asks from now on; nothing is worked out until a support can have ended.
440    fn prune(&mut self) {
441        let now = self.driver.at;
442        let supports = &self.supports;
443        let ending = *self
444            .ending
445            .get_or_insert_with(|| supports.values().map(|s| s.end).min());
446        if ending.is_none_or(|end| end > now) {
447            return;
448        }
449        let (table, tys) = (&self.driver.table, &self.world.typing);
450        let last = self.driver.last();
451        let asked = match tys.id(NOTES).and_then(|notes| table.of(notes)) {
452            Some(notes) if now < last => {
453                self.demands += 1;
454                let needs = table.demand(Extent::new(now, last));
455                needs[notes].hold.iter().next().map(|asked| asked.start)
456            }
457            Some(_) => None,
458            None => Some(i64::MIN),
459        };
460        let supports = &self.supports;
461        let gone = |handle: Handle| {
462            let support = supports.get(&handle);
463            support.is_some_and(|s| s.end <= now && asked.is_none_or(|from| s.end <= from))
464        };
465        let support = |handle: Handle| supports.get(&handle).copied();
466        let went = self.terms.retire(&gone, &support);
467        if !went.is_empty() {
468            self.generation += 1;
469        }
470        for handle in went {
471            self.supports.remove(&handle);
472            self.ending = None;
473        }
474    }
475
476    /// Every segment of its own clock the stream computed of `node`'s value, in order.
477    pub fn evaluated(&self, node: &str) -> Vec<sva_samples::Extent> {
478        let table = &self.driver.table;
479        self.world
480            .typing
481            .id(node)
482            .and_then(|id| table.of(id))
483            .map_or(Vec::new(), |at| table.values[at].evaluated.clone())
484    }
485
486    pub fn pruned(&self) -> sva_samples::Pruned {
487        self.driver.table.pruned(&self.world.typing)
488    }
489
490    pub fn landed(&self, handle: Handle) -> Option<i64> {
491        self.terms.landed(handle)
492    }
493
494    pub fn position(&self) -> i64 {
495        self.driver.at
496    }
497
498    pub fn work(&self) -> Work {
499        self.driver.work
500    }
501
502    pub fn stats(&self) -> CacheStats {
503        self.driver.recording.stats()
504    }
505
506    /// The bytes its values hold, samples and state.
507    pub fn held_bytes(&self) -> usize {
508        self.driver.table.bytes()
509    }
510
511    pub fn end(&self) -> Option<i64> {
512        self.driver.end()
513    }
514
515    pub fn width(&self) -> usize {
516        self.width
517    }
518
519    pub fn config(&self) -> &StreamConfig {
520        &self.config
521    }
522}
523
524/// An edit, each with the graph that reaches it and all the stream plays.
525pub enum Change {
526    Target(Graph, Expr),
527    Add(Graph, Expr, Placed),
528    Replace(Handle, Graph, Expr, Placed),
529    /// A sounding term is cut where the stream stands as the edit is built; what played stays.
530    Remove(Handle),
531}
532
533/// Where a term's sample 0 sits: the stream's own, or the sample its add lands at.
534#[derive(Clone, Copy, Debug, PartialEq, Eq)]
535pub enum Placed {
536    Written,
537    Landing,
538}
539
540/// A replace or remove answers whether the stream still held its handle.
541#[derive(Clone, Copy, Debug, PartialEq, Eq)]
542pub enum Changed {
543    Edited,
544    Added(Handle),
545    Held(bool),
546}
547
548/// What a change wants played, and what it answers once it lands.
549struct Prospect {
550    target: Expr,
551    terms: Terms,
552    term: Option<(Handle, Expr)>,
553    from: Option<(Graph, Vec<String>)>,
554    answer: Changed,
555    landing: Option<i64>,
556    /// Its own expression, where it brought one.
557    parsed: usize,
558}
559
560/// What one try at a change came to.
561enum Attempt {
562    Landed(Changed),
563    Asks(Vec<Hash>),
564    Reads(Vec<(Hash, Extent)>),
565    /// The stream left the sample it was placed at.
566    Moved,
567}
568
569/// What one change has looked up and read, across its tries.
570#[derive(Default)]
571struct Local {
572    fetched: Vec<(Hash, Vec<Arc<Buffer>>)>,
573    asked: Vec<(Hash, Extent)>,
574    unread: BTreeSet<Hash>,
575    lookups: usize,
576    issued: i64,
577}
578
579impl Local {
580    async fn look<B: Backend>(&mut self, keys: Vec<Hash>, tier: &Tier<B>, round: u64) {
581        for key in keys {
582            self.lookups += 1;
583            tier.lookup(key, round).await;
584        }
585    }
586
587    /// What `wants` asks, off memory; one a fetch's reads did not reach is asked again.
588    async fn read<B: Backend>(&mut self, wants: Vec<(Hash, Extent)>, tier: &Tier<B>) {
589        let fetched = tier.fetch(&wants).await;
590        for (key, over) in wants {
591            if fetched.left.contains(&(key, over)) {
592                continue;
593            }
594            self.asked.push((key, over));
595            if !fetched.handed.iter().any(|(held, _)| *held == key) {
596                self.unread.insert(key);
597            }
598        }
599        self.fetched.extend(fetched.handed);
600    }
601
602    /// Whether this change already read `key` over `over`, or could not.
603    fn holds(&self, key: Hash, over: Extent) -> bool {
604        self.unread.contains(&key)
605            || self
606                .asked
607                .iter()
608                .any(|(k, e)| *k == key && e.intersect(over) == over)
609    }
610}
611
612/// `build`'s change, the stream held only while each try runs, between two blocks: an exact
613/// stream pulled once its edit is done plays it where issued. `build` runs again whenever the
614/// stream changed under it.
615pub async fn change<E: From<EngineError>, B: Backend>(
616    stream: &RefCell<Stream>,
617    mut build: impl FnMut(&Stream) -> Result<Change, E>,
618    tier: &Tier<B>,
619) -> Result<Changed, E> {
620    let mut local = Local {
621        issued: stream.borrow().driver.at,
622        ..Local::default()
623    };
624    let round = tier.begin();
625    loop {
626        let change = build(&stream.borrow())?;
627        let prospect = match stream.borrow().prospect(change) {
628            Ok(prospect) => prospect,
629            Err(answer) => return Ok(answer),
630        };
631        let generation = stream.borrow().generation;
632        loop {
633            if stream.borrow().generation != generation {
634                break;
635            }
636            let memory = (tier.memory(), round);
637            let attempt = stream.borrow_mut().attempt(&prospect, &mut local, memory)?;
638            match attempt {
639                Attempt::Landed(answer) => return Ok(answer),
640                Attempt::Moved => break,
641                Attempt::Asks(keys) => local.look(keys, tier, round).await,
642                Attempt::Reads(wants) => local.read(wants, tier).await,
643            }
644        }
645    }
646}
647
648/// What the next second of `stream` reads off memory, as far as one fetch reaches; a block
649/// reading samples not ready computes them, or, live, starts them silent.
650pub async fn fetch<B: Backend>(stream: &RefCell<Stream>, tier: &Tier<B>) {
651    let needs = stream.borrow().needs();
652    let fetched = tier.fetch(&needs).await;
653    let mut stream = stream.borrow_mut();
654    for (key, parts) in fetched.handed {
655        stream.driver.table.took(key, &parts);
656    }
657}
658
659fn ahead(at: i64, rate: u32) -> Extent {
660    Extent::new(at, at.saturating_add(i64::from(rate)))
661}
662
663fn widens(plays: usize, width: usize) -> Result<(), EngineError> {
664    match plays == width || plays == 1 {
665        true => Ok(()),
666        false => Err(refused(
667            "engine.stream_width",
668            format!("this plays {plays} channel(s), and the stream plays {width}"),
669            "play as many channels as the stream, or one, or open a new stream for it",
670        )),
671    }
672}
673
674fn blocked(config: &StreamConfig) -> Result<&StreamConfig, EngineError> {
675    match config.block {
676        0 => Err(refusal("a block of no samples".to_string())),
677        _ => Ok(config),
678    }
679}
680
681pub(super) fn refusal(what: String) -> EngineError {
682    refused(
683        "engine.no_stream",
684        format!("this target opens no stream: {what}"),
685        "stream an expression over the nodes the composition defines",
686    )
687}
688
689fn refused(code: &str, message: String, help: &str) -> EngineError {
690    EngineError::refused(Diagnostic {
691        code: code.to_string(),
692        message,
693        location: Located::at(STREAMED, None),
694        help: help.to_string(),
695    })
696}
sva_engine/render/stream.rs

sva_engine/render/
stream.rs
