sva_engine/render/
run.rs

1// Concern: renders a target over memory, from the root down to what it answers | Non-concern: what memory keeps or writes, computing a value | IO: (&Graph, target, Tier) -> Render
2
3use std::collections::BTreeMap;
4use std::pin::Pin;
5use std::rc::Rc;
6use std::sync::Arc;
7
8use sva_ast::Graph;
9use sva_formula::NodeId;
10use sva_samples::Buffer;
11
12use super::value_graph;
13use super::world::{Reach, Reached, Walking, World};
14use super::{
15    Render, RenderConfig, closed, cut_of, driving, dropped, drove, ended, planned_over, told_end,
16};
17use crate::cache::{Backend, Call, Kind, Known, Phase, Recording, Stored, Tier};
18use crate::error::EngineError;
19use crate::schedule;
20
21/// Whether a render's caller let it go, asked between blocks.
22pub trait Abandon {
23    fn abandoned(&self) -> Pin<Box<dyn Future<Output = bool> + '_>>;
24}
25
26pub struct Never;
27
28/// What hears the root's samples block by block as they are pulled.
29pub trait Hear {
30    /// False lets the render go.
31    fn heard(&self, block: &Buffer) -> bool;
32}
33
34/// Renders made in turn, each advancing the version the last held.
35#[derive(Default)]
36pub struct Session {
37    own: Option<World>,
38    /// The composition with every volatile parameter at its stand-in.
39    pub(super) stand_in: Option<World>,
40}
41
42impl Abandon for Never {
43    fn abandoned(&self) -> Pin<Box<dyn Future<Output = bool> + '_>> {
44        Box::pin(std::future::ready(false))
45    }
46}
47
48/// `target` over `tier`, from the root down, in a session of its own.
49pub async fn render_over<B: Backend>(
50    graph: &Graph,
51    target: &str,
52    config: RenderConfig,
53    tier: &Tier<B>,
54) -> Result<Render, EngineError> {
55    render_in(
56        &mut Session::default(),
57        graph,
58        target,
59        config,
60        tier,
61        (&Never, None),
62    )
63    .await
64}
65
66/// `target` over `tier`, from the root down, typed anew only where `session` typed it
67/// otherwise; a node memory answers stands as its samples. With `out` dropped and no reading,
68/// a root it answers ends the render unread. Abandoned, or a block not heard, it stops before
69/// its next block.
70pub async fn render_in<B: Backend>(
71    session: &mut Session,
72    graph: &Graph,
73    target: &str,
74    config: RenderConfig,
75    tier: &Tier<B>,
76    hooks: (&dyn Abandon, Option<&dyn Hear>),
77) -> Result<Render, EngineError> {
78    let call = tier.call(Kind::Render, target, Phase::Typing);
79    render_called(session, graph, target, config, tier, hooks, call).await
80}
81
82/// `render_in` reporting as `call`, one its caller began.
83pub async fn render_called<B: Backend>(
84    session: &mut Session,
85    graph: &Graph,
86    target: &str,
87    config: RenderConfig,
88    tier: &Tier<B>,
89    (abandon, hear): (&dyn Abandon, Option<&dyn Hear>),
90    call: Call,
91) -> Result<Render, EngineError> {
92    let hooks = (abandon, hear, &call);
93    let done = run(session, (graph, target), config, tier, hooks).await;
94    call.end(match &done {
95        Ok(_) => Phase::Done,
96        Err(EngineError::Abandoned) => Phase::Abandoned,
97        Err(_) => Phase::Failed,
98    });
99    done
100}
101
102/// Where `driver` stands, told to `call`'s subscriber once `every` samples passed.
103fn delivered(call: &Call, driver: &super::drive::Driver, every: u64) {
104    let to = driver.end().unwrap_or(driver.at);
105    let last = driver.end().unwrap_or(driver.last());
106    let (samples, total) = ((to - driver.start) as u64, (last - driver.start) as u64);
107    call.delivered((samples, Some(total), every), || {
108        (driver.work.computed_samples, driver.recording.tally())
109    });
110}
111
112async fn run<B: Backend>(
113    session: &mut Session,
114    (graph, target): (&Graph, &str),
115    config: RenderConfig,
116    tier: &Tier<B>,
117    (abandon, hear, call): (&dyn Abandon, Option<&dyn Hear>, &Call),
118) -> Result<Render, EngineError> {
119    call.phase(Phase::Typing);
120    let memory = tier.memory();
121    let mut recording = Recording::over(memory);
122    let Session { own, stand_in } = session;
123    let (world, root) = World::rendered(own, graph, target, config.rate)?;
124    let (instances, typed) = (&world.instances, &world.typing);
125    let id = typed
126        .id(&root)
127        .ok_or_else(|| EngineError::UnknownNode(root.clone()))?;
128    let profile = config.profile;
129    let round = tier.begin();
130    let (mut supports, mut end) = world.bounds(id, &profile);
131    let known = supports.len();
132    let mut told = 0;
133    call.phase(Phase::LookingUp);
134    if end.is_none() && config.range.end.is_none() {
135        let stored = root_stored(world, &root, &config, (tier, round)).await;
136        let stated = stored.and_then(|stored| Some((stored.support, told_end(&stored, &profile)?)));
137        if let Some((support, stated)) = stated {
138            told += usize::from(supports.told(id, support));
139            end = Some(stated);
140        }
141    }
142    let (config, mut decided) = ended(typed, id, config, (supports, end));
143    world.bound(id, &profile, &decided);
144    let cut = decided.1.map(|end| cut_of(&end, &profile));
145    let lowered = typed.lowered().to_vec();
146    let found = walked(
147        world,
148        (&root, &config, false),
149        (tier, round, &mut recording),
150    )
151    .await;
152    call.looked(Phase::Planning, || recording.tally());
153    let tys = Rc::clone(typed);
154    if dropped(&config) && found.held.contains_key(&root) {
155        let schedule = schedule::plan(&tys, id, &config.asks);
156        let mut held = Render::shell((tys, id), (config, schedule), memory.clone());
157        let mut stats = recording.stats(memory);
158        stats.typed = lowered;
159        stats.analysed = decided.0.len().saturating_sub(known + told);
160        held.cache_stats = Some(stats);
161        return Ok(held);
162    }
163    let hits: BTreeMap<NodeId, Arc<Stored>> = found
164        .held
165        .iter()
166        .filter_map(|(path, stored)| Some((tys.id(path)?, Arc::clone(stored))))
167        .collect();
168    for (hit, stored) in &hits {
169        told += usize::from(decided.0.told(*hit, stored.support));
170    }
171    let mut analysed = decided.0.len();
172    let mut held = planned_over(
173        (graph, target, instances),
174        (tys, id),
175        (config, &mut *stand_in, memory.clone()),
176        (decided, &hits),
177    )?;
178    if let Some(value_graph) = &held.value_graph {
179        world.bound(id, &profile, &(value_graph.supports.clone(), None));
180    }
181    let mut retyped = lowered;
182    retyped.append(&mut held.stand_in_typed);
183    if let (Some(value_graph), Some(range)) = (&mut held.value_graph, held.range) {
184        loop {
185            let mut needs = value_graph.needs(range);
186            while !needs.is_empty() {
187                let fetched = tier.fetch(&needs).await;
188                for (key, parts) in fetched.handed {
189                    value_graph.took(key, &parts);
190                }
191                needs = fetched.left;
192                call.looked(Phase::Fetching, || recording.tally());
193            }
194            let short = value_graph.short((value_graph.root, range, value_graph::Past::Held));
195            if short.is_empty() {
196                break;
197            }
198            for at in short {
199                let under = held
200                    .tys
201                    .name(value_graph.answered(&held.tys, at))
202                    .to_string();
203                let walking = (under.as_str(), &held.config, true);
204                let found = walked(world, walking, (tier, round, &mut recording)).await;
205                let hits = found.held.iter();
206                let hits = hits
207                    .filter_map(|(path, stored)| Some((held.tys.id(path)?, Arc::clone(stored))));
208                let hits: BTreeMap<NodeId, Arc<Stored>> = hits.collect();
209                for (hit, stored) in &hits {
210                    told += usize::from(value_graph.supports.told(*hit, stored.support));
211                }
212                value_graph.read_on(&held.tys, at, &hits)?;
213            }
214            value_graph.settled(value_graph.root, &[], (range.start, false));
215            value_graph.refuse_endless(range)?;
216        }
217        value_graph.offers(&held.tys, range, cut);
218        analysed = value_graph.supports.len();
219    }
220    let analysed = analysed.saturating_sub(known + told);
221    let walked = recording.stats(memory);
222    match driving(&mut held, (memory, recording))? {
223        Some(mut driver) => {
224            let every = u64::from(held.config.rate / 10);
225            delivered(call, &driver, every);
226            loop {
227                if abandon.abandoned().await {
228                    return Err(EngineError::Abandoned);
229                }
230                let from = driver.at;
231                if !driver.pull()? {
232                    break;
233                }
234                if hear.is_some_and(|hear| !hear.heard(&driver.played(from))) {
235                    return Err(EngineError::Abandoned);
236                }
237                let needs = driver.value_graph.needs(driver.next());
238                for (key, parts) in tier.fetch(&needs).await.handed {
239                    driver.value_graph.took(key, &parts);
240                }
241                delivered(call, &driver, every);
242            }
243            if driver.stop().is_some() {
244                delivered(call, &driver, 0);
245            }
246            call.phase(Phase::Finishing);
247            drove(&mut held, driver);
248        }
249        None => held.cache_stats = Some(walked),
250    }
251    let computed = held.value_graph.as_ref().map_or(Vec::new(), |value_graph| {
252        let computed = value_graph.values.iter();
253        let computed =
254            computed.filter(|(_, v)| !matches!(v.kind, value_graph::Kind::Resident { .. }));
255        computed.map(|(_, v)| v.name.clone()).collect()
256    });
257    if let Some(stats) = &mut held.cache_stats {
258        stats.typed = retyped;
259        stats.planned = computed;
260        stats.analysed = analysed;
261        stats.unslotted = held.unslotted.clone();
262    }
263    closed(&mut held)?;
264    Ok(held)
265}
266
267/// `root`'s own node as stored, asked before the walk: its end decides the range walked.
268async fn root_stored<B: Backend>(
269    world: &World,
270    root: &str,
271    config: &RenderConfig,
272    (tier, round): (&Tier<B>, u64),
273) -> Option<Arc<Stored>> {
274    let key = world.root_key(root, config)?;
275    loop {
276        match tier.memory().answer(key, round) {
277            Known::Hit(stored) => return Some(stored),
278            Known::Miss => return None,
279            Known::Unknown => tier.lookup(key, round).await,
280        }
281    }
282}
283
284/// What memory answers from `root` down, or `under` it alone, each key it cannot answer looked
285/// up this round.
286async fn walked<B: Backend>(
287    world: &World,
288    (root, config, under): (&str, &RenderConfig, bool),
289    (tier, round, seen): (&Tier<B>, u64, &mut Recording),
290) -> Reached {
291    let walking = Walking {
292        root,
293        config,
294        whole: true,
295        under,
296    };
297    loop {
298        let memory = tier.memory();
299        match world.walk(&walking, &mut |node, key| {
300            memory.answered((key, round), (node, seen))
301        }) {
302            Reach::Reached(reached) => return reached,
303            Reach::Asks(keys) => {
304                for key in keys {
305                    tier.lookup(key, round).await;
306                }
307            }
308        }
309    }
310}
sva_engine/render/run.rs

sva_engine/render/
run.rs
