sva-engine 0.9.6

// 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

use std::collections::BTreeMap;
use std::pin::Pin;
use std::rc::Rc;
use std::sync::Arc;

use sva_ast::Graph;
use sva_formula::NodeId;
use sva_samples::Buffer;

use super::value_graph;
use super::world::{Reach, Reached, Walking, World};
use super::{
    Render, RenderConfig, closed, cut_of, driving, dropped, drove, ended, planned_over, told_end,
};
use crate::cache::{Backend, Call, Kind, Known, Phase, Recording, Stored, Tier};
use crate::error::EngineError;
use crate::schedule;

/// Whether a render's caller let it go, asked between blocks.
pub trait Abandon {
    fn abandoned(&self) -> Pin<Box<dyn Future<Output = bool> + '_>>;
}

pub struct Never;

/// What hears the root's samples block by block as they are pulled.
pub trait Hear {
    /// False lets the render go.
    fn heard(&self, block: &Buffer) -> bool;
}

/// Renders made in turn, each advancing the version the last held.
#[derive(Default)]
pub struct Session {
    own: Option<World>,
    /// The composition with every volatile parameter at its stand-in.
    pub(super) stand_in: Option<World>,
}

impl Abandon for Never {
    fn abandoned(&self) -> Pin<Box<dyn Future<Output = bool> + '_>> {
        Box::pin(std::future::ready(false))
    }
}

/// `target` over `tier`, from the root down, in a session of its own.
pub async fn render_over<B: Backend>(
    graph: &Graph,
    target: &str,
    config: RenderConfig,
    tier: &Tier<B>,
) -> Result<Render, EngineError> {
    render_in(
        &mut Session::default(),
        graph,
        target,
        config,
        tier,
        (&Never, None),
    )
    .await
}

/// `target` over `tier`, from the root down, typed anew only where `session` typed it
/// otherwise; a node memory answers stands as its samples. With `out` dropped and no reading,
/// a root it answers ends the render unread. Abandoned, or a block not heard, it stops before
/// its next block.
pub async fn render_in<B: Backend>(
    session: &mut Session,
    graph: &Graph,
    target: &str,
    config: RenderConfig,
    tier: &Tier<B>,
    hooks: (&dyn Abandon, Option<&dyn Hear>),
) -> Result<Render, EngineError> {
    let call = tier.call(Kind::Render, target, Phase::Typing);
    render_called(session, graph, target, config, tier, hooks, call).await
}

/// `render_in` reporting as `call`, one its caller began.
pub async fn render_called<B: Backend>(
    session: &mut Session,
    graph: &Graph,
    target: &str,
    config: RenderConfig,
    tier: &Tier<B>,
    (abandon, hear): (&dyn Abandon, Option<&dyn Hear>),
    call: Call,
) -> Result<Render, EngineError> {
    let hooks = (abandon, hear, &call);
    let done = run(session, (graph, target), config, tier, hooks).await;
    call.end(match &done {
        Ok(_) => Phase::Done,
        Err(EngineError::Abandoned) => Phase::Abandoned,
        Err(_) => Phase::Failed,
    });
    done
}

/// Where `driver` stands, told to `call`'s subscriber once `every` samples passed.
fn delivered(call: &Call, driver: &super::drive::Driver, every: u64) {
    let to = driver.end().unwrap_or(driver.at);
    let last = driver.end().unwrap_or(driver.last());
    let (samples, total) = ((to - driver.start) as u64, (last - driver.start) as u64);
    call.delivered((samples, Some(total), every), || {
        (driver.work.computed_samples, driver.recording.tally())
    });
}

async fn run<B: Backend>(
    session: &mut Session,
    (graph, target): (&Graph, &str),
    config: RenderConfig,
    tier: &Tier<B>,
    (abandon, hear, call): (&dyn Abandon, Option<&dyn Hear>, &Call),
) -> Result<Render, EngineError> {
    call.phase(Phase::Typing);
    let memory = tier.memory();
    let mut recording = Recording::over(memory);
    let Session { own, stand_in } = session;
    let (world, root) = World::rendered(own, graph, target, config.rate)?;
    let (instances, typed) = (&world.instances, &world.typing);
    let id = typed
        .id(&root)
        .ok_or_else(|| EngineError::UnknownNode(root.clone()))?;
    let profile = config.profile;
    let round = tier.begin();
    let (mut supports, mut end) = world.bounds(id, &profile);
    let known = supports.len();
    let mut told = 0;
    call.phase(Phase::LookingUp);
    if end.is_none() && config.range.end.is_none() {
        let stored = root_stored(world, &root, &config, (tier, round)).await;
        let stated = stored.and_then(|stored| Some((stored.support, told_end(&stored, &profile)?)));
        if let Some((support, stated)) = stated {
            told += usize::from(supports.told(id, support));
            end = Some(stated);
        }
    }
    let (config, mut decided) = ended(typed, id, config, (supports, end));
    world.bound(id, &profile, &decided);
    let cut = decided.1.map(|end| cut_of(&end, &profile));
    let lowered = typed.lowered().to_vec();
    let found = walked(
        world,
        (&root, &config, false),
        (tier, round, &mut recording),
    )
    .await;
    call.looked(Phase::Planning, || recording.tally());
    let tys = Rc::clone(typed);
    if dropped(&config) && found.held.contains_key(&root) {
        let schedule = schedule::plan(&tys, id, &config.asks);
        let mut held = Render::shell((tys, id), (config, schedule), memory.clone());
        let mut stats = recording.stats(memory);
        stats.typed = lowered;
        stats.analysed = decided.0.len().saturating_sub(known + told);
        held.cache_stats = Some(stats);
        return Ok(held);
    }
    let hits: BTreeMap<NodeId, Arc<Stored>> = found
        .held
        .iter()
        .filter_map(|(path, stored)| Some((tys.id(path)?, Arc::clone(stored))))
        .collect();
    for (hit, stored) in &hits {
        told += usize::from(decided.0.told(*hit, stored.support));
    }
    let mut analysed = decided.0.len();
    let mut held = planned_over(
        (graph, target, instances),
        (tys, id),
        (config, &mut *stand_in, memory.clone()),
        (decided, &hits),
    )?;
    if let Some(value_graph) = &held.value_graph {
        world.bound(id, &profile, &(value_graph.supports.clone(), None));
    }
    let mut retyped = lowered;
    retyped.append(&mut held.stand_in_typed);
    if let (Some(value_graph), Some(range)) = (&mut held.value_graph, held.range) {
        loop {
            let mut needs = value_graph.needs(range);
            while !needs.is_empty() {
                let fetched = tier.fetch(&needs).await;
                for (key, parts) in fetched.handed {
                    value_graph.took(key, &parts);
                }
                needs = fetched.left;
                call.looked(Phase::Fetching, || recording.tally());
            }
            let short = value_graph.short((value_graph.root, range, value_graph::Past::Held));
            if short.is_empty() {
                break;
            }
            for at in short {
                let under = held
                    .tys
                    .name(value_graph.answered(&held.tys, at))
                    .to_string();
                let walking = (under.as_str(), &held.config, true);
                let found = walked(world, walking, (tier, round, &mut recording)).await;
                let hits = found.held.iter();
                let hits = hits
                    .filter_map(|(path, stored)| Some((held.tys.id(path)?, Arc::clone(stored))));
                let hits: BTreeMap<NodeId, Arc<Stored>> = hits.collect();
                for (hit, stored) in &hits {
                    told += usize::from(value_graph.supports.told(*hit, stored.support));
                }
                value_graph.read_on(&held.tys, at, &hits)?;
            }
            value_graph.settled(value_graph.root, &[], (range.start, false));
            value_graph.refuse_endless(range)?;
        }
        value_graph.offers(&held.tys, range, cut);
        analysed = value_graph.supports.len();
    }
    let analysed = analysed.saturating_sub(known + told);
    let walked = recording.stats(memory);
    match driving(&mut held, (memory, recording))? {
        Some(mut driver) => {
            let every = u64::from(held.config.rate / 10);
            delivered(call, &driver, every);
            loop {
                if abandon.abandoned().await {
                    return Err(EngineError::Abandoned);
                }
                let from = driver.at;
                if !driver.pull()? {
                    break;
                }
                if hear.is_some_and(|hear| !hear.heard(&driver.played(from))) {
                    return Err(EngineError::Abandoned);
                }
                let needs = driver.value_graph.needs(driver.next());
                for (key, parts) in tier.fetch(&needs).await.handed {
                    driver.value_graph.took(key, &parts);
                }
                delivered(call, &driver, every);
            }
            if driver.stop().is_some() {
                delivered(call, &driver, 0);
            }
            call.phase(Phase::Finishing);
            drove(&mut held, driver);
        }
        None => held.cache_stats = Some(walked),
    }
    let computed = held.value_graph.as_ref().map_or(Vec::new(), |value_graph| {
        let computed = value_graph.values.iter();
        let computed =
            computed.filter(|(_, v)| !matches!(v.kind, value_graph::Kind::Resident { .. }));
        computed.map(|(_, v)| v.name.clone()).collect()
    });
    if let Some(stats) = &mut held.cache_stats {
        stats.typed = retyped;
        stats.planned = computed;
        stats.analysed = analysed;
        stats.unslotted = held.unslotted.clone();
    }
    closed(&mut held)?;
    Ok(held)
}

/// `root`'s own node as stored, asked before the walk: its end decides the range walked.
async fn root_stored<B: Backend>(
    world: &World,
    root: &str,
    config: &RenderConfig,
    (tier, round): (&Tier<B>, u64),
) -> Option<Arc<Stored>> {
    let key = world.root_key(root, config)?;
    loop {
        match tier.memory().answer(key, round) {
            Known::Hit(stored) => return Some(stored),
            Known::Miss => return None,
            Known::Unknown => tier.lookup(key, round).await,
        }
    }
}

/// What memory answers from `root` down, or `under` it alone, each key it cannot answer looked
/// up this round.
async fn walked<B: Backend>(
    world: &World,
    (root, config, under): (&str, &RenderConfig, bool),
    (tier, round, seen): (&Tier<B>, u64, &mut Recording),
) -> Reached {
    let walking = Walking {
        root,
        config,
        whole: true,
        under,
    };
    loop {
        let memory = tier.memory();
        match world.walk(&walking, &mut |node, key| {
            memory.answered((key, round), (node, seen))
        }) {
            Reach::Reached(reached) => return reached,
            Reach::Asks(keys) => {
                for key in keys {
                    tier.lookup(key, round).await;
                }
            }
        }
    }
}