sva_engine/render/
through.rs

1// Concern: renders or warms from the root down through a persistent store, staging what it computes | Non-concern: the store's medium and commit | IO: (&Graph, target, Store) -> Render or CacheStats
2
3use std::collections::{BTreeMap, BTreeSet};
4
5use sva_ast::Graph;
6use sva_formula::{Hash, Held as Representation, NodeId};
7
8use super::table::spill::Spill;
9use super::table::{self, Table};
10use super::{Render, RenderConfig, closed, drive, driving, drove, frontier, planned_over};
11use crate::cache::{Backend, CacheStats, Lookup, Outcome, Recording, Store, Stored, Through};
12use crate::error::{Diagnostic, EngineError, Located};
13use crate::instantiate;
14use crate::schedule;
15use crate::typing::{self, Typing};
16
17/// `render` through `store`, from the root down: a node the store answers stands as its
18/// samples, and nothing under it is typed, planned or looked up. What the rest computes is
19/// staged beside the store as the render drops it; only `persist` writes the store.
20pub async fn render_through<B: Backend>(
21    graph: &Graph,
22    target: &str,
23    config: RenderConfig,
24    store: &Store<B>,
25) -> Result<Render, EngineError> {
26    match through(graph, target, config, store, Keep::Wanted).await? {
27        Reached::Render(held) => Ok(*held),
28        Reached::Held(_) => unreachable!("a render reads a stored root's samples"),
29    }
30}
31
32/// `render_through`'s staging alone, `config` asking nothing: a stored root ends it unread.
33pub async fn warm<B: Backend>(
34    graph: &Graph,
35    target: &str,
36    config: RenderConfig,
37    store: &Store<B>,
38) -> Result<CacheStats, EngineError> {
39    Ok(
40        match through(graph, target, config, store, Keep::Nothing).await? {
41            Reached::Render(mut held) => held.cache_stats.take().expect("a render through a store"),
42            Reached::Held(lookups) => CacheStats {
43                lookups,
44                ..CacheStats::default()
45            },
46        },
47    )
48}
49
50#[derive(Clone, Copy, PartialEq, Eq)]
51enum Keep {
52    Wanted,
53    Nothing,
54}
55
56enum Reached {
57    Render(Box<Render>),
58    Held(Vec<Lookup>),
59}
60
61async fn through<B: Backend>(
62    graph: &Graph,
63    target: &str,
64    config: RenderConfig,
65    store: &Store<B>,
66    keep: Keep,
67) -> Result<Reached, EngineError> {
68    let instances = instantiate::instantiate(graph, target, config.rate)?;
69    let root = instances.instance_of(target)?;
70    let order = schedule::schedule_from(&instances, std::slice::from_ref(&root))?;
71    let keys = keys(graph, &instances, &order, &config);
72    let mut found = frontier::Frontier::from((&instances, &order), &keys, &root, (&config, false));
73    found.walk(store).await;
74    if keep == Keep::Nothing && found.stored.contains_key(&root) {
75        return Ok(Reached::Held(found.lookups));
76    }
77    let (mut held, typed) = loop {
78        found.walk(store).await;
79        let tys = typing::infer_over(&instances, &order.within(&found.visited), &found.stored)?;
80        let id = tys
81            .id(&root)
82            .ok_or_else(|| EngineError::UnknownNode(root.clone()))?;
83        let bounds: BTreeSet<NodeId> = keys
84            .keys()
85            .filter_map(|path| tys.id(path))
86            .filter(|id| readable(&tys, *id))
87            .collect();
88        let typed = tys.lowered().to_vec();
89        let mut held = planned_over(&instances, (tys, id), config.clone(), &bounds)?;
90        let short = match (&mut held.table, held.range) {
91            (Some(table), Some(range)) => {
92                load(table, store, range).await;
93                short(table, range)
94            }
95            _ => Vec::new(),
96        };
97        if short.is_empty() {
98            break (held, typed);
99        }
100        for path in short {
101            found.reopen(&path);
102        }
103    };
104    let staging = staging(&held, &keys, &found);
105    let mut kept = BTreeSet::new();
106    if let Some(mut driver) = driving(&mut held, Recording::over(None, None))? {
107        driver.spill = Some(Spill::over(staging));
108        while spilled(&mut driver, store, &mut kept).await? {}
109        let rest = driver.spill.take().map(|mut s| s.rest(&driver.table));
110        for (key, meta) in rest.into_iter().flatten() {
111            store.stage_meta(key, &meta).await.map_err(unstaged)?;
112            kept.insert(key);
113        }
114        drove(&mut held, driver, keep == Keep::Wanted);
115    }
116    let computed = held.table.as_ref().map_or(Vec::new(), |table| {
117        let computed = table.values.iter();
118        let computed = computed.filter(|v| !matches!(v.kind, table::Kind::Stored { .. }));
119        computed.map(|v| v.name.clone()).collect()
120    });
121    let mut lookups = found.lookups;
122    for lookup in &mut lookups {
123        if lookup.outcome == Outcome::ComputedNotStored && kept.contains(&lookup.key) {
124            lookup.outcome = Outcome::ComputedStored;
125        }
126    }
127    let reached = held.cache_stats.take().map_or(Vec::new(), |stats| {
128        let walked = lookups.len();
129        stats.reached.iter().map(|(at, _)| (*at, walked)).collect()
130    });
131    held.cache_stats = Some(CacheStats {
132        lookups,
133        reached,
134        typed,
135        planned: computed,
136        ..CacheStats::default()
137    });
138    closed(&mut held)?;
139    Ok(Reached::Render(Box::new(held)))
140}
141
142/// Each instance's store key: its source identity at the render's rate and profile.
143pub(crate) fn keys(
144    graph: &Graph,
145    instances: &instantiate::Instances,
146    order: &schedule::Order,
147    config: &RenderConfig,
148) -> BTreeMap<String, Hash> {
149    crate::source::identities(graph, instances, order)
150        .into_iter()
151        .map(|(path, identity)| {
152            let key = crate::cache::node_key(identity, config.rate, &config.profile);
153            (path, key)
154        })
155        .collect()
156}
157
158/// What `window` asks of each stored value's samples, read off `store`; what it cannot read
159/// stays short.
160pub(crate) async fn load(table: &mut Table, store: &impl Through, window: sva_samples::Extent) {
161    for (stored, over) in table.wants(window) {
162        if let Some(samples) = store.read(&stored, over).await {
163            table.took(stored.key, samples);
164        }
165    }
166}
167
168/// Each stored node whose samples miss some its readers ask over `range`.
169fn short(table: &Table, range: sva_samples::Extent) -> Vec<String> {
170    let needs = table.demand(range);
171    table
172        .values
173        .iter()
174        .zip(&needs)
175        .filter(|(value, need)| {
176            matches!(value.kind, table::Kind::Stored { .. }) && !need.compute.is_empty()
177        })
178        .map(|(value, _)| value.name.clone())
179        .collect()
180}
181
182/// One block pulled, and what it handed out moved into the staging area.
183async fn spilled<B: Backend>(
184    driver: &mut drive::Driver,
185    store: &Store<B>,
186    kept: &mut BTreeSet<Hash>,
187) -> Result<bool, EngineError> {
188    let more = driver.pull()?;
189    let spill = driver
190        .spill
191        .as_mut()
192        .expect("a render through a store spills");
193    for (key, samples) in std::mem::take(&mut spill.samples) {
194        store.stage(key, &samples).await.map_err(unstaged)?;
195    }
196    for (key, meta) in spill.whole(&driver.table) {
197        store.stage_meta(key, &meta).await.map_err(unstaged)?;
198        kept.insert(key);
199    }
200    Ok(more)
201}
202
203fn unstaged(why: String) -> EngineError {
204    EngineError::refused(Diagnostic {
205        code: "store.unwritable".to_string(),
206        message: format!("what the render computed could not be staged beside the store: {why}"),
207        location: Located::default(),
208        help: "free the store's disk, move it with `--cache <path>`, or pass `--cache none`"
209            .to_string(),
210    })
211}
212
213/// A node another may read as its samples alone: samples, and nothing that reads it between
214/// them.
215fn readable(tys: &Typing, id: NodeId) -> bool {
216    tys.ty(id).held == Representation::Sampled && !schedule::anywhere(tys, id)
217}
218
219/// Each value a missed node computes, at its own node's key: the root whatever it is, any
220/// other where a reader may take its samples. One that only moves a value stored or staged
221/// stands for that value's samples, holding none of its own.
222fn staging(
223    held: &Render,
224    keys: &BTreeMap<String, Hash>,
225    found: &frontier::Frontier<'_>,
226) -> Vec<(usize, Hash, Stored)> {
227    let Some(table) = &held.table else {
228        return Vec::new();
229    };
230    let tys = &held.tys;
231    let mut out = Vec::new();
232    for (path, key) in keys {
233        if found.stored.contains_key(path) || !found.visited.contains(path) {
234            continue;
235        }
236        let Some((id, at)) = tys.id(path).and_then(|id| Some((id, table.of(id)?))) else {
237            continue;
238        };
239        let value = &table.values[at];
240        let own = tys.name(id) == path;
241        let readable = readable(tys, id) && value.alias().is_none();
242        let kept = value.pure
243            && value.period.is_none()
244            && !matches!(value.kind, table::Kind::Frames { .. });
245        if !own || !kept || !(at == table.root || readable) {
246            continue;
247        }
248        let (priced, moved) = under(table, at);
249        let ty = tys.ty(id);
250        let meta = Stored {
251            key: *key,
252            samples: Default::default(),
253            label: table.label(at),
254            width: u8::try_from(value.width).expect("a width the typing held"),
255            codomain: ty.codomain,
256            rate: ty.rate,
257            grid: tys.grid(id),
258            support: value.support,
259            priced,
260            moved,
261            readable,
262        };
263        out.push((at, *key, meta));
264    }
265    let staged: BTreeMap<usize, Hash> = out.iter().map(|(at, key, _)| (*at, *key)).collect();
266    for (at, _, meta) in &mut out {
267        let Some((moved, by)) = moves(table, *at) else {
268            continue;
269        };
270        let stored = table.values[moved].node.and_then(|id| {
271            let (file, shift) = found.stored.get(tys.name(id))?.file()?;
272            Some((file, by - shift))
273        });
274        let staged = staged.get(&moved).map(|key| (*key, by));
275        if let Some((of, by)) = staged.or(stored) {
276            *meta = meta.clone().referring(of, by);
277        }
278    }
279    out
280}
281
282/// The value `at` only moves, through every move between, and by how much.
283fn moves(table: &Table, at: usize) -> Option<(usize, i64)> {
284    let (mut read, mut by) = table.values[at].moves()?;
285    while let Some((next, shift)) = table.values[read].moves() {
286        (read, by) = (next, by + shift);
287    }
288    Some((read, by))
289}
290
291/// What a value and every value under it cost over the range, and the most any moved a read.
292fn under(table: &Table, at: usize) -> (u128, f64) {
293    let (mut seen, mut open) = (BTreeSet::from([at]), vec![at]);
294    let (mut priced, mut moved) = (0u128, 0.0f64);
295    while let Some(at) = open.pop() {
296        priced += table.planned[at];
297        moved = moved.max(table.values[at].moved);
298        open.extend(
299            table.values[at]
300                .reads
301                .iter()
302                .filter(|read| seen.insert(**read)),
303        );
304    }
305    (priced, moved)
306}
sva_engine/render/through.rs

sva_engine/render/
through.rs
