use std::sync::{Arc, Mutex, OnceLock};
use std::sync::atomic::{AtomicU64, Ordering};
use crate::ast::Value;
use super::WireSource;
use super::program::PolydatProgram;
fn nbrs_dirty_debug_enabled() -> bool {
static FLAG: OnceLock<bool> = OnceLock::new();
*FLAG.get_or_init(|| std::env::var("NBRS_DIRTY_DEBUG").is_ok())
}
pub struct SharedCellInner {
pub value: Mutex<Value>,
pub revision: AtomicU64,
pub scope_intent_dirty: Arc<AtomicU64>,
pub bit: u8,
}
impl std::fmt::Debug for SharedCellInner {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SharedCellInner")
.field("revision", &self.revision.load(Ordering::Relaxed))
.field("bit", &self.bit)
.finish_non_exhaustive()
}
}
impl SharedCellInner {
pub fn new(
initial: Value,
scope_intent_dirty: Arc<AtomicU64>,
bit: u8,
) -> Self {
debug_assert!(
bit < 64,
"bit-within-word {bit} must be < 64; the allocator splits >64-bit \
scope vectors across multiple words"
);
Self {
value: Mutex::new(initial),
revision: AtomicU64::new(0),
scope_intent_dirty,
bit,
}
}
pub fn publish(&self, value: Value) {
{
let mut guard = self.value.lock().unwrap();
*guard = value;
}
self.revision.fetch_add(1, Ordering::Release);
self.scope_intent_dirty
.fetch_or(1u64 << self.bit, Ordering::Release);
}
pub fn snapshot(&self) -> (Value, u64) {
let value = self.value.lock().unwrap().clone();
let revision = self.revision.load(Ordering::Acquire);
(value, revision)
}
}
pub type SharedCell = Arc<SharedCellInner>;
#[derive(Debug, Default, Clone)]
pub(crate) struct CellCone {
pub(crate) groups: Vec<CellConeGroup>,
}
#[derive(Debug, Clone)]
pub(crate) struct CellConeGroup {
pub(crate) intent_dirty: Arc<AtomicU64>,
pub(crate) interest_mask: u64,
pub(crate) cells: Vec<CellConeEntry>,
}
#[derive(Debug, Clone, Copy)]
pub(crate) struct CellConeEntry {
pub(crate) bit: u8,
pub(crate) input_slot: usize,
}
#[derive(Clone, Debug)]
pub struct SharedCellEntry {
pub name: String,
pub port_type: crate::ast::PortType,
pub cell: SharedCell,
}
static PANIC_REPORTING_DOWNSTREAM: std::sync::atomic::AtomicBool =
std::sync::atomic::AtomicBool::new(false);
pub fn set_panic_reporting_downstream(on: bool) {
PANIC_REPORTING_DOWNSTREAM.store(on, std::sync::atomic::Ordering::Relaxed);
}
thread_local! {
static EVAL_PANIC_CAPTURE: std::cell::Cell<bool> =
const { std::cell::Cell::new(false) };
static EVAL_PANIC_LOCATION: std::cell::RefCell<Option<String>> =
const { std::cell::RefCell::new(None) };
static RERAISE_SHORT: std::cell::Cell<bool> =
const { std::cell::Cell::new(false) };
}
fn install_eval_panic_hook() {
static HOOK: std::sync::Once = std::sync::Once::new();
HOOK.call_once(|| {
let prev = std::panic::take_hook();
std::panic::set_hook(Box::new(move |info| {
if EVAL_PANIC_CAPTURE.with(|c| c.get()) {
let loc = info.location().map(|l| l.to_string());
EVAL_PANIC_LOCATION.with(|slot| *slot.borrow_mut() = loc);
} else if RERAISE_SHORT.with(|c| c.replace(false)) {
let first = info
.payload()
.downcast_ref::<String>()
.map(String::as_str)
.and_then(|m| m.lines().next())
.unwrap_or("<non-string panic payload>");
eprintln!("op eval panic (detail in phase errors): {first}");
} else {
prev(info);
}
}));
});
}
struct EvalPanicCaptureGuard {
prev: bool,
}
impl EvalPanicCaptureGuard {
fn arm() -> Self {
install_eval_panic_hook();
let prev = EVAL_PANIC_CAPTURE.with(|c| c.replace(true));
EVAL_PANIC_LOCATION.with(|slot| slot.borrow_mut().take());
Self { prev }
}
}
impl Drop for EvalPanicCaptureGuard {
fn drop(&mut self) {
EVAL_PANIC_CAPTURE.with(|c| c.set(self.prev));
}
}
fn enrich_eval_panic(
payload: Box<dyn std::any::Any + Send>,
program: &PolydatProgram,
node_idx: usize,
inputs: &[Value],
) -> String {
let original = payload
.downcast_ref::<&'static str>().map(|s| (*s).to_string())
.or_else(|| payload.downcast_ref::<String>().cloned())
.unwrap_or_else(|| "<non-string panic payload>".into());
let location_line = if original.contains("↳ in node") {
String::new()
} else {
EVAL_PANIC_LOCATION
.with(|slot| slot.borrow_mut().take())
.map(|loc| format!("\n ↳ panicked at {loc}"))
.unwrap_or_default()
};
let node_name = program.nodes.get(node_idx)
.map(|n| n.meta().name.to_string())
.unwrap_or_else(|| format!("<unknown node #{node_idx}>"));
let mut output_names: Vec<&str> = program.output_map_iter()
.filter_map(|(name, (n_idx, _))| {
if *n_idx == node_idx { Some(name.as_str()) } else { None }
})
.collect();
output_names.sort();
let outputs_label = if output_names.is_empty() {
"no declared output".to_string()
} else {
format!("output{} {}",
if output_names.len() == 1 { "" } else { "s" },
output_names.join(", "))
};
let mut input_label = String::new();
for (i, v) in inputs.iter().enumerate() {
if i > 0 { input_label.push_str(", "); }
input_label.push_str(&format!("[{i}]={}", format_value_for_diag(v)));
}
format!(
"{original}{location_line}\n ↳ in node `{node_name}` ({outputs_label}) \
while evaluating {context}\n \
↳ inputs: [{input_label}]",
context = program.context(),
)
}
fn format_value_for_diag(v: &Value) -> String {
match v {
Value::U64(n) => format!("U64({n})"),
Value::F64(n) => format!("F64({n})"),
Value::Bool(b) => format!("Bool({b})"),
Value::Str(s) => {
let trimmed: String = s.chars().take(40).collect();
if s.chars().count() > 40 {
format!("Str({trimmed:?}…)")
} else {
format!("Str({trimmed:?})")
}
}
Value::None => "None".to_string(),
other => format!("{:?}", other.port_type()),
}
}
pub struct EngineCore {
pub(crate) buffers: Vec<Vec<Value>>,
pub(crate) node_clean: Vec<bool>,
pub(crate) inputs: Vec<Value>,
pub(crate) input_defaults: Vec<Value>,
pub(crate) shared_cells: Vec<Option<SharedCell>>,
pub(crate) output_cells: Vec<Option<SharedCell>>,
pub(crate) input_scratch: Vec<Value>,
pub(crate) scope_intent_words: Vec<Arc<AtomicU64>>,
pub(crate) next_cell_bit: u32,
pub(crate) last_seen: std::collections::HashMap<*const SharedCellInner, u64>,
pub(crate) cell_cones: Vec<Option<CellCone>>,
}
unsafe impl Send for EngineCore {}
unsafe impl Sync for EngineCore {}
impl EngineCore {
pub(crate) fn allocate_cell_bit(&mut self) -> (Arc<AtomicU64>, u8) {
let bit = self.next_cell_bit;
let word_idx = (bit / 64) as usize;
let bit_in_word = (bit % 64) as u8;
while self.scope_intent_words.len() <= word_idx {
self.scope_intent_words.push(Arc::new(AtomicU64::new(0)));
}
let word = self.scope_intent_words[word_idx].clone();
self.next_cell_bit += 1;
(word, bit_in_word)
}
pub(crate) fn make_shared_cell(&mut self, initial: Value) -> SharedCell {
let (word, bit) = self.allocate_cell_bit();
Arc::new(SharedCellInner::new(initial, word, bit))
}
}
impl EngineCore {
#[inline]
pub(crate) fn read_input(&self, idx: usize) -> Value {
if let Some(cell) = self.shared_cells.get(idx).and_then(|c| c.as_ref()) {
return cell.value.lock().unwrap().clone();
}
self.inputs[idx].clone()
}
fn build_cell_cone(&self, program: &PolydatProgram, node_idx: usize) -> CellCone {
let empty = crate::kernel::ProvMask::empty();
let prov = program.input_provenance
.get(node_idx)
.unwrap_or(&empty);
let mut groups: Vec<CellConeGroup> = Vec::new();
for input_idx in prov.iter_ones() {
let Some(Some(cell)) = self.shared_cells.get(input_idx) else { continue; };
let group_idx = groups.iter()
.position(|g| Arc::ptr_eq(&g.intent_dirty, &cell.scope_intent_dirty));
let i = match group_idx {
Some(i) => i,
None => {
groups.push(CellConeGroup {
intent_dirty: cell.scope_intent_dirty.clone(),
interest_mask: 0,
cells: Vec::new(),
});
groups.len() - 1
}
};
groups[i].interest_mask |= 1u64 << cell.bit;
groups[i].cells.push(CellConeEntry {
bit: cell.bit,
input_slot: input_idx,
});
}
CellCone { groups }
}
fn check_cell_clean(
&mut self,
program: &PolydatProgram,
node_idx: usize,
) -> bool {
if self.cell_cones.len() <= node_idx {
self.cell_cones.resize_with(node_idx + 1, || None);
}
if self.cell_cones[node_idx].is_none() {
let cone = self.build_cell_cone(program, node_idx);
self.cell_cones[node_idx] = Some(cone);
}
let mut dirty: Vec<(*const SharedCellInner, u64, usize)> = Vec::new();
{
let cone = self.cell_cones[node_idx].as_ref().unwrap();
for group in &cone.groups {
let intent = group.intent_dirty.load(Ordering::Acquire);
let masked = intent & group.interest_mask;
if masked == 0 { continue; }
for entry in &group.cells {
if masked & (1u64 << entry.bit) == 0 { continue; }
let Some(Some(cell)) = self.shared_cells.get(entry.input_slot) else {
continue;
};
let r = cell.revision.load(Ordering::Acquire);
let ptr = Arc::as_ptr(cell);
let prev = self.last_seen.get(&ptr).copied().unwrap_or(0);
if r != prev {
dirty.push((ptr, r, entry.input_slot));
}
}
}
}
let clean = dirty.is_empty();
if !clean {
let mut dirty_mask = crate::kernel::ProvMask::empty();
for (ptr, r, slot) in dirty {
self.last_seen.insert(ptr, r);
dirty_mask.set(slot);
}
for node_idx in 0..program.nodes.len() {
if program.input_provenance
.get(node_idx)
.is_some_and(|prov| prov.intersects(&dirty_mask))
{
self.node_clean[node_idx] = false;
}
}
}
clean
}
pub(crate) fn invalidate_cell_cones(&mut self) {
for cone in self.cell_cones.iter_mut() {
*cone = None;
}
}
pub fn eval_node(&mut self, program: &PolydatProgram, node_idx: usize) {
if self.node_clean[node_idx] {
if self.check_cell_clean(program, node_idx) {
return;
}
self.node_clean[node_idx] = false;
}
let wiring = &program.wiring[node_idx];
for source in wiring.iter() {
if let WireSource::NodeOutput(upstream_idx, _) = source {
self.eval_node(program, *upstream_idx);
}
}
for (i, source) in wiring.iter().enumerate() {
self.input_scratch[i] = match source {
WireSource::Input(idx) => self.read_input(*idx),
WireSource::NodeOutput(upstream_idx, port_idx) => {
self.buffers[*upstream_idx][*port_idx].clone()
}
};
}
let input_count = wiring.len();
let node_ref = &*program.nodes[node_idx];
if !node_ref.accepts_none_inputs()
&& self.input_scratch[..input_count]
.iter()
.any(|v| matches!(v, Value::None))
{
for slot in &mut self.buffers[node_idx] {
*slot = Value::None;
}
self.node_clean[node_idx] = true;
return;
}
let guard = EvalPanicCaptureGuard::arm();
let payload = std::panic::catch_unwind(
std::panic::AssertUnwindSafe(|| {
program.nodes[node_idx].eval(
&self.input_scratch[..input_count],
&mut self.buffers[node_idx],
);
})
);
drop(guard);
if let Err(e) = payload {
let enriched = enrich_eval_panic(
e, program, node_idx,
&self.input_scratch[..input_count],
);
if PANIC_REPORTING_DOWNSTREAM.load(std::sync::atomic::Ordering::Relaxed) {
RERAISE_SHORT.with(|c| c.set(true));
}
std::panic::panic_any(enriched);
}
self.node_clean[node_idx] = true;
}
pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
let (node_idx, port_idx) = *program.output_map
.get(output_name)
.unwrap_or_else(|| panic!("unknown output variate: {output_name}"));
self.eval_node(program, node_idx);
if let Some(output_idx) = program.output_index(output_name)
&& let Some(Some(cell)) = self.output_cells.get(output_idx)
{
let v = self.buffers[node_idx][port_idx].clone();
cell.publish(v);
}
&self.buffers[node_idx][port_idx]
}
pub(crate) fn seed_output_cells(&mut self, program: &PolydatProgram) {
let n = program.output_names().len();
if self.output_cells.len() == n { return; }
let initials: Vec<Value> = (0..n).map(|i| {
let name = &program.output_list()[i].0;
let (node_idx, port_idx) = program.output_map[name];
self.buffers.get(node_idx)
.and_then(|b| b.get(port_idx))
.cloned()
.unwrap_or(Value::None)
}).collect();
self.output_cells = initials.into_iter()
.map(|init| Some(self.make_shared_cell(init)))
.collect();
}
pub(crate) fn output_cell(&self, program: &PolydatProgram, name: &str) -> Option<SharedCell> {
let idx = program.output_index(name)?;
self.output_cells.get(idx).and_then(|c| c.clone())
}
}
pub struct PolydatState {
pub core: EngineCore,
input_dependents: Vec<Vec<usize>>,
nondeterministic_nodes: Vec<usize>,
}
impl PolydatState {
pub(crate) fn from_parts(
core: EngineCore,
input_dependents: Vec<Vec<usize>>,
nondeterministic_nodes: Vec<usize>,
) -> Self {
Self { core, input_dependents, nondeterministic_nodes }
}
pub fn set_inputs(&mut self, coords: &[u64]) {
for (i, &c) in coords.iter().enumerate().take(self.core.inputs.len()) {
self.core.inputs[i] = Value::U64(c);
if i < self.input_dependents.len() {
for &node_idx in &self.input_dependents[i] {
self.core.node_clean[node_idx] = false;
}
}
}
for &idx in &self.nondeterministic_nodes {
self.core.node_clean[idx] = false;
}
}
pub fn set_input(&mut self, idx: usize, value: Value) {
if let Some(cell) = self.core.shared_cells.get(idx).and_then(|c| c.as_ref()) {
cell.publish(value);
} else {
self.core.inputs[idx] = value;
}
let dirty_debug = nbrs_dirty_debug_enabled();
if idx < self.input_dependents.len() {
if dirty_debug {
eprintln!(
"DIRTY: set_input idx={idx} input_count={} dependents_for_idx={} \
total_input_dependents_len={}",
self.core.inputs.len(),
self.input_dependents[idx].len(),
self.input_dependents.len()
);
}
for &node_idx in &self.input_dependents[idx] {
self.core.node_clean[node_idx] = false;
}
} else if dirty_debug {
eprintln!(
"DIRTY: set_input idx={idx} OUT_OF_RANGE input_dependents_len={}",
self.input_dependents.len()
);
}
for &idx in &self.nondeterministic_nodes {
self.core.node_clean[idx] = false;
}
}
pub fn get_input(&self, idx: usize) -> Value {
self.core.read_input(idx)
}
pub fn read_input_value(&self, idx: usize) -> Value {
self.core.read_input(idx)
}
pub fn attach_shared_cell(&mut self, idx: usize, cell: SharedCell) {
if idx >= self.core.shared_cells.len() {
self.core.shared_cells.resize(idx + 1, None);
}
self.core.shared_cells[idx] = Some(cell);
if idx < self.input_dependents.len() {
for &node_idx in &self.input_dependents[idx] {
self.core.node_clean[node_idx] = false;
}
}
self.core.invalidate_cell_cones();
}
pub fn shared_cell(&self, idx: usize) -> Option<SharedCell> {
self.core.shared_cells.get(idx).and_then(|c| c.clone())
}
pub fn reset_inputs_from(&mut self, from_idx: usize) {
for i in from_idx..self.core.inputs.len() {
if self.core.shared_cells.get(i).is_some_and(|c| c.is_some()) {
continue;
}
if self.core.inputs[i] != self.core.input_defaults[i] {
self.core.inputs[i] = self.core.input_defaults[i].clone();
if i < self.input_dependents.len() {
for &node_idx in &self.input_dependents[i] {
self.core.node_clean[node_idx] = false;
}
}
}
}
}
pub fn invalidate_all(&mut self) {
self.core.inputs.clone_from_slice(&self.core.input_defaults);
for clean in &mut self.core.node_clean {
*clean = false;
}
}
pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
self.core.pull(program, output_name)
}
pub fn seed_node_buffer(&mut self, node_idx: usize, port_idx: usize, value: Value) {
if node_idx >= self.core.buffers.len() { return; }
if port_idx >= self.core.buffers[node_idx].len() { return; }
self.core.buffers[node_idx][port_idx] = value;
self.core.node_clean[node_idx] = true;
}
pub fn node_buffer(&self, node_idx: usize, port_idx: usize) -> Option<&Value> {
self.core.buffers.get(node_idx)
.and_then(|ports| ports.get(port_idx))
}
pub fn pull_by_index(&mut self, program: &PolydatProgram, output_idx: usize) -> &Value {
let (node_idx, port_idx) = program.resolve_output_by_index(output_idx);
self.core.eval_node(program, node_idx);
&self.core.buffers[node_idx][port_idx]
}
pub fn pull_all<'a>(&'a mut self, program: &PolydatProgram) -> Vec<&'a Value> {
for i in 0..program.output_count() {
let (node_idx, _) = program.resolve_output_by_index(i);
self.core.eval_node(program, node_idx);
}
(0..program.output_count())
.map(|i| {
let (ni, pi) = program.resolve_output_by_index(i);
&self.core.buffers[ni][pi]
})
.collect()
}
pub fn accessor(program: &PolydatProgram, names: &[&str]) -> OutputAccessor {
let indices: Vec<usize> = names.iter()
.filter_map(|n| program.output_index(n))
.collect();
OutputAccessor { indices }
}
pub(crate) fn eval_node_public(&mut self, program: &PolydatProgram, node_idx: usize) {
self.core.eval_node(program, node_idx);
}
}
pub struct OutputAccessor {
indices: Vec<usize>,
}
impl OutputAccessor {
pub fn pull_all<'a>(&self, state: &'a mut PolydatState, program: &PolydatProgram) -> Vec<&'a Value> {
for &idx in &self.indices {
let (node_idx, _) = program.resolve_output_by_index(idx);
state.core.eval_node(program, node_idx);
}
self.indices.iter()
.map(|&idx| {
let (ni, pi) = program.resolve_output_by_index(idx);
&state.core.buffers[ni][pi]
})
.collect()
}
pub fn len(&self) -> usize {
self.indices.len()
}
pub fn is_empty(&self) -> bool {
self.indices.is_empty()
}
}
pub struct RawState {
pub core: EngineCore,
}
impl RawState {
pub fn set_inputs(&mut self, coords: &[u64]) {
for (i, &c) in coords.iter().enumerate().take(self.core.inputs.len()) {
self.core.inputs[i] = Value::U64(c);
}
for clean in &mut self.core.node_clean {
*clean = false;
}
}
pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
self.core.pull(program, output_name)
}
}
pub struct ProvScanState {
pub core: EngineCore,
input_provenance: Vec<crate::kernel::ProvMask>,
nondeterministic_nodes: Vec<usize>,
}
impl ProvScanState {
pub(crate) fn from_parts(
core: EngineCore,
input_provenance: Vec<crate::kernel::ProvMask>,
nondeterministic_nodes: Vec<usize>,
) -> Self {
Self { core, input_provenance, nondeterministic_nodes }
}
pub fn set_inputs(&mut self, coords: &[u64]) {
let mut mask = crate::kernel::ProvMask::empty();
for (i, &c) in coords.iter().enumerate().take(self.core.inputs.len()) {
self.core.inputs[i] = Value::U64(c);
mask.set(i);
}
if !mask.is_zero() {
for (i, clean) in self.core.node_clean.iter_mut().enumerate() {
if *clean && self.input_provenance[i].intersects(&mask) {
*clean = false;
}
}
}
for &idx in &self.nondeterministic_nodes {
self.core.node_clean[idx] = false;
}
}
pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
self.core.pull(program, output_name)
}
}
#[cfg(test)]
mod panic_enrichment_tests {
use crate::dsl::compile::compile_polydat_with_libs;
#[test]
fn type_mismatch_panic_carries_node_and_output_context() {
let mut k = compile_polydat_with_libs(
"extern x: u64\n\
doubled := mul(x, 2)\n",
None, vec![], &[], false, "test_workload",
).expect("compile");
let idx = k.program().find_input("x").unwrap();
k.state().set_input(idx, crate::ast::Value::Str("oops".into()));
let result = std::panic::catch_unwind(
std::panic::AssertUnwindSafe(|| { k.pull("doubled"); })
);
let err = result.expect_err("pull should panic on type mismatch");
let msg = err.downcast_ref::<String>().cloned()
.or_else(|| err.downcast_ref::<&'static str>().map(|s| (*s).to_string()))
.expect("panic payload should be a String");
assert!(msg.contains("expected U64"),
"missing original panic body in: {msg}");
assert!(msg.contains("`mul`"),
"missing node name in enriched message: {msg}");
assert!(msg.contains("doubled"),
"missing output binding in enriched message: {msg}");
assert!(msg.contains("test_workload"),
"missing program context in enriched message: {msg}");
assert!(msg.contains("\"oops\""),
"missing input snapshot in enriched message: {msg}");
assert!(msg.contains("panicked at") && msg.contains("ast.rs"),
"missing original panic location in enriched message: {msg}");
eprintln!("== enriched message ==\n{msg}\n======================");
}
}