265 lines
9.3 KiB
Rust
265 lines
9.3 KiB
Rust
//! The T0 runtime: per-owner state, memo invalidation, and deterministic replay.
|
|
//!
|
|
//! Replaces `side_channel_generation.rs`. The registries a widget reads besides its graph
|
|
//! (files, models, completions, captures, board latches) were process-wide statics with a
|
|
//! process-wide generation counter. They now belong to a `Runtime`, so these tests need no
|
|
//! lock: each builds its own.
|
|
|
|
use flowrs_core::graph::GraphExecutor;
|
|
use flowrs_core::graph::llm_runtime::{Ask, Completion};
|
|
use flowrs_core::graph::runtime::{Event, Runtime, replay};
|
|
use flowrs_core::graph::twin_runtime::Capture;
|
|
use flowrs_core::graph::widget::WidgetRegistry;
|
|
use flowrs_core::graph::{Graph, WireMode};
|
|
|
|
fn capture() -> Capture {
|
|
Capture {
|
|
samples: Some(vec![1.0, 0.0]),
|
|
sample_rate_hz: 1000.0,
|
|
aliased: false,
|
|
source: "test".into(),
|
|
serial: String::new(),
|
|
}
|
|
}
|
|
|
|
/// Every mutator must invalidate the T0 memo, which folds in the generation; a read must
|
|
/// not.
|
|
#[test]
|
|
fn every_mutator_bumps_the_generation_and_reads_do_not() {
|
|
let rt = Runtime::new();
|
|
let mut last = rt.generation();
|
|
let mut check = |label: &str, rt: &Runtime| {
|
|
let g = rt.generation();
|
|
assert!(g > last, "{label} did not bump the generation");
|
|
last = g;
|
|
};
|
|
rt.put_file("f", vec![1]);
|
|
check("put_file", &rt);
|
|
rt.put_model("m", vec![1]);
|
|
check("put_model", &rt);
|
|
rt.put_capture("c", capture());
|
|
check("put_capture", &rt);
|
|
rt.put_completion(
|
|
&Ask::new("p", "m", 0.0),
|
|
Completion {
|
|
text: "t".into(),
|
|
model: "m".into(),
|
|
},
|
|
);
|
|
check("put_completion", &rt);
|
|
for kind in ["file", "model", "capture", "completion"] {
|
|
rt.clear(kind);
|
|
check(kind, &rt);
|
|
}
|
|
let g = rt.generation();
|
|
let _ = rt.files.get("f");
|
|
let _ = rt.models.names();
|
|
let _ = rt.captures.get("c");
|
|
let _ = rt.completions.is_empty();
|
|
assert_eq!(rt.generation(), g, "a read bumped the generation");
|
|
}
|
|
|
|
/// Regression: two runtimes in one process no longer share state. A file one owner
|
|
/// published, or a pin one board latched, was visible to every other.
|
|
#[test]
|
|
fn two_runtimes_share_nothing() {
|
|
let a = Runtime::new();
|
|
let b = Runtime::new();
|
|
a.put_file("run.csv", b"x".to_vec());
|
|
a.board.digital_write(13, true);
|
|
assert_eq!(b.files.get("run.csv"), None);
|
|
assert!(!b.board.digital_read(13));
|
|
assert!(a.board.digital_read(13));
|
|
}
|
|
|
|
fn adc_graph() -> Graph {
|
|
let mut g = Graph::new();
|
|
let src = g.add_node("embedded.adc", 0.0, 0.0);
|
|
g.get_node_mut(src)
|
|
.unwrap()
|
|
.properties
|
|
.insert("channel".into(), "3".into());
|
|
let abs = g.add_node("math.abs", 200.0, 0.0);
|
|
let reg = WidgetRegistry::with_builtins();
|
|
let out = reg.get("embedded.adc").unwrap().outputs()[0].name.clone();
|
|
let _ = g.add_wire(src, &out, abs, "value", WireMode::Move);
|
|
g
|
|
}
|
|
|
|
/// A run is a function of its graph and the logged external changes, so a log replays:
|
|
/// every step and in-node change reproduces on a fresh runtime.
|
|
#[test]
|
|
fn a_logged_run_replays_exactly() {
|
|
let rt = Runtime::new();
|
|
let mut exec = GraphExecutor::with_runtime(WidgetRegistry::with_builtins(), rt.clone());
|
|
let g = adc_graph();
|
|
rt.put_file("notes.txt", b"bench 4".to_vec());
|
|
exec.run(&g).expect("first run");
|
|
rt.put_model("m.onnx", vec![1, 2, 3]);
|
|
exec.run(&g).expect("second run");
|
|
|
|
let log = rt.log();
|
|
let steps = log
|
|
.events
|
|
.iter()
|
|
.filter(|e| matches!(e, Event::Step { .. }))
|
|
.count();
|
|
assert_eq!(steps, 4, "two nodes, two runs: {:?}", log.events);
|
|
assert!(
|
|
log.events.iter().any(|e| matches!(e, Event::External(_))),
|
|
"external changes are logged"
|
|
);
|
|
replay(&log, &g, WidgetRegistry::with_builtins()).expect("the log reproduces");
|
|
}
|
|
|
|
/// Replay is a check, not an echo: a log whose recorded outputs were tampered with fails.
|
|
#[test]
|
|
fn a_tampered_log_does_not_replay() {
|
|
let rt = Runtime::new();
|
|
let mut exec = GraphExecutor::with_runtime(WidgetRegistry::with_builtins(), rt.clone());
|
|
let g = adc_graph();
|
|
exec.run(&g).expect("run");
|
|
let mut log = rt.log();
|
|
for e in &mut log.events {
|
|
if let Event::Step { outputs, .. } = e {
|
|
for v in outputs.values_mut() {
|
|
v.push_str(" (edited)");
|
|
}
|
|
}
|
|
}
|
|
let err = replay(&log, &g, WidgetRegistry::with_builtins()).unwrap_err();
|
|
assert!(err[0].contains("diverges"), "{err:?}");
|
|
}
|
|
|
|
/// After an edit, the runs made with the old graph diverge by construction; `replay_from`
|
|
/// checks only the runs since, starting from the state they saw (the file published
|
|
/// before run 0 is applied, not lost with the run that is skipped).
|
|
#[test]
|
|
fn replay_from_checks_only_the_runs_since_an_edit() {
|
|
use flowrs_core::graph::runtime::replay_from;
|
|
let rt = Runtime::new();
|
|
let mut exec = GraphExecutor::with_runtime(WidgetRegistry::with_builtins(), rt.clone());
|
|
let mut g = Graph::new();
|
|
let path = g.add_node("StringLiteral", 0.0, 0.0);
|
|
g.get_node_mut(path)
|
|
.unwrap()
|
|
.properties
|
|
.insert("value".into(), "in.txt".into());
|
|
let read = g.add_node("ReadFile", 0.0, 0.0);
|
|
g.add_wire(path, "value", read, "path", WireMode::Copy)
|
|
.unwrap();
|
|
rt.put_file("in.txt", b"abc".to_vec());
|
|
exec.run(&g).expect("run 0");
|
|
// The edit: the read is now checked.
|
|
let ok = g.add_node("IsOk", 0.0, 0.0);
|
|
g.add_wire(read, "contents", ok, "input", WireMode::Copy)
|
|
.unwrap();
|
|
exec.run(&g).expect("run 1");
|
|
assert!(exec.unexecuted().is_empty(), "{:?}", exec.unexecuted());
|
|
|
|
let log = rt.log();
|
|
let whole = replay(&log, &g, WidgetRegistry::with_builtins()).unwrap_err();
|
|
assert!(
|
|
whole.len() == 1 && whole[0].starts_with("run 0 "),
|
|
"{whole:?}"
|
|
);
|
|
replay_from(&log, &g, WidgetRegistry::with_builtins(), 1).expect("run 1 reproduces");
|
|
}
|
|
|
|
/// A change a NODE made during a skipped run is state the next run starts from too: a
|
|
/// file written in run 0 is there when run 1 reads it. (No built-in node writes a store
|
|
/// while it runs today, so the writer here is the runtime's own step API.)
|
|
#[test]
|
|
fn replay_from_applies_what_skipped_runs_wrote() {
|
|
use flowrs_core::graph::runtime::replay_from;
|
|
use std::collections::HashMap;
|
|
let rt = Runtime::new();
|
|
let none = HashMap::new();
|
|
rt.begin_run();
|
|
rt.begin_step(99);
|
|
rt.put_file("in.txt", b"written in run 0".to_vec());
|
|
rt.end_step(99, "test.writer", &HashMap::new(), &none, &none);
|
|
|
|
let mut g = Graph::new();
|
|
let path = g.add_node("StringLiteral", 0.0, 0.0);
|
|
g.get_node_mut(path)
|
|
.unwrap()
|
|
.properties
|
|
.insert("value".into(), "in.txt".into());
|
|
let read = g.add_node("ReadFile", 0.0, 0.0);
|
|
g.add_wire(path, "value", read, "path", WireMode::Copy)
|
|
.unwrap();
|
|
let mut exec = GraphExecutor::with_runtime(WidgetRegistry::with_builtins(), rt.clone());
|
|
exec.run(&g).expect("run 1");
|
|
assert!(exec.unexecuted().is_empty(), "{:?}", exec.unexecuted());
|
|
let log = rt.log();
|
|
assert!(
|
|
log.events
|
|
.iter()
|
|
.any(|e| matches!(e, Event::Internal { node: 99, .. })),
|
|
"the write is logged as the node's own"
|
|
);
|
|
replay_from(&log, &g, WidgetRegistry::with_builtins(), 1).expect("run 1 reproduces");
|
|
}
|
|
|
|
/// A twin capture lives in the runtime of whoever ran the twin; the exported program
|
|
/// carries it only when codegen is given that runtime.
|
|
#[test]
|
|
fn the_exported_program_carries_the_runtimes_capture() {
|
|
let mut g = Graph::new();
|
|
let n = g.add_node("twin.capture", 0.0, 0.0);
|
|
g.get_node_mut(n)
|
|
.unwrap()
|
|
.properties
|
|
.insert("capture".into(), "bench".into());
|
|
let rt = Runtime::new();
|
|
rt.put_capture(
|
|
"bench",
|
|
Capture {
|
|
samples: Some(vec![1.0, 0.0, 1.0]),
|
|
source: "uno D13".into(),
|
|
..capture()
|
|
},
|
|
);
|
|
let with = g.to_rust_with_runtime(&rt).expect("codegen");
|
|
assert!(with.contains("uno D13"), "{with}");
|
|
let without = g.to_rust().expect("codegen");
|
|
assert!(without.contains("no capture"), "{without}");
|
|
}
|
|
|
|
#[test]
|
|
fn the_replay_log_is_a_ring_buffer_of_whole_runs() {
|
|
use flowrs_core::graph::runtime::{Event, Runtime};
|
|
use std::collections::HashMap;
|
|
let rt = Runtime::new();
|
|
rt.set_log_cap(Some(7));
|
|
let params: HashMap<String, String> = HashMap::new();
|
|
let none = HashMap::new();
|
|
for run in 0..100u64 {
|
|
rt.begin_run();
|
|
for n in 0..2 {
|
|
rt.begin_step(n);
|
|
rt.end_step(n, "t", ¶ms, &none, &none);
|
|
}
|
|
let log = rt.log();
|
|
assert!(log.events.len() <= 7, "run {run}: {}", log.events.len());
|
|
assert!(
|
|
matches!(log.events[0], Event::RunStart),
|
|
"starts at a run boundary"
|
|
);
|
|
}
|
|
let log = rt.log();
|
|
// Three events per run; 7 keeps the last two whole runs.
|
|
assert_eq!(log.events.len(), 6);
|
|
assert_eq!(log.dropped, 300 - 6);
|
|
// A single run longer than the cap is cut from its front.
|
|
rt.set_log_cap(Some(2));
|
|
assert_eq!(rt.log().events.len(), 2);
|
|
// Unbounded by default.
|
|
let free = Runtime::new();
|
|
for _ in 0..1000 {
|
|
free.begin_run();
|
|
}
|
|
assert_eq!(free.log().events.len(), 1000);
|
|
assert_eq!(free.log().dropped, 0);
|
|
}
|