polydat_core/kernel/engines.rs
1// Copyright 2024-2026 Jonathan Shook
2// SPDX-License-Identifier: Apache-2.0
3
4//! Polydat evaluation engines: EngineCore (shared eval loop) and the three
5//! P1 engine types — PolydatState (dependent-list), RawState (no provenance),
6//! and ProvScanState (provenance-scan).
7
8use std::sync::atomic::{AtomicU64, Ordering};
9use std::sync::{Arc, Mutex, OnceLock};
10
11use super::WireSource;
12use super::program::PolydatProgram;
13use crate::ast::Value;
14
15/// Cached lookup of the `NBRS_DIRTY_DEBUG` env var. Called from
16/// the per-cycle hot path (`PolydatState::set_input`); reading the
17/// real `std::env::var` on every cycle costs ~30% of CPU on
18/// single-fiber dryrun benches (it walks the libc env table and
19/// formats a fresh CString each call). The OnceLock evaluates
20/// once on first touch and every subsequent call is one atomic
21/// load.
22fn nbrs_dirty_debug_enabled() -> bool {
23 static FLAG: OnceLock<bool> = OnceLock::new();
24 *FLAG.get_or_init(|| std::env::var("NBRS_DIRTY_DEBUG").is_ok())
25}
26
27/// A cross-kernel mutable cell for a `shared`-modifier wire.
28///
29/// When a `shared` output in an outer scope is bound into an
30/// inner kernel via `materialize_wiring_from_outer`, both kernels' input
31/// slots reference the same `SharedCell`. Writes from inner via
32/// `set_input` flow through to the cell; reads on either side
33/// pick up the latest value.
34///
35/// Concurrent writers serialize at the Mutex; the current
36/// semantic is **last-write-wins** (lock-acquisition order).
37/// Future templated patterns (atomic-fetch-add, sum-reduction,
38/// merge, etc.) — see SRD-16 §"Open: concurrent shared
39/// mutation" — will introduce alternative cell types selected
40/// per binding declaration.
41///
42/// ## Cross-fiber validity tracking
43///
44/// Each cell carries its own validity-tracking handles per
45/// `polydat/docs/design/cross_fiber_invalidation.md`:
46///
47/// - `revision: AtomicU64` — monotonic counter, bumped on every
48/// write. Consumer fibers cache the last revision they
49/// observed in their per-fiber `last_seen` map; a mismatch
50/// tells the cone walker to re-evaluate.
51/// - `scope_intent_dirty: Arc<AtomicU64>` — one intent word, shared
52/// with every other cell allocated from the same word. The
53/// cell's `bit` position is set on every write, allowing
54/// consumers to do an O(1) bulk check ("any cell in this
55/// scope dirty?") before drilling down to the per-cell
56/// revision compare.
57/// - `bit: u8` — this cell's position within its word. The scope
58/// keeps one `Arc<AtomicU64>` per 64 cells, grown on demand by
59/// the defining scope's `EngineCore::allocate_cell_bit`.
60///
61/// The reader contract (S5 §1.1) is preserved: a producer's
62/// `publish` writes value + revision + intent bit in three
63/// Release stores; a consumer's `check_clean` walk on its next
64/// read observes the change without any host-side ceremony.
65pub struct SharedCellInner {
66 /// Cell value. The mutex serialises concurrent writers and
67 /// gives readers single-value atomicity.
68 pub value: Mutex<Value>,
69 /// Monotonic revision counter. Bumped on every write
70 /// (Release); compared by consumers (Acquire) against
71 /// per-fiber `last_seen`.
72 pub revision: AtomicU64,
73 /// Defining scope's intent-dirty bit-vector. Shared by Arc
74 /// across every cell allocated by the same scope. On every
75 /// write the producer ORs `1 << self.bit` into this
76 /// (Release) so consumers' bulk-mask check sees the scope
77 /// as dirty.
78 pub scope_intent_dirty: Arc<AtomicU64>,
79 /// This cell's bit position in `scope_intent_dirty`. Stable
80 /// for the cell's lifetime.
81 pub bit: u8,
82}
83
84impl std::fmt::Debug for SharedCellInner {
85 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
86 f.debug_struct("SharedCellInner")
87 .field("revision", &self.revision.load(Ordering::Relaxed))
88 .field("bit", &self.bit)
89 .finish_non_exhaustive()
90 }
91}
92
93impl SharedCellInner {
94 /// Construct a new cell with the given initial value, bound
95 /// to the defining scope's intent-dirty word at the given
96 /// bit position within that word. Callers must allocate
97 /// `(word, bit)` via `EngineCore::allocate_cell_bit` —
98 /// the bit is not reusable for the cell's lifetime.
99 pub fn new(initial: Value, scope_intent_dirty: Arc<AtomicU64>, bit: u8) -> Self {
100 debug_assert!(
101 bit < 64,
102 "bit-within-word {bit} must be < 64; the allocator splits >64-bit \
103 scope vectors across multiple words"
104 );
105 Self {
106 value: Mutex::new(initial),
107 revision: AtomicU64::new(0),
108 scope_intent_dirty,
109 bit,
110 }
111 }
112
113 /// Producer-side write: replace the cell's value, bump the
114 /// revision, set the intent bit. Three Release stores
115 /// publish the write across any consumer fiber per
116 /// `cross_fiber_invalidation.md` §6. The mutex critical
117 /// section is held only for the value swap; the atomics
118 /// run outside it.
119 pub fn publish(&self, value: Value) {
120 {
121 let mut guard = self.value.lock().unwrap();
122 *guard = value;
123 }
124 self.revision.fetch_add(1, Ordering::Release);
125 self.scope_intent_dirty
126 .fetch_or(1u64 << self.bit, Ordering::Release);
127 }
128
129 /// Consumer-side read: snapshot the cell's value and the
130 /// revision it was published at. Returns a pair so the
131 /// caller can update its `last_seen[cell] = revision`
132 /// alongside taking the value, without a second cell access.
133 pub fn snapshot(&self) -> (Value, u64) {
134 // Acquire-load the revision first so the value read
135 // synchronises-with the producer's value publication.
136 // The mutex itself provides the memory barrier for the
137 // value, but the revision is read with explicit Acquire
138 // for the cross-fiber happens-before relation.
139 let value = self.value.lock().unwrap().clone();
140 let revision = self.revision.load(Ordering::Acquire);
141 (value, revision)
142 }
143}
144
145/// Externally-held handle to a shared cell. `Arc<SharedCellInner>`
146/// so a single cell can be referenced from many kernels at
147/// once. The handle is cheap to clone (Arc bump).
148pub type SharedCell = Arc<SharedCellInner>;
149
150/// Per-node cone metadata for cell-bound input dependencies.
151///
152/// Built lazily on first `check_cell_clean` per node and
153/// cached in [`EngineCore::cell_cones`]; invalidated by
154/// clearing the cache whenever cells are attached or detached.
155///
156/// The structure groups a node's cell-bound input dependencies
157/// by the defining scope's `intent_dirty` Arc (compared by
158/// `Arc::ptr_eq`). Each group carries the bulk-check
159/// `interest_mask` for the scope plus per-cell drill-down
160/// entries — implementing the bulk-mask + per-cell-revision
161/// protocol from `cross_fiber_invalidation.md` §5.
162#[derive(Debug, Default, Clone)]
163pub(crate) struct CellCone {
164 /// Cells grouped by defining-scope's `intent_dirty`. Empty
165 /// = no cell-bound deps; check returns trivially clean.
166 pub(crate) groups: Vec<CellConeGroup>,
167}
168
169#[derive(Debug, Clone)]
170pub(crate) struct CellConeGroup {
171 /// Defining scope's `intent_dirty` vector (Arc cloned from
172 /// the cells). Bulk-mask check: AND this against
173 /// `interest_mask`; if zero, every cell in this group is
174 /// clean for this consumer (modulo last_seen) — skip the
175 /// drill-down.
176 pub(crate) intent_dirty: Arc<AtomicU64>,
177 /// OR of `1 << cell.bit` for every cell in this group.
178 pub(crate) interest_mask: u64,
179 /// Per-cell drill-down entries. Each gives the bit position
180 /// in `intent_dirty` plus the input slot index where the
181 /// cell is attached, for revision compare against
182 /// `last_seen`.
183 pub(crate) cells: Vec<CellConeEntry>,
184}
185
186#[derive(Debug, Clone, Copy)]
187pub(crate) struct CellConeEntry {
188 pub(crate) bit: u8,
189 pub(crate) input_slot: usize,
190}
191
192/// One named shared cell propagated through the parent → child
193/// scope chain. Carried on `PolydatKernel` (and surfaced through
194/// `ScopeKernel::shared_cells_in_scope`) so a descendant whose
195/// program declares a matching input slot can attach the cell —
196/// even when intermediate scopes' bodies never name it and so
197/// have no input slot for it themselves.
198///
199/// Without this carrier, an ancestral `shared X := …` cell
200/// becomes invisible past the first intermediate scope under
201/// the closure-binding economy. With it, every spawn step
202/// computes "every cell visible at this scope" and threads the
203/// full set forward — the cascade is transitive by
204/// construction.
205#[derive(Clone, Debug)]
206pub struct SharedCellEntry {
207 /// The binding's name.
208 pub name: String,
209 /// The cell's declared type.
210 pub port_type: crate::ast::PortType,
211 /// The cell.
212 pub cell: SharedCell,
213}
214
215/// SRD-82 §"Panic reporting: one full render" — set by a host
216/// runtime that catches worker panics and renders the full
217/// enriched diagnostic itself (the `errors:` block). When set,
218/// the re-raise hook below prints a single first-line notice
219/// instead of the full body; bare polydat consumers never set it
220/// and keep the full print.
221static PANIC_REPORTING_DOWNSTREAM: std::sync::atomic::AtomicBool =
222 std::sync::atomic::AtomicBool::new(false);
223
224/// Declare that a downstream reporter will render eval-panic
225/// diagnostics in full (see `PANIC_REPORTING_DOWNSTREAM`).
226pub fn set_panic_reporting_downstream(on: bool) {
227 PANIC_REPORTING_DOWNSTREAM.store(on, std::sync::atomic::Ordering::Relaxed);
228}
229
230thread_local! {
231 /// True while a node eval runs inside the enrichment
232 /// catch_unwind in `eval_node`. The suppression hook checks
233 /// this to swallow the raw std panic-hook print (bare payload
234 /// + backtrace pointer at the original panic site) — that
235 /// same panic is about to be caught, enriched with
236 /// node/output/input context, and re-raised via `panic_any`,
237 /// which fires the hook again with this flag clear. Net
238 /// effect: exactly ONE hook print, and it's the enriched one.
239 static EVAL_PANIC_CAPTURE: std::cell::Cell<bool> =
240 const { std::cell::Cell::new(false) };
241 /// Original panic location captured by the suppression hook
242 /// while the flag above is set. Folded into the enriched
243 /// message so the true `file:line` survives the re-raise
244 /// (the re-raised panic's own location points at the
245 /// re-raise site, which is useless).
246 static EVAL_PANIC_LOCATION: std::cell::RefCell<Option<String>> =
247 const { std::cell::RefCell::new(None) };
248 /// One-shot marker armed just before the enriched re-raise
249 /// when a downstream reporter exists: the hook prints a short
250 /// first-line notice for that panic instead of the full body.
251 static RERAISE_SHORT: std::cell::Cell<bool> =
252 const { std::cell::Cell::new(false) };
253}
254
255/// Install (once, process-wide) a panic hook that chains to the
256/// previously installed hook unless the current thread is inside
257/// the wrapped node eval, in which case it records the panic
258/// location and stays quiet.
259fn install_eval_panic_hook() {
260 static HOOK: std::sync::Once = std::sync::Once::new();
261 HOOK.call_once(|| {
262 let prev = std::panic::take_hook();
263 std::panic::set_hook(Box::new(move |info| {
264 if EVAL_PANIC_CAPTURE.with(|c| c.get()) {
265 let loc = info.location().map(|l| l.to_string());
266 EVAL_PANIC_LOCATION.with(|slot| *slot.borrow_mut() = loc);
267 } else if RERAISE_SHORT.with(|c| c.replace(false)) {
268 // The runtime will render the full enriched
269 // diagnostic in the phase error list; one short
270 // line keeps the terminal signal without the
271 // four-fold repeat (SRD-82 §one full render).
272 let first = info
273 .payload()
274 .downcast_ref::<String>()
275 .map(String::as_str)
276 .and_then(|m| m.lines().next())
277 .unwrap_or("<non-string panic payload>");
278 eprintln!("op eval panic (detail in phase errors): {first}");
279 } else {
280 prev(info);
281 }
282 }));
283 });
284}
285
286/// RAII guard arming the suppression hook for one wrapped eval.
287/// Saves and restores the previous flag value: nodes that drive
288/// sub-kernels (comprehensions, gk-call) nest evals, and each
289/// level's catch_unwind must see its own panics suppressed.
290pub(crate) struct EvalPanicCaptureGuard {
291 prev: bool,
292}
293
294impl EvalPanicCaptureGuard {
295 pub(crate) fn arm() -> Self {
296 install_eval_panic_hook();
297 let prev = EVAL_PANIC_CAPTURE.with(|c| c.replace(true));
298 EVAL_PANIC_LOCATION.with(|slot| slot.borrow_mut().take());
299 Self { prev }
300 }
301}
302
303impl Drop for EvalPanicCaptureGuard {
304 fn drop(&mut self) {
305 EVAL_PANIC_CAPTURE.with(|c| c.set(self.prev));
306 }
307}
308
309/// The text of a panic payload: a `String` or a `&str`, else a marker.
310pub(crate) fn panic_payload_text(payload: &(dyn std::any::Any + Send)) -> String {
311 payload
312 .downcast_ref::<&'static str>()
313 .map(|s| (*s).to_string())
314 .or_else(|| payload.downcast_ref::<String>().cloned())
315 .unwrap_or_else(|| "<non-string panic payload>".into())
316}
317
318/// Build the rich diagnostic message for a node-level eval panic, on
319/// every engine (engines.md §3.4): the original payload, the
320/// panic location the capture guard recorded, the node's function
321/// name, every output it feeds, the program's diagnostic context
322/// (typically the source path / scope label), and the input values,
323/// already formatted (the interpreter's `Value`s through
324/// [`format_value_for_diag`], a compiled kernel's slots through its
325/// decoder). This is what the user sees instead of the bare panic
326/// payload, and it reads the same whichever engine raised it.
327pub(crate) fn enrich_panic(
328 payload: Box<dyn std::any::Any + Send>,
329 node_name: &str,
330 output_names: &[&str],
331 context: &str,
332 inputs: &[String],
333) -> String {
334 let original = panic_payload_text(payload.as_ref());
335 // A payload that already carries node context came from a
336 // nested wrapped eval's re-raise; its captured "location" is
337 // the re-raise site, not the original panic — skip it.
338 let location_line = if original.contains("↳ in node") {
339 String::new()
340 } else {
341 EVAL_PANIC_LOCATION
342 .with(|slot| slot.borrow_mut().take())
343 .map(|loc| format!("\n ↳ panicked at {loc}"))
344 .unwrap_or_default()
345 };
346 let outputs_label = if output_names.is_empty() {
347 "no declared output".to_string()
348 } else {
349 format!(
350 "output{} {}",
351 if output_names.len() == 1 { "" } else { "s" },
352 output_names.join(", ")
353 )
354 };
355 let mut input_label = String::new();
356 for (i, v) in inputs.iter().enumerate() {
357 if i > 0 {
358 input_label.push_str(", ");
359 }
360 input_label.push_str(&format!("[{i}]={v}"));
361 }
362 format!(
363 "{original}{location_line}\n ↳ in node `{node_name}` ({outputs_label}) \
364 while evaluating {context}\n \
365 ↳ inputs: [{input_label}]"
366 )
367}
368
369/// Drop the `↳ panicked at <file>:<line>` line from an enriched
370/// message, keeping the node, outputs and inputs.
371///
372/// For a failure at *build* — the compile-constant fold — the reason is
373/// a property of the program, and polydat's own source location reads
374/// as an internal defect rather than the diagnosis it is: the user sees
375/// "bad spec `s97`" and a line number in a file they do not have. The
376/// three compiled engines never carried it there, so stripping it on
377/// the interpreter is what makes the fold error the same sentence on
378/// every engine ([`crate::KernelError::ConstantFold`]).
379///
380/// Evaluation failures keep the location: a panic at run time is as
381/// likely to be a node's bug as a program's, and then it is the first
382/// thing worth knowing.
383pub(crate) fn without_panic_location(message: String) -> String {
384 message
385 .lines()
386 .filter(|l| !l.trim_start().starts_with("↳ panicked at "))
387 .collect::<Vec<_>>()
388 .join("\n")
389}
390
391/// Re-raise an enriched message as the interpreter does: through
392/// `panic_any`, so the hook prints it once, or prints the short notice
393/// when a downstream reporter renders the full body (SRD-82).
394pub(crate) fn reraise_enriched(enriched: String) -> ! {
395 if PANIC_REPORTING_DOWNSTREAM.load(std::sync::atomic::Ordering::Relaxed) {
396 RERAISE_SHORT.with(|c| c.set(true));
397 }
398 std::panic::panic_any(enriched)
399}
400
401/// The interpreter's enrichment: the node's name and outputs from the
402/// program, the inputs as the `Value`s it was called with.
403fn enrich_eval_panic(
404 payload: Box<dyn std::any::Any + Send>,
405 program: &PolydatProgram,
406 node_idx: usize,
407 inputs: &[Value],
408) -> String {
409 let node_name = program
410 .nodes
411 .get(node_idx)
412 .map(|n| n.meta().name.to_string())
413 .unwrap_or_else(|| format!("<unknown node #{node_idx}>"));
414 let mut output_names: Vec<&str> = program
415 .output_map_iter()
416 .filter_map(|(name, (n_idx, _))| {
417 if *n_idx == node_idx {
418 Some(name.as_str())
419 } else {
420 None
421 }
422 })
423 .collect();
424 output_names.sort();
425 let inputs: Vec<String> = inputs.iter().map(format_value_for_diag).collect();
426 enrich_panic(
427 payload,
428 &node_name,
429 &output_names,
430 program.context(),
431 &inputs,
432 )
433}
434
435/// Format a `Value` into a short diagnostic string. Strings are
436/// quoted + truncated; vectors print their length not contents.
437pub(crate) fn format_value_for_diag(v: &Value) -> String {
438 match v {
439 Value::U64(n) => format!("U64({n})"),
440 Value::F64(n) => format!("F64({n})"),
441 Value::Bool(b) => format!("Bool({b})"),
442 Value::Str(s) => {
443 let trimmed: String = s.chars().take(40).collect();
444 if s.chars().count() > 40 {
445 format!("Str({trimmed:?}…)")
446 } else {
447 format!("Str({trimmed:?})")
448 }
449 }
450 Value::None => "None".to_string(),
451 other => format!("{:?}", other.port_type()),
452 }
453}
454
455/// Shared evaluation state for all Polydat engines. Contains the node
456/// output buffers, input values, and the eval loop.
457/// Engine types wrap this and provide their own invalidation strategy.
458pub struct EngineCore {
459 /// Per-node output value buffers, reused across evaluations.
460 pub(crate) buffers: Vec<Vec<Value>>,
461 /// Per-node: true = cached output is valid, false = needs eval.
462 pub(crate) node_clean: Vec<bool>,
463 /// Current input values (coordinates + captures, all unified).
464 /// For a cell-bound slot this entry is unused: the cell is the
465 /// slot's only register (`read_input` reads it, `set_input`
466 /// publishes to it).
467 pub(crate) inputs: Vec<Value>,
468 /// Default values for each input (used by reset_inputs).
469 pub(crate) input_defaults: Vec<Value>,
470 /// Optional cross-kernel shared cell per input slot. `None`
471 /// = local-only input (the common case). `Some(cell)` =
472 /// the slot is bound to a shared cell; writes propagate
473 /// through the cell to whatever other kernels share it.
474 pub(crate) shared_cells: Vec<Option<SharedCell>>,
475 /// SRD-13f Push B.2 — per-output broadcast cell. Indexed
476 /// by output position in `program.output_list`. `Some(cell)`
477 /// = the output broadcasts its value to descendants via
478 /// the cell whenever the owner pulls the output; `None` =
479 /// no broadcast subscribers were set up (no descendant
480 /// scope binds against this output's name).
481 ///
482 /// `materialize_wiring_from_outer` plumbs the same `Arc<SharedCell>`
483 /// onto the matching input slot on the inner kernel — at
484 /// that point both ends share the storage. Inner reads
485 /// transparently through the cell on every `read_input`;
486 /// outer's `pull` writes the freshly computed value into
487 /// the cell so subsequent inner reads return the current
488 /// value with no traversal.
489 pub(crate) output_cells: Vec<Option<SharedCell>>,
490 /// Pre-allocated scratch buffer for node input gathering.
491 pub(crate) input_scratch: Vec<Value>,
492 /// Per node, the scratch entries the node declared through
493 /// `scratch_layout` (a native cone's own slot buffer): storage
494 /// belongs to the state, never to the node, which is shared by
495 /// every state of the program (axiom S3).
496 pub(crate) node_scratch: Vec<Vec<crate::ast::ScratchBuf>>,
497 /// This scope's intent-dirty bit-vector. One `AtomicU64`
498 /// word per 64 cells allocated by this scope; new words
499 /// are appended on demand by [`Self::allocate_cell_bit`].
500 /// Each cell carries a clone of the specific `Arc<AtomicU64>`
501 /// for its word (and its bit-within-word). Consumer fibers'
502 /// bulk-mask check (per `cross_fiber_invalidation.md` §5)
503 /// groups cells by `Arc::ptr_eq` of their word and ANDs
504 /// the loaded word against the cone's interest mask for
505 /// that word.
506 ///
507 /// The `Vec<Arc<...>>` shape — rather than a single
508 /// `Arc<Vec<AtomicU64>>` — lets cells take a stable
509 /// per-word handle that the scope can grow without
510 /// invalidating any existing cell's reference.
511 pub(crate) scope_intent_words: Vec<Arc<AtomicU64>>,
512 /// Next bit position to allocate from
513 /// [`Self::scope_intent_words`]. Word index is
514 /// `next_cell_bit / 64`; bit within word is
515 /// `next_cell_bit % 64`. Monotonic; bits are never reused
516 /// within a scope's lifetime.
517 pub(crate) next_cell_bit: u32,
518 /// Per-fiber cache of the last revision this engine observed
519 /// for each cell it has read. Keyed by `Arc::as_ptr` of the
520 /// `SharedCellInner`. Sparse; entries are inserted lazily
521 /// on first observation via `check_cell_clean`.
522 ///
523 /// Per-fiber state — no contention. Pointer keys are stable
524 /// for the cell's lifetime; orphaned entries for dropped
525 /// cells are harmless (the handle is never observed again).
526 pub(crate) last_seen: std::collections::HashMap<*const SharedCellInner, u64>,
527 /// Per-node cone metadata for cell-bound input deps. Lazy:
528 /// `None` until first `check_cell_clean` for that node;
529 /// then built once and reused. Cleared in bulk on any
530 /// attach/detach of shared cells.
531 pub(crate) cell_cones: Vec<Option<CellCone>>,
532}
533
534// `last_seen` keys are `*const SharedCellInner` raw pointers,
535// which Rust treats as non-Send/non-Sync. The pointers are
536// only compared by identity (never dereferenced) and each
537// `EngineCore` is owned by exactly one fiber, so the
538// non-thread-safe pointer keys are sound. The Send/Sync
539// markers here cover that gap explicitly.
540unsafe impl Send for EngineCore {}
541unsafe impl Sync for EngineCore {}
542
543impl EngineCore {
544 /// Allocate the next bit position from this scope's
545 /// intent-dirty vector for a newly-created cell. Returns
546 /// the specific word's `Arc<AtomicU64>` plus the bit
547 /// position within that word. Grows
548 /// [`Self::scope_intent_words`] on demand — each new word
549 /// is a freshly-allocated `Arc<AtomicU64>` so existing
550 /// cells' references stay stable.
551 pub(crate) fn allocate_cell_bit(&mut self) -> (Arc<AtomicU64>, u8) {
552 let bit = self.next_cell_bit;
553 let word_idx = (bit / 64) as usize;
554 let bit_in_word = (bit % 64) as u8;
555 while self.scope_intent_words.len() <= word_idx {
556 self.scope_intent_words.push(Arc::new(AtomicU64::new(0)));
557 }
558 let word = self.scope_intent_words[word_idx].clone();
559 self.next_cell_bit += 1;
560 (word, bit_in_word)
561 }
562
563 /// Construct a new `SharedCell` bound to this scope's
564 /// intent-dirty vector. Convenience wrapper that allocates
565 /// a fresh bit and builds the cell — every cell creation
566 /// site goes through here so the scope's bit allocator
567 /// stays the single source of truth.
568 pub(crate) fn make_shared_cell(&mut self, initial: Value) -> SharedCell {
569 let (word, bit) = self.allocate_cell_bit();
570 Arc::new(SharedCellInner::new(initial, word, bit))
571 }
572}
573
574impl EngineCore {
575 /// Read an input slot's current value, transparent to whether
576 /// it's a plain slot or backed by a `SharedCell`. The
577 /// canonical read path used by both `eval_node` and
578 /// `PolydatState::get_input` — there's no separate "refresh" step
579 /// the caller must remember; the cell is queried on every
580 /// read.
581 ///
582 /// Cost: one Mutex lock per read on shared slots; a clone of
583 /// `inputs[idx]` on plain slots (Value's clone is cheap —
584 /// Arc-based for vectors, primitive copy otherwise).
585 #[inline]
586 pub(crate) fn read_input(&self, idx: usize) -> Value {
587 if let Some(cell) = self.shared_cells.get(idx).and_then(|c| c.as_ref()) {
588 return cell.value.lock().unwrap().clone();
589 }
590 self.inputs[idx].clone()
591 }
592
593 /// Build the cone metadata for `node_idx` — the per-scope
594 /// groups of cell-bound input dependencies, derived from
595 /// `program.input_provenance[node_idx]` and the cells
596 /// currently attached on this engine.
597 ///
598 /// Returns an empty `CellCone { groups: [] }` for nodes
599 /// with no cell-bound deps (the common case).
600 fn build_cell_cone(&self, program: &PolydatProgram, node_idx: usize) -> CellCone {
601 let empty = crate::kernel::ProvMask::empty();
602 let prov = program.input_provenance.get(node_idx).unwrap_or(&empty);
603 let mut groups: Vec<CellConeGroup> = Vec::new();
604 // Iterate set bits of `prov` directly: each bit is an
605 // input slot that flows into this node transitively.
606 for input_idx in prov.iter_ones() {
607 let Some(Some(cell)) = self.shared_cells.get(input_idx) else {
608 continue;
609 };
610 // Group by Arc-pointer identity of scope_intent_dirty.
611 let group_idx = groups
612 .iter()
613 .position(|g| Arc::ptr_eq(&g.intent_dirty, &cell.scope_intent_dirty));
614 let i = match group_idx {
615 Some(i) => i,
616 None => {
617 groups.push(CellConeGroup {
618 intent_dirty: cell.scope_intent_dirty.clone(),
619 interest_mask: 0,
620 cells: Vec::new(),
621 });
622 groups.len() - 1
623 }
624 };
625 groups[i].interest_mask |= 1u64 << cell.bit;
626 groups[i].cells.push(CellConeEntry {
627 bit: cell.bit,
628 input_slot: input_idx,
629 });
630 }
631 CellCone { groups }
632 }
633
634 /// Cross-fiber check: return `true` if this fiber's
635 /// `last_seen` is up-to-date for every cell in `node_idx`'s
636 /// cone (no cross-fiber writes since last observation).
637 /// Returns `false` if any cell's revision has advanced,
638 /// updating `last_seen` to reflect the new revisions in
639 /// preparation for the caller's re-evaluation.
640 ///
641 /// Per cross_fiber_invalidation.md §5: bulk-mask check
642 /// (one Acquire load + AND per scope group) early-outs
643 /// when nothing in the scope is dirty; per-cell drill-down
644 /// runs only on set bits.
645 fn check_cell_clean(&mut self, program: &PolydatProgram, node_idx: usize) -> bool {
646 // Lazy build the cone metadata.
647 if self.cell_cones.len() <= node_idx {
648 self.cell_cones.resize_with(node_idx + 1, || None);
649 }
650 if self.cell_cones[node_idx].is_none() {
651 let cone = self.build_cell_cone(program, node_idx);
652 self.cell_cones[node_idx] = Some(cone);
653 }
654
655 // First pass: walk the cone, collect mismatches. The
656 // immutable borrow of `self.cell_cones`,
657 // `self.shared_cells`, and `self.last_seen` coexist
658 // because they're disjoint fields of `self`.
659 let mut dirty: Vec<(*const SharedCellInner, u64, usize)> = Vec::new();
660 {
661 let cone = self.cell_cones[node_idx].as_ref().unwrap();
662 for group in &cone.groups {
663 let intent = group.intent_dirty.load(Ordering::Acquire);
664 let masked = intent & group.interest_mask;
665 if masked == 0 {
666 continue;
667 }
668 for entry in &group.cells {
669 if masked & (1u64 << entry.bit) == 0 {
670 continue;
671 }
672 let Some(Some(cell)) = self.shared_cells.get(entry.input_slot) else {
673 continue;
674 };
675 let r = cell.revision.load(Ordering::Acquire);
676 let ptr = Arc::as_ptr(cell);
677 let prev = self.last_seen.get(&ptr).copied().unwrap_or(0);
678 if r != prev {
679 dirty.push((ptr, r, entry.input_slot));
680 }
681 }
682 }
683 }
684 let clean = dirty.is_empty();
685 // Second pass: update last_seen for every cell whose
686 // revision we observed has advanced. Done in a
687 // separate pass to release the cone borrow above.
688 //
689 // Updating `last_seen` CONSUMES the dirty signal for this
690 // fiber, so the re-evaluation it triggers must reach every
691 // memoized node between the dirty slot and any consumer —
692 // not just the node that happened to check first. The
693 // caller only re-evaluates the CHECKED node; its recursive
694 // upstream walk re-checks each parent's own cone, which now
695 // reads the just-updated `last_seen` and comes back clean,
696 // leaving the intermediate buffers stale — the checked node
697 // then recomputes from stale parents (observed as a
698 // phase-poll predicate memoized at its pre-write value
699 // forever). Mirror `set_input`'s write-side rule on the
700 // read side: a detected cross-fiber write invalidates every
701 // node whose transitive input provenance covers the dirty
702 // slot.
703 if !clean {
704 // Exact multi-word mask: slots >= 64 invalidate too
705 // (the one-word form silently SKIPPED them — a latent
706 // under-invalidation on >64-input scopes).
707 let mut dirty_mask = crate::kernel::ProvMask::empty();
708 for (ptr, r, slot) in dirty {
709 self.last_seen.insert(ptr, r);
710 dirty_mask.set(slot);
711 }
712 for node_idx in 0..program.nodes.len() {
713 if program
714 .input_provenance
715 .get(node_idx)
716 .is_some_and(|prov| prov.intersects(&dirty_mask))
717 {
718 self.node_clean[node_idx] = false;
719 }
720 }
721 }
722 clean
723 }
724
725 /// Mark `cell_cones` as stale. Called after any change to
726 /// `shared_cells` that could affect the per-node cone
727 /// metadata (attach, detach). Next `check_cell_clean` on
728 /// any node will rebuild on demand.
729 pub(crate) fn invalidate_cell_cones(&mut self) {
730 for cone in self.cell_cones.iter_mut() {
731 *cone = None;
732 }
733 }
734
735 /// Evaluate a node by index. Shared by all engines.
736 /// Checks the clean flag, recursively evaluates upstream, gathers
737 /// inputs, calls node.eval(), marks clean.
738 pub fn eval_node(&mut self, program: &PolydatProgram, node_idx: usize) {
739 if self.node_clean[node_idx] {
740 // Memoization hit candidate — confirm cell-bound
741 // inputs in this node's cone are still at the
742 // revisions this fiber last observed. If any
743 // producer fiber has bumped a cell's revision since
744 // then, force a re-eval (the cache is stale even
745 // though `node_clean` is true) per
746 // cross_fiber_invalidation.md §5.
747 if self.check_cell_clean(program, node_idx) {
748 return;
749 }
750 self.node_clean[node_idx] = false;
751 }
752
753 let wiring = &program.wiring[node_idx];
754 for source in wiring.iter() {
755 if let WireSource::NodeOutput(upstream_idx, _) = source {
756 self.eval_node(program, *upstream_idx);
757 }
758 }
759
760 for (i, source) in wiring.iter().enumerate() {
761 self.input_scratch[i] = match source {
762 // `read_input` transparently reads the cell for
763 // `shared`-bound slots, so per-cycle eval picks
764 // up cross-kernel writes without any explicit
765 // refresh.
766 WireSource::Input(idx) => self.read_input(*idx),
767 WireSource::NodeOutput(upstream_idx, port_idx) => {
768 self.buffers[*upstream_idx][*port_idx].clone()
769 }
770 };
771 }
772
773 let input_count = wiring.len();
774
775 // SRD-74 Rule 1 — None propagation lifted to the kernel
776 // level. Any node whose inputs include `Value::None`
777 // emits `Value::None` on every output without invoking
778 // the node's `eval`. This holds the SQL-NULL / Rust
779 // `Option::?` propagation rule uniformly for ALL GK
780 // nodes, avoiding the dozens of duplicate per-node
781 // `if matches!(input, Value::None)` checks. Individual
782 // nodes (e.g. `Printf`) keep their checks redundant but
783 // harmless — the kernel guard fires first.
784 //
785 // Opt-out: nodes whose semantics explicitly consume
786 // `Value::None` (coalesce-style `default_or`, explicit
787 // optionality handlers per SRD-74 Rule 2) override
788 // `PolydatNode::accepts_none_inputs` to skip this guard. Such
789 // nodes handle `None` in their own `eval`.
790 let node_ref = &*program.nodes[node_idx];
791 if !node_ref.accepts_none_inputs()
792 && self.input_scratch[..input_count]
793 .iter()
794 .any(|v| matches!(v, Value::None))
795 {
796 for slot in &mut self.buffers[node_idx] {
797 *slot = Value::None;
798 }
799 self.node_clean[node_idx] = true;
800 return;
801 }
802
803 // Wrap the node's eval in catch_unwind so a node-level
804 // panic (e.g. `Value::as_u64` on a Str) can be re-raised
805 // with the diagnostic context the user actually needs:
806 // which node panicked, which output(s) it feeds, what
807 // the input values were, and where in the source the
808 // node came from. Without this, the fiber-level catcher
809 // sees only the bare message — "expected U64, got Str"
810 // — and the user has no way to find the offending
811 // binding short of bisecting the workload.
812 //
813 // Cost: one catch_unwind frame per slow-path node eval.
814 // The JIT path doesn't go through here. On the success
815 // path the frame is a few stack words; on the panic
816 // path it's strictly an improvement over what the
817 // user sees today.
818 //
819 // The capture guard suppresses the std panic hook for
820 // the duration: without it, the hook prints the BARE
821 // payload ("expected U64, got F64" + backtrace) at the
822 // original panic site, before enrichment exists, and
823 // that raw print is the loudest thing the user sees.
824 // Re-raising with `panic_any` (not `resume_unwind`)
825 // fires the hook again — now unsuppressed — so the one
826 // message that prints is the enriched one.
827 let guard = EvalPanicCaptureGuard::arm();
828 let payload = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
829 program.nodes[node_idx].eval_in(
830 &mut self.node_scratch[node_idx],
831 &self.input_scratch[..input_count],
832 &mut self.buffers[node_idx],
833 );
834 }));
835 drop(guard);
836 if let Err(e) = payload {
837 // A native cone re-raises its member's failure already
838 // enriched with the member's name, its inputs, and this
839 // program's context (A7); the cone itself is not a frame,
840 // so the report reads as it does on every other engine.
841 if program.nodes[node_idx].fusion_subgraph().is_some()
842 && e.downcast_ref::<String>()
843 .is_some_and(|s| s.contains("↳ in node"))
844 {
845 let enriched = *e.downcast::<String>().expect("checked above");
846 reraise_enriched(enriched);
847 }
848 let enriched =
849 enrich_eval_panic(e, program, node_idx, &self.input_scratch[..input_count]);
850 reraise_enriched(enriched);
851 }
852 self.node_clean[node_idx] = true;
853 }
854
855 /// Pull a named output.
856 pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
857 let (node_idx, port_idx) = *program
858 .output_map
859 .get(output_name)
860 .unwrap_or_else(|| panic!("unknown output variate: {output_name}"));
861 self.eval_node(program, node_idx);
862 // SRD-13f Push B.2: broadcast the freshly computed
863 // value through this output's cell (if attached) so
864 // descendant kernels that bound their matching input
865 // slot to the same cell see the current value on
866 // their next read.
867 if let Some(output_idx) = program.output_index(output_name)
868 && let Some(Some(cell)) = self.output_cells.get(output_idx)
869 {
870 let v = self.buffers[node_idx][port_idx].clone();
871 // `publish` does the mutex write + revision bump +
872 // intent-bit set in three Release stores so the
873 // descendant's cone walker observes the change on
874 // its next read (cross_fiber_invalidation.md §5).
875 cell.publish(v);
876 }
877 &self.buffers[node_idx][port_idx]
878 }
879
880 /// SRD-13f Push B.2 — allocate broadcast cells for every
881 /// output in `program`. Idempotent: if cells are already
882 /// allocated (size matches the program's output count),
883 /// the call is a no-op. Initial cell value is taken from
884 /// the current buffer (typically `Value::None` at
885 /// construction, before any pull has fired).
886 ///
887 /// Called from kernel constructors and from
888 /// `materialize_wiring_from_outer`-style operations that materialize
889 /// new descendants — the inner side needs the cell to
890 /// exist before it can attach to its input slot.
891 pub(crate) fn seed_output_cells(&mut self, program: &PolydatProgram) {
892 let n = program.output_names().len();
893 if self.output_cells.len() == n {
894 return;
895 }
896 // Two-pass to avoid borrowing `self` immutably (for
897 // buffer lookups) while also borrowing it mutably (for
898 // `make_shared_cell`). First collect initial values,
899 // then construct the cells.
900 let initials: Vec<Value> = (0..n)
901 .map(|i| {
902 let name = &program.output_list()[i].0;
903 let (node_idx, port_idx) = program.output_map[name];
904 // Defensive bounds-check: some construction paths
905 // (raw state, partial programs) may not populate
906 // buffers for every node referenced in the output
907 // map. Seed with `Value::None` rather than panic.
908 self.buffers
909 .get(node_idx)
910 .and_then(|b| b.get(port_idx))
911 .cloned()
912 .unwrap_or(Value::None)
913 })
914 .collect();
915 self.output_cells = initials
916 .into_iter()
917 .map(|init| Some(self.make_shared_cell(init)))
918 .collect();
919 }
920
921 /// Output broadcast cell for the named output, if seeded.
922 pub(crate) fn output_cell(&self, program: &PolydatProgram, name: &str) -> Option<SharedCell> {
923 let idx = program.output_index(name)?;
924 self.output_cells.get(idx).and_then(|c| c.clone())
925 }
926}
927
928// =================================================================
929// PolydatState: dependent-list engine (default, O(affected) invalidation)
930// =================================================================
931
932/// Polydat evaluation engine using precomputed per-input dependent lists.
933///
934/// On `set_input()`, only nodes that depend on the changed input
935/// are dirtied. O(affected_nodes) per input change.
936/// This is the interpreter's state, the default of its three; the
937/// default engine for engine-less entry points is the compiled P3 engine.
938pub struct PolydatState {
939 /// Shared evaluation core (buffers, clean flags, inputs).
940 pub core: EngineCore,
941 /// Per-input dependent node lists for O(affected) invalidation.
942 input_dependents: Vec<Vec<usize>>,
943 /// Indices of non-deterministic nodes (zero-provenance, no declared inputs).
944 ///
945 /// These nodes produce a different value on every evaluation (e.g.,
946 /// `counter()`, `current_epoch_millis()`). They are unconditionally
947 /// marked dirty on every `set_input()` call so they are never cached.
948 nondeterministic_nodes: Vec<usize>,
949}
950
951impl PolydatState {
952 /// Construct a PolydatState from its component parts.
953 pub(crate) fn from_parts(
954 core: EngineCore,
955 input_dependents: Vec<Vec<usize>>,
956 nondeterministic_nodes: Vec<usize>,
957 ) -> Self {
958 Self {
959 core,
960 input_dependents,
961 nondeterministic_nodes,
962 }
963 }
964
965 /// Set all coordinate inputs at once. Wraps each u64 as
966 /// `Value::U64` and sets them at indices 0..N with per-input
967 /// change detection.
968 pub fn set_inputs(&mut self, coords: &[u64]) {
969 self.write_coordinates(coords);
970 }
971
972 /// Write the coordinates: what construction does to seed a state's
973 /// folded constants. A host's write is [`Self::set_inputs`].
974 pub(crate) fn seed_inputs(&mut self, coords: &[u64]) {
975 self.write_coordinates(coords);
976 }
977
978 fn write_coordinates(&mut self, coords: &[u64]) {
979 for (i, &c) in coords.iter().enumerate().take(self.core.inputs.len()) {
980 self.core.inputs[i] = Value::U64(c);
981 // Unconditional invalidation: the write itself is the
982 // signal — see `set_input` for the rationale.
983 if i < self.input_dependents.len() {
984 for &node_idx in &self.input_dependents[i] {
985 self.core.node_clean[node_idx] = false;
986 }
987 }
988 }
989 // Non-deterministic nodes must always re-evaluate.
990 for &idx in &self.nondeterministic_nodes {
991 self.core.node_clean[idx] = false;
992 }
993 }
994
995 /// Set a single input by index, dirtying only dependent nodes.
996 ///
997 /// Single-register semantics: a cell-bound slot's only
998 /// register IS the cell — `set_input` writes through the
999 /// cell. A non-cell slot's register is the local
1000 /// `inputs[idx]` array. There's no second snapshot kept in
1001 /// lockstep with the cell; reads always go to whichever is
1002 /// the slot's register.
1003 ///
1004 /// Dependents-marking is the dependent-list invalidation
1005 /// strategy carried by `PolydatState`; it's the write-side
1006 /// half of the engine's dirty-tracking. Other engines
1007 /// (`RawState`, `ProvScanState`) implement different
1008 /// strategies — see their own `set_inputs` impls.
1009 pub fn set_input(&mut self, idx: usize, value: Value) {
1010 if let Some(cell) = self.core.shared_cells.get(idx).and_then(|c| c.as_ref()) {
1011 // Cell-bound slot: the cell is the register. We do
1012 // NOT mirror the value into `inputs[idx]`; that
1013 // array slot is unused for cell-bound inputs.
1014 //
1015 // `publish` does the mutex write + revision bump +
1016 // intent-bit set in three Release stores so the
1017 // any other fiber's cone walker observes the
1018 // change on its next read
1019 // (cross_fiber_invalidation.md §5).
1020 cell.publish(value);
1021 } else {
1022 self.core.inputs[idx] = value;
1023 }
1024 // Mark every transitive dependent dirty unconditionally.
1025 // The act of writing an input IS the invalidation
1026 // signal — we don't gate on value equality because (a)
1027 // structural equality on rich Value variants
1028 // (Json/Bytes/VecF32) is expensive enough to defeat
1029 // the purpose of the optimisation, and (b) a same-
1030 // value rewrite is still a legitimate "the upstream
1031 // owner asked for a re-evaluation" signal that
1032 // downstream side-effecting nodes (`log_*`, audit
1033 // emitters, time-stamped observers) MUST honour.
1034 let dirty_debug = nbrs_dirty_debug_enabled();
1035 if idx < self.input_dependents.len() {
1036 if dirty_debug {
1037 eprintln!(
1038 "DIRTY: set_input idx={idx} input_count={} dependents_for_idx={} \
1039 total_input_dependents_len={}",
1040 self.core.inputs.len(),
1041 self.input_dependents[idx].len(),
1042 self.input_dependents.len()
1043 );
1044 }
1045 for &node_idx in &self.input_dependents[idx] {
1046 self.core.node_clean[node_idx] = false;
1047 }
1048 } else if dirty_debug {
1049 eprintln!(
1050 "DIRTY: set_input idx={idx} OUT_OF_RANGE input_dependents_len={}",
1051 self.input_dependents.len()
1052 );
1053 }
1054 // Non-deterministic nodes must always re-evaluate.
1055 for &idx in &self.nondeterministic_nodes {
1056 self.core.node_clean[idx] = false;
1057 }
1058 }
1059
1060 /// Read the value of an input by index.
1061 ///
1062 /// Single-register read: cell-bound slots return the cell's
1063 /// current value; non-cell slots return the local register.
1064 /// One canonical value per slot, no stale snapshot.
1065 pub fn get_input(&self, idx: usize) -> Value {
1066 self.core.read_input(idx)
1067 }
1068
1069 /// Alias for [`Self::get_input`]; kept for legacy callers
1070 /// that picked the more explicit name. Both read the cell
1071 /// when one is attached.
1072 pub fn read_input_value(&self, idx: usize) -> Value {
1073 self.core.read_input(idx)
1074 }
1075
1076 /// Attach a `SharedCell` to an input slot.
1077 ///
1078 /// After this call the cell becomes the slot's sole
1079 /// register: reads via `read_input` go through the cell,
1080 /// `set_input` writes through the cell. The local
1081 /// `inputs[idx]` array entry for this slot is unused for
1082 /// cell-bound slots — there is no second register kept in
1083 /// lockstep.
1084 ///
1085 /// Dependents are dirtied because the slot's effective
1086 /// value just changed from the local default to whatever
1087 /// the cell currently holds.
1088 pub fn attach_shared_cell(&mut self, idx: usize, cell: SharedCell) {
1089 if idx >= self.core.shared_cells.len() {
1090 self.core.shared_cells.resize(idx + 1, None);
1091 }
1092 self.core.shared_cells[idx] = Some(cell);
1093 if idx < self.input_dependents.len() {
1094 for &node_idx in &self.input_dependents[idx] {
1095 self.core.node_clean[node_idx] = false;
1096 }
1097 }
1098 // Cone metadata depends on which slots have cells; the
1099 // new attachment invalidates any cached cone groups.
1100 // Next `check_cell_clean` per node rebuilds on demand
1101 // per cross_fiber_invalidation.md §3.1.
1102 self.core.invalidate_cell_cones();
1103 }
1104
1105 /// Returns the `SharedCell` attached to an input slot, if any.
1106 /// Used by `materialize_wiring_from_outer` to share an existing cell with
1107 /// inner kernels.
1108 pub fn shared_cell(&self, idx: usize) -> Option<SharedCell> {
1109 self.core.shared_cells.get(idx).and_then(|c| c.clone())
1110 }
1111
1112 /// Reset a range of inputs to their defaults. Used at stanza
1113 /// boundaries to prevent capture leakage across stanzas.
1114 /// `from_idx` is typically `coord_count` (skip coordinates,
1115 /// reset only capture inputs).
1116 ///
1117 /// Cell-bound slots are skipped: the cell is cross-kernel
1118 /// shared state with its own lifecycle (managed by the
1119 /// owning ancestor scope), and a stanza-local reset must
1120 /// not clobber other kernels' views.
1121 pub fn reset_inputs_from(&mut self, from_idx: usize) {
1122 for i in from_idx..self.core.inputs.len() {
1123 // Cell-bound slots: the cell is the register, owned
1124 // by the ancestor that declared `shared X := init`.
1125 // Don't touch.
1126 if self.core.shared_cells.get(i).is_some_and(|c| c.is_some()) {
1127 continue;
1128 }
1129 if self.core.inputs[i] != self.core.input_defaults[i] {
1130 self.core.inputs[i] = self.core.input_defaults[i].clone();
1131 if i < self.input_dependents.len() {
1132 for &node_idx in &self.input_dependents[i] {
1133 self.core.node_clean[node_idx] = false;
1134 }
1135 }
1136 }
1137 }
1138 }
1139
1140 /// Mark every node dirty and leave the inputs as they are: every
1141 /// node reruns at the next pull, as if the cycle had moved. What
1142 /// `Kernel::invalidate_all` means on every engine; a host that
1143 /// wants the inputs back at their defaults calls
1144 /// [`Self::reset_inputs_from`] as well.
1145 pub fn invalidate_all(&mut self) {
1146 self.core.node_clean.fill(false);
1147 }
1148
1149 /// Pull a named output variate from the program.
1150 pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
1151 self.core.pull(program, output_name)
1152 }
1153
1154 /// Pre-populate a node's output buffer slot and mark it clean,
1155 /// suppressing on-demand evaluation. Used by the scope-init
1156 /// pass (SRD 11 §"Init Binding Contract" Plan B) to seed
1157 /// per-fiber states with init binding values that the
1158 /// activation kernel already evaluated, so each fiber doesn't
1159 /// re-fire the eval at first pull.
1160 pub fn seed_node_buffer(&mut self, node_idx: usize, port_idx: usize, value: Value) {
1161 if node_idx >= self.core.buffers.len() {
1162 return;
1163 }
1164 if port_idx >= self.core.buffers[node_idx].len() {
1165 return;
1166 }
1167 self.core.buffers[node_idx][port_idx] = value;
1168 self.core.node_clean[node_idx] = true;
1169 }
1170
1171 /// Read a node's output buffer slot. Used by the scope-init
1172 /// pass to extract a pre-pulled init binding value from one
1173 /// state and seed it into another.
1174 pub fn node_buffer(&self, node_idx: usize, port_idx: usize) -> Option<&Value> {
1175 self.core
1176 .buffers
1177 .get(node_idx)
1178 .and_then(|ports| ports.get(port_idx))
1179 }
1180
1181 /// Pull an output by index (declaration order). Only evaluates
1182 /// the computation cone for this specific output.
1183 pub fn pull_by_index(&mut self, program: &PolydatProgram, output_idx: usize) -> &Value {
1184 let (node_idx, port_idx) = program.resolve_output_by_index(output_idx);
1185 self.core.eval_node(program, node_idx);
1186 &self.core.buffers[node_idx][port_idx]
1187 }
1188
1189 /// Pull all outputs in declaration order.
1190 pub fn pull_all<'a>(&'a mut self, program: &PolydatProgram) -> Vec<&'a Value> {
1191 for i in 0..program.output_count() {
1192 let (node_idx, _) = program.resolve_output_by_index(i);
1193 self.core.eval_node(program, node_idx);
1194 }
1195 (0..program.output_count())
1196 .map(|i| {
1197 let (ni, pi) = program.resolve_output_by_index(i);
1198 &self.core.buffers[ni][pi]
1199 })
1200 .collect()
1201 }
1202
1203 /// Create a memoized accessor for a named subset of outputs.
1204 /// Resolves names to indices once; subsequent access uses indices only.
1205 pub fn accessor(program: &PolydatProgram, names: &[&str]) -> OutputAccessor {
1206 let indices: Vec<usize> = names
1207 .iter()
1208 .filter_map(|n| program.output_index(n))
1209 .collect();
1210 OutputAccessor { indices }
1211 }
1212
1213 /// Evaluate a node by index (exposed for constant folding in PolydatProgram).
1214 pub(crate) fn eval_node_public(&mut self, program: &PolydatProgram, node_idx: usize) {
1215 self.core.eval_node(program, node_idx);
1216 }
1217}
1218
1219/// Memoized output accessor for a named subset of outputs.
1220///
1221/// Created once from output names via `PolydatState::accessor()`.
1222/// Subsequent pulls use pre-resolved indices — no name lookups.
1223pub struct OutputAccessor {
1224 indices: Vec<usize>,
1225}
1226
1227impl OutputAccessor {
1228 /// Pull all outputs in this accessor from the given state.
1229 pub fn pull_all<'a>(
1230 &self,
1231 state: &'a mut PolydatState,
1232 program: &PolydatProgram,
1233 ) -> Vec<&'a Value> {
1234 for &idx in &self.indices {
1235 let (node_idx, _) = program.resolve_output_by_index(idx);
1236 state.core.eval_node(program, node_idx);
1237 }
1238 self.indices
1239 .iter()
1240 .map(|&idx| {
1241 let (ni, pi) = program.resolve_output_by_index(idx);
1242 &state.core.buffers[ni][pi]
1243 })
1244 .collect()
1245 }
1246
1247 /// Number of outputs in this accessor.
1248 pub fn len(&self) -> usize {
1249 self.indices.len()
1250 }
1251
1252 /// Whether this accessor has no outputs.
1253 pub fn is_empty(&self) -> bool {
1254 self.indices.is_empty()
1255 }
1256}
1257
1258// =================================================================
1259// RawState: no provenance engine (all nodes dirty every eval)
1260// =================================================================
1261
1262/// Polydat evaluation engine with no provenance. Every `set_inputs()`
1263/// marks all nodes dirty. Baseline for benchmarking provenance overhead.
1264pub struct RawState {
1265 /// Shared evaluation core.
1266 pub core: EngineCore,
1267}
1268
1269impl RawState {
1270 /// Set new input values and mark all nodes dirty (no provenance check).
1271 pub fn set_inputs(&mut self, coords: &[u64]) {
1272 for (i, &c) in coords.iter().enumerate().take(self.core.inputs.len()) {
1273 self.core.inputs[i] = Value::U64(c);
1274 }
1275 self.core.node_clean.fill(false);
1276 }
1277
1278 /// Pull a named output variate from the program.
1279 pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
1280 self.core.pull(program, output_name)
1281 }
1282}
1283
1284// =================================================================
1285// ProvScanState: provenance-scan engine (O(all) invalidation)
1286// =================================================================
1287
1288/// Polydat evaluation engine using provenance bitmask scanning.
1289///
1290/// On `set_inputs()`, scans ALL nodes and checks each node's
1291/// provenance bitmask against the changed-inputs mask.
1292/// O(all_nodes) per input change regardless of how many changed.
1293pub struct ProvScanState {
1294 /// Shared evaluation core.
1295 pub core: EngineCore,
1296 input_provenance: Vec<crate::kernel::ProvMask>,
1297 /// Indices of non-deterministic nodes.
1298 nondeterministic_nodes: Vec<usize>,
1299}
1300
1301impl ProvScanState {
1302 /// Construct a ProvScanState from its component parts.
1303 pub(crate) fn from_parts(
1304 core: EngineCore,
1305 input_provenance: Vec<crate::kernel::ProvMask>,
1306 nondeterministic_nodes: Vec<usize>,
1307 ) -> Self {
1308 Self {
1309 core,
1310 input_provenance,
1311 nondeterministic_nodes,
1312 }
1313 }
1314
1315 /// Set new input values and invalidate affected nodes.
1316 pub fn set_inputs(&mut self, coords: &[u64]) {
1317 let mut mask = crate::kernel::ProvMask::empty();
1318 for (i, &c) in coords.iter().enumerate().take(self.core.inputs.len()) {
1319 self.core.inputs[i] = Value::U64(c);
1320 // Unconditional: writing the input IS the
1321 // invalidation signal regardless of value equality.
1322 mask.set(i);
1323 }
1324 if !mask.is_zero() {
1325 for (i, clean) in self.core.node_clean.iter_mut().enumerate() {
1326 if *clean && self.input_provenance[i].intersects(&mask) {
1327 *clean = false;
1328 }
1329 }
1330 }
1331 for &idx in &self.nondeterministic_nodes {
1332 self.core.node_clean[idx] = false;
1333 }
1334 }
1335
1336 /// Pull a named output variate from the program.
1337 pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
1338 self.core.pull(program, output_name)
1339 }
1340}