sva-engine 0.7.17

// Concern: which values a render offers memory as nodes, with their meta, and when | Non-concern: whether memory keeps or writes them | IO: (Render, keys, frontier) -> offers; (Table, Memory) -> offered

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::rc::Rc;
use std::sync::Arc;

use sva_formula::{Hash, Held as Representation, NodeId};
use sva_samples::Extent;

use super::Render;
use super::frontier::Frontier;
use super::table::segments::Segments;
use super::table::{self, Table};
use crate::cache::{Facts, Memory, Offered, Stored};
use crate::schedule;
use crate::typing::Typing;

struct Offer {
    at: usize,
    stored: Stored,
    own: Own,
    facts: Facts,
}

enum Own {
    /// Another offered or resident node's, moved.
    Moves,
    Shares(Offered),
    /// Nowhere a value keeps them: a period's samples, laid out over the range.
    Copies,
}

/// The values a render offers, each once its whole support is computed or the render ends.
pub(crate) struct Offers {
    pending: Vec<Offer>,
    keys: BTreeMap<usize, Hash>,
    range: Extent,
}

impl Offers {
    /// Each value a missed node computes, at its own node's key, with what memory decides from.
    pub(crate) fn of(
        held: &mut Render,
        (keys, found): (&BTreeMap<String, Hash>, &Frontier),
        memory: &Memory,
    ) -> Offers {
        let mut pending = Vec::new();
        let range = held.range.unwrap_or(Extent::NOWHERE);
        if let Some(table) = &mut held.table {
            let tys = &held.tys;
            let own = table.cuts_by_name(tys);
            let cuts = found.beneath(&|name| own.get(name).map(Vec::as_slice));
            let under = Under::of(table);
            for (path, key) in keys {
                if found.stored.contains_key(path) || !found.visited.contains(path) {
                    continue;
                }
                let Some((id, at)) = tys.id(path).and_then(|id| Some((id, table.of(id)?))) else {
                    continue;
                };
                if let Some(mut stored) = offerable(table, tys, (id, at), (path, *key), &under) {
                    let held = cuts.get(path.as_str());
                    stored.cuts = held.map_or(Vec::new(), |set| set.iter().copied().collect());
                    let facts = Facts {
                        slot: table.slot(at),
                        settled: false,
                        target: at == table.root,
                        samples: range.intersect(stored.support).len() as u64,
                    };
                    pending.push(Offer {
                        at,
                        stored,
                        own: Own::Moves,
                        facts,
                    });
                }
            }
        }
        let keys: BTreeMap<usize, Hash> = pending.iter().map(|o| (o.at, o.stored.key)).collect();
        if let Some(table) = &mut held.table {
            let feet = moves(table);
            for offer in &mut pending {
                if moved(table, (offer.at, &feet), &keys).is_none() {
                    offer.own = table.offered(offer.at).map_or(Own::Copies, Own::Shares);
                }
                if let Own::Shares(source) = &offer.own {
                    let stored = offer.stored.clone();
                    memory.offer(stored, source.clone(), offer.facts);
                }
            }
        }
        Offers {
            pending,
            keys,
            range,
        }
    }

    /// Each value whose whole support is computed, offered; one that moves another waits for
    /// the end, so what it moves is offered first.
    pub(crate) fn whole(&mut self, table: &Table, memory: &Memory) {
        let done = |offer: &Offer| {
            let value = &table.values[offer.at];
            if !matches!(offer.own, Own::Shares(_)) {
                return false;
            }
            let support = value.support();
            let mut computed = Segments::default();
            value.evaluated.iter().for_each(|e| computed.add(*e));
            support.is_bounded() && computed.covers(&Segments::of(support))
        };
        let (whole, rest): (Vec<Offer>, Vec<Offer>) = std::mem::take(&mut self.pending)
            .into_iter()
            .partition(done);
        self.pending = rest;
        self.offer(whole, table, memory);
    }

    /// Every value not yet offered, each that moves another after the rest.
    pub(crate) fn rest(&mut self, table: &Table, memory: &Memory) {
        let mut rest = std::mem::take(&mut self.pending);
        rest.sort_by_key(|offer| matches!(offer.own, Own::Moves));
        self.offer(rest, table, memory);
    }

    fn offer(&mut self, offers: Vec<Offer>, table: &Table, memory: &Memory) {
        let moving = offers.iter().any(|offer| matches!(offer.own, Own::Moves));
        let feet = if moving { moves(table) } else { HashMap::new() };
        for mut offer in offers {
            offer.stored.label = table.label(offer.at);
            let source = match offer.own {
                Own::Shares(source) => source,
                Own::Moves => match moved(table, (offer.at, &feet), &self.keys) {
                    Some(source) => source,
                    None => continue,
                },
                Own::Copies => {
                    let over = self.range.intersect(offer.stored.support);
                    Offered::Held(vec![Arc::new(table.samples(offer.at, over))])
                }
            };
            let facts = Facts {
                settled: true,
                ..offer.facts
            };
            memory.offer(offer.stored, source, facts);
        }
    }
}

fn offerable(
    table: &Table,
    tys: &Typing,
    (id, at): (NodeId, usize),
    (path, key): (&str, Hash),
    under: &Under,
) -> Option<Stored> {
    let value = &table.values[at];
    let own = tys.name(id) == path && crate::refs::passes(tys, id).is_none();
    let samples = value.pure && !matches!(value.kind, table::Kind::Frames { .. });
    if !own || !samples {
        return None;
    }
    let (priced, moved) = under.of_value(at);
    let ty = tys.ty(id);
    Some(Stored {
        key,
        identity: crate::refs::identity(tys, id).ok()?,
        label: table.label(at),
        width: u8::try_from(value.width).expect("a width the typing held"),
        codomain: ty.codomain,
        rate: ty.rate,
        grid: tys.grid(id),
        support: value.support(),
        priced,
        moved,
        readable: readable(tys, id) && value.alias().is_none(),
        sampled: ty.held == Representation::Sampled,
        cuts: Vec::new(),
        held: Vec::new(),
    })
}

/// A node another may read as its samples alone: samples, and nothing that reads it between
/// them.
pub(crate) fn readable(tys: &Typing, id: NodeId) -> bool {
    tys.ty(id).held == Representation::Sampled && !schedule::anywhere(tys, id)
}

fn moved(
    table: &Table,
    (at, feet): (usize, &HashMap<usize, (usize, i64)>),
    offered: &BTreeMap<usize, Hash>,
) -> Option<Offered> {
    let (moved, by) = *feet.get(&at)?;
    let resident = match &table.values[moved].kind {
        table::Kind::Resident(stored) => Some(stored.key),
        _ => None,
    };
    let of = offered.get(&moved).copied().or(resident)?;
    Some(Offered::Moves { of, by })
}

/// Each value that only moves another, mapped to the value at the end of what it moves and
/// the shift to it: found in the table's order, readers after what they read, each once.
fn moves(table: &Table) -> HashMap<usize, (usize, i64)> {
    let mut feet: HashMap<usize, (usize, i64)> = HashMap::new();
    for at in table.values.ordered() {
        if let Some((read, by)) = table.values[at].moves() {
            let foot = feet
                .get(&read)
                .map_or((read, by), |(foot, more)| (*foot, by + more));
            feet.insert(at, foot);
        }
    }
    feet
}

/// What each value and all under it cost, and the most any moved a read: a value one other
/// reads sums into its reader's own tree; one more read is summed once into each value over it.
struct Under {
    own: Vec<u128>,
    apart: Vec<Rc<BTreeSet<usize>>>,
    moved: Vec<f64>,
}

impl Under {
    fn of(table: &Table) -> Under {
        let span = table.values.span();
        let mut readers = vec![0u32; span];
        let distinct = |at: usize| {
            let mut reads = table.values[at].reads.clone();
            reads.sort_unstable();
            reads.dedup();
            reads
        };
        for at in table.values.ordered() {
            for read in distinct(at) {
                readers[read] += 1;
            }
        }
        let mut under = Under {
            own: vec![0; span],
            apart: vec![Rc::default(); span],
            moved: vec![0.0; span],
        };
        for at in table.values.ordered() {
            let (mut own, mut moved) = (table.planned[at], table.values[at].moved);
            let mut sets: Vec<Rc<BTreeSet<usize>>> = Vec::new();
            let mut more = Vec::new();
            for read in distinct(at) {
                crate::steps::step(1);
                moved = moved.max(under.moved[read]);
                match readers[read] {
                    1 => own += under.own[read],
                    _ => more.push(read),
                }
                let held = &under.apart[read];
                if !held.is_empty() && !sets.iter().any(|set| Rc::ptr_eq(set, held)) {
                    sets.push(Rc::clone(held));
                }
            }
            under.apart[at] = match (sets.len(), more.is_empty()) {
                (0, true) => Rc::default(),
                (1, true) => Rc::clone(&sets[0]),
                _ => Rc::new(
                    sets.iter()
                        .flat_map(|s| s.iter())
                        .copied()
                        .chain(more)
                        .collect(),
                ),
            };
            (under.own[at], under.moved[at]) = (own, moved);
        }
        under
    }

    fn of_value(&self, at: usize) -> (u128, f64) {
        crate::steps::step(self.apart[at].len());
        let apart = self.apart[at].iter().map(|s| self.own[*s]).sum::<u128>();
        (self.own[at] + apart, self.moved[at])
    }
}

#[cfg(test)]
mod tests {
    use crate::{RenderConfig, Tier, render};

    /// The steps a render folds, offering and bounding, over a chain `depth` nodes deep, each
    /// reading the one below 10 ms late, every node a miss.
    fn folded(depth: usize) -> u64 {
        folded_as(depth, "@P(t - 10ms)")
    }

    fn folded_as(depth: usize, body: &str) -> u64 {
        let mut files = sva_ast::Composition::new();
        let decays = "sample(crop(sin(2*pi*440*t)*exp(-t/0.1), 0s, 10s))\n";
        files.insert("c0", decays);
        for k in 1..=depth {
            files.insert(
                format!("c{k}"),
                body.replace('P', &format!("c{}", k - 1)) + "\n",
            );
        }
        let g = sva_ast::load(&files).expect("a composition");
        let before = crate::steps::taken();
        let top = format!("c{depth}");
        render(&g, &top, RenderConfig::at(8_000), &Tier::default()).expect("a render");
        crate::steps::taken() - before
    }

    /// Four times the nodes, four times the steps: no node walks what lies under it.
    #[test]
    fn a_chains_offers_and_bounds_fold_steps_linear_in_its_nodes() {
        let (short, long) = (folded(100), folded(400));
        assert!(short > 0);
        assert!(long <= 4 * short, "{short} then {long}");
    }

    /// Four times the nodes, under five times the steps.
    #[test]
    fn a_chain_of_scaled_and_summed_reads_folds_steps_linear_in_its_nodes() {
        for body in ["0.9*@P(t - 10ms)", "@P(t)*0.5 + @P(t - 10ms)*0.4"] {
            let (short, long) = (folded_as(50, body), folded_as(200, body));
            assert!(long < 5 * short, "{body}: {short} then {long}");
        }
    }
}