Struct ReactiveEngineV0 

pub struct ReactiveEngineV0 { /* private fields */ }

Implementations

impl ReactiveEngineV0

pub fn node_count(&self) -> usize

pub fn node_kind( &self, node: ReactiveNodeIdV0, ) -> Result<ReactiveNodeKindV0, ReactiveEngineErrorV0>

pub fn state( &self, node: ReactiveNodeIdV0, ) -> Result<&ReactiveStateV0, ReactiveEngineErrorV0>

pub fn is_necessary( &self, node: ReactiveNodeIdV0, ) -> Result<bool, ReactiveEngineErrorV0>

pub fn is_stale( &self, node: ReactiveNodeIdV0, ) -> Result<bool, ReactiveEngineErrorV0>

pub fn node_recompute_count( &self, node: ReactiveNodeIdV0, ) -> Result<usize, ReactiveEngineErrorV0>

pub fn delta_update_count( &self, node: ReactiveNodeIdV0, ) -> Result<usize, ReactiveEngineErrorV0>

pub fn current_wave(&self) -> u64

pub fn is_stabilizing(&self) -> bool

pub fn has_pending_work(&self) -> bool

pub fn deposit( &mut self, node: ReactiveNodeIdV0, state: ReactiveStateV0, ) -> Result<(), ReactiveEngineErrorV0>

Queues the latest value for an input-like node. The deposit is applied only when the next wave begins, including when it arrives between bounded steps of the current wave.

Examples found in repository

examples/performance_envelope.rs (lines 72-75)

fn measure(changes_input: bool) -> Result<Vec<Duration>, Box<dyn Error>> {
    let (mut engine, inputs) = graph()?;
    let mut samples = Vec::with_capacity(EVENT_COUNT);
    for event in 0..EVENT_COUNT {
        let index = event % INPUT_COUNT;
        let value = if changes_input {
            (index + event + 1) as u64
        } else {
            index as u64
        };
        let started = Instant::now();
        engine.deposit(
            inputs[index],
            ReactiveStateV0::available(ReactiveValueV0::Counter(value)),
        )?;
        settle(&mut engine)?;
        engine.drain_effect_receipts();
        samples.push(started.elapsed());
    }
    Ok(samples)
}

pub fn observe( &mut self, node: ReactiveNodeIdV0, ) -> Result<(), ReactiveEngineErrorV0>

Examples found in repository

examples/performance_envelope.rs (line 55)

fn graph() -> Result<(ReactiveEngineV0, Vec<ReactiveNodeIdV0>), Box<dyn Error>> {
    let policy = ChangePolicyV0::exact("performanceSemanticValue");
    let mut graph = ReactiveGraphBuilderV0::new();
    let mut inputs = Vec::with_capacity(INPUT_COUNT);
    let mut projections = Vec::with_capacity(INPUT_COUNT);
    for index in 0..INPUT_COUNT {
        let input = graph.add_input(
            ReactiveStateV0::available(ReactiveValueV0::Counter(index as u64)),
            policy,
        );
        inputs.push(input);
        projections.push((
            format!("projection-{index}"),
            graph.add_map(input, project_counter, policy),
        ));
    }
    let fold = graph.add_delta_fold(projections, policy)?;
    let boundary = graph.add_effect_boundary(fold, "performance-observation", policy);
    let mut engine = graph.build()?;
    engine.observe(boundary)?;
    settle(&mut engine)?;
    engine.drain_effect_receipts();
    Ok((engine, inputs))
}

pub fn unobserve( &mut self, node: ReactiveNodeIdV0, ) -> Result<(), ReactiveEngineErrorV0>

pub fn stabilize_step( &mut self, maximum_recomputes: usize, ) -> Result<StabilizeStatusV0, ReactiveEngineErrorV0>

Examples found in repository

examples/performance_envelope.rs (line 28)

fn settle(engine: &mut ReactiveEngineV0) -> Result<(), Box<dyn Error>> {
    loop {
        if matches!(
            engine.stabilize_step(4_096)?,
            StabilizeStatusV0::Settled { .. }
        ) {
            return Ok(());
        }
    }
}

pub fn stabilize_until_settled( &mut self, maximum_recomputes: usize, ) -> Result<StabilizeStatusV0, ReactiveEngineErrorV0>

pub fn drain_effect_receipts(&mut self) -> Vec<EffectReceiptV0>

Examples found in repository

examples/performance_envelope.rs (line 57)

fn graph() -> Result<(ReactiveEngineV0, Vec<ReactiveNodeIdV0>), Box<dyn Error>> {
    let policy = ChangePolicyV0::exact("performanceSemanticValue");
    let mut graph = ReactiveGraphBuilderV0::new();
    let mut inputs = Vec::with_capacity(INPUT_COUNT);
    let mut projections = Vec::with_capacity(INPUT_COUNT);
    for index in 0..INPUT_COUNT {
        let input = graph.add_input(
            ReactiveStateV0::available(ReactiveValueV0::Counter(index as u64)),
            policy,
        );
        inputs.push(input);
        projections.push((
            format!("projection-{index}"),
            graph.add_map(input, project_counter, policy),
        ));
    }
    let fold = graph.add_delta_fold(projections, policy)?;
    let boundary = graph.add_effect_boundary(fold, "performance-observation", policy);
    let mut engine = graph.build()?;
    engine.observe(boundary)?;
    settle(&mut engine)?;
    engine.drain_effect_receipts();
    Ok((engine, inputs))
}

fn measure(changes_input: bool) -> Result<Vec<Duration>, Box<dyn Error>> {
    let (mut engine, inputs) = graph()?;
    let mut samples = Vec::with_capacity(EVENT_COUNT);
    for event in 0..EVENT_COUNT {
        let index = event % INPUT_COUNT;
        let value = if changes_input {
            (index + event + 1) as u64
        } else {
            index as u64
        };
        let started = Instant::now();
        engine.deposit(
            inputs[index],
            ReactiveStateV0::available(ReactiveValueV0::Counter(value)),
        )?;
        settle(&mut engine)?;
        engine.drain_effect_receipts();
        samples.push(started.elapsed());
    }
    Ok(samples)
}

pub fn verify_delta_fold( &self, node: ReactiveNodeIdV0, ) -> Result<(), DeltaFoldParityErrorV0>