use std::collections::HashMap;
use crate::ast::SlotShape;
use crate::ast::{CompiledU64Op, PolydatNode};
use crate::kernel::WireSource;
#[cfg(feature = "jit")]
use crate::compile::jit::{self, JitOp};
enum HybridStep {
#[cfg(feature = "jit")]
Jit(JitSegment),
Closure(ClosureStep),
}
#[cfg(feature = "jit")]
struct JitSegment {
code_fn: crate::compile::jit::NativeFn,
_module: crate::compile::jit::JitCode,
fallible: bool,
input_slots: Vec<usize>,
output_slots: Vec<usize>,
nodes: Vec<usize>,
}
impl HybridStep {
fn input_slots(&self) -> &[usize] {
match self {
#[cfg(feature = "jit")]
HybridStep::Jit(seg) => &seg.input_slots,
HybridStep::Closure(cs) => &cs.input_slots,
}
}
fn output_slots(&self) -> &[usize] {
match self {
#[cfg(feature = "jit")]
HybridStep::Jit(seg) => &seg.output_slots,
HybridStep::Closure(cs) => &cs.output_slots,
}
}
fn accepts_none(&self) -> bool {
match self {
#[cfg(feature = "jit")]
HybridStep::Jit(_) => false,
HybridStep::Closure(cs) => cs.accepts_none,
}
}
#[cfg_attr(not(feature = "jit"), allow(unused_variables))]
fn failing_node(&self, buffer: &[u64], tracker: usize) -> usize {
match self {
#[cfg(feature = "jit")]
HybridStep::Jit(seg) => seg
.nodes
.get(buffer[tracker] as usize)
.copied()
.unwrap_or(usize::MAX),
HybridStep::Closure(cs) => cs.node,
}
}
}
enum ClosureOp {
U64(CompiledU64Op),
Slot(crate::ast::CompiledSlotOp),
}
struct ClosureStep {
op: ClosureOp,
input_slots: Vec<usize>,
output_slots: Vec<usize>,
scratch_range: (usize, usize),
accepts_none: bool,
node: usize,
}
type ResolvedOutput = (usize, crate::ast::PortType, Option<std::sync::Arc<[usize]>>);
struct HybridCore {
buffer: Vec<u64>,
coord_count: usize,
steps: std::sync::Arc<Vec<HybridStep>>,
output_map: HashMap<String, usize>,
gather_buf: Vec<u64>,
scatter_buf: Vec<u64>,
scratch: Vec<crate::ast::ScratchBuf>,
ref_slots: Vec<bool>,
ref_scratch: Vec<(usize, usize)>,
output_types: HashMap<String, crate::ast::PortType>,
externs: crate::compile::externs::Externs,
traversals: std::sync::Arc<[crate::dsl::traversal::Traversal]>,
resolved_outputs: Vec<Option<ResolvedOutput>>,
_nodes: std::sync::Arc<Vec<Box<dyn PolydatNode>>>,
drive: crate::compile::Drive,
none: Vec<bool>,
ran: Vec<u64>,
epoch: u64,
all_ran: bool,
clean: Vec<bool>,
use_clean: bool,
plan: std::sync::Arc<crate::compile::Invalidation>,
volatile: std::sync::Arc<[bool]>,
side: std::sync::Arc<[bool]>,
slot_step: std::sync::Arc<[Option<usize>]>,
sites: std::sync::Arc<crate::compile::Attribution>,
cur_step: usize,
tracker: usize,
all: std::sync::Arc<[usize]>,
dirty: std::sync::Arc<[Vec<usize>]>,
any_none: bool,
volatile_steps: std::sync::Arc<[usize]>,
}
impl Clone for HybridCore {
fn clone(&self) -> Self {
let mut core = HybridCore {
buffer: self.buffer.clone(),
coord_count: self.coord_count,
steps: self.steps.clone(),
output_map: self.output_map.clone(),
gather_buf: self.gather_buf.clone(),
scatter_buf: self.scatter_buf.clone(),
scratch: self.scratch.clone(),
ref_slots: self.ref_slots.clone(),
ref_scratch: self.ref_scratch.clone(),
output_types: self.output_types.clone(),
externs: self.externs.clone(),
traversals: self.traversals.clone(),
resolved_outputs: self.resolved_outputs.clone(),
_nodes: self._nodes.clone(),
drive: self.drive.clone(),
none: self.none.clone(),
ran: self.ran.clone(),
epoch: self.epoch,
all_ran: self.all_ran,
clean: self.clean.clone(),
use_clean: self.use_clean,
plan: self.plan.clone(),
volatile: self.volatile.clone(),
side: self.side.clone(),
slot_step: self.slot_step.clone(),
sites: self.sites.clone(),
cur_step: self.cur_step,
tracker: self.tracker,
all: self.all.clone(),
dirty: self.dirty.clone(),
any_none: self.any_none,
volatile_steps: self.volatile_steps.clone(),
};
core.republish_refs();
core
}
}
impl HybridCore {
fn republish_refs(&mut self) {
for &(slot, idx) in &self.ref_scratch {
let (p, l) = self.scratch[idx].ptr_len();
self.buffer[slot] = p;
self.buffer[slot + 1] = l;
}
self.externs.seed(&mut self.buffer, None);
}
}
impl HybridCore {
#[cfg(debug_assertions)]
fn validate_refs(&self) {
let skip = |slot: usize| {
self.none[slot]
|| matches!(self.slot_step.get(slot), Some(Some(step)) if self.ran[*step] == 0)
};
for &(slot, idx) in &self.ref_scratch {
if skip(slot) {
continue;
}
let (p, l) = self.scratch[idx].ptr_len();
assert!(
self.buffer[slot] == p && self.buffer[slot + 1] == l,
"S9 ref-validator: slot pair ({slot}, {}) = ({:#x}, {}) \
does not match scratch[{idx}] = ({p:#x}, {l})",
slot + 1,
self.buffer[slot],
self.buffer[slot + 1],
);
}
}
#[inline]
fn guard_ref_slot(&self, slot: usize) {
if self.ref_slots.get(slot).copied().unwrap_or(false) {
panic!(
"S2 pointer containment: slot {slot} is Ref2-colored; raw u64 readers \
would leak an interior address. Use the typed borrow-checked accessor \
(read_vec_*), the boundary decode, or copy out."
);
}
}
fn ref_entry(&self, slot: usize) -> &crate::ast::ScratchBuf {
match self.ref_scratch.iter().find(|(s, _)| *s == slot) {
Some(&(_, idx)) => &self.scratch[idx],
None if self.ref_slots.get(slot).copied().unwrap_or(false) => panic!(
"slot {slot} is a Ref pair owned by the CALLER (a kernel \
input) — read it on the caller side"
),
None => panic!("slot {slot} is not a Ref2-colored slot"),
}
}
}
impl HybridCore {
#[inline]
fn begin_epoch(&mut self) {
if self.externs.cells_dirty() {
self.externs.refresh_cells(&mut self.buffer);
}
self.dirty_refreshed();
self.epoch += 1;
self.all_ran = false;
for &i in self.volatile_steps.iter() {
self.clean[i] = false;
}
self.drive.stale = false;
}
#[cfg(feature = "jit")]
fn set_use_clean(&mut self, on: bool) {
self.use_clean = on;
let side = std::sync::Arc::clone(&self.side);
self.dirty = self
.plan
.input_dependents
.iter()
.map(|deps| {
if on {
deps.clone()
} else {
deps.iter().copied().filter(|&i| side[i]).collect()
}
})
.collect::<Vec<_>>()
.into();
}
#[inline]
fn dirty_refreshed(&mut self) {
if !self.externs.has_changed() {
return;
}
let changed = self.externs.take_changed();
for &slot in &changed {
if let Some(deps) = self.plan.input_dependents.get(slot) {
for &i in deps {
self.ran[i] = 0;
self.clean[i] = false;
}
self.all_ran = false;
}
}
self.externs.return_changed(changed);
}
#[inline]
fn refresh_cells(&mut self) {
if self.externs.cells_dirty() {
self.externs.refresh_cells(&mut self.buffer);
self.dirty_refreshed();
}
}
fn attach_cell(&mut self, name: &str, cell: crate::kernel::SharedCell) -> Result<(), String> {
let slot = self.externs.attach_cell(name, cell)?;
self.dirty_input(slot);
self.drive.stale = true;
Ok(())
}
#[inline]
fn dirty_input(&mut self, slot: usize) {
if let Some(deps) = self.dirty.get(slot) {
for &i in deps {
self.clean[i] = false;
}
}
}
#[inline]
fn run_steps(&mut self, order: &[usize]) {
self.run_guarded(|core| core.run_order(order));
}
#[inline]
fn run_guarded(&mut self, body: impl FnOnce(&mut Self)) {
let capture = crate::kernel::engines::EvalPanicCaptureGuard::arm();
let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| body(self)));
drop(capture);
if let Err(payload) = outcome {
let sites = std::sync::Arc::clone(&self.sites);
let node = self.steps[self.cur_step].failing_node(&self.buffer, self.tracker);
sites.reraise(payload, node, &self.buffer, Some(&self.none));
}
#[cfg(debug_assertions)]
self.validate_refs();
}
#[inline]
fn run_order(&mut self, order: &[usize]) {
let steps = &self.steps;
let none_free = !self.any_none;
for &i in order {
if self.all_ran || self.ran[i] == self.epoch {
continue;
}
let never = self.volatile[i];
if (self.use_clean || self.side[i]) && self.clean[i] && !never {
self.ran[i] = self.epoch;
continue;
}
self.cur_step = i;
run_hybrid_step(
&steps[i],
none_free,
&mut self.buffer,
&mut self.none,
&mut self.gather_buf,
&mut self.scatter_buf,
&mut self.scratch,
);
self.ran[i] = self.epoch;
self.clean[i] = !never;
}
}
#[inline]
fn run_fresh(&mut self) {
let steps = &self.steps;
for (i, step) in steps.iter().enumerate() {
if self.side[i] {
let never = self.volatile[i];
if self.clean[i] && !never {
continue;
}
self.clean[i] = !never;
}
self.cur_step = i;
run_hybrid_step(
step,
true,
&mut self.buffer,
&mut self.none,
&mut self.gather_buf,
&mut self.scatter_buf,
&mut self.scratch,
);
}
self.all_ran = true;
}
#[inline]
fn eval_all(&mut self) {
let fresh = self.drive.stale;
if fresh {
self.begin_epoch();
} else {
self.refresh_cells();
}
if fresh && !self.use_clean && !self.any_none {
self.run_guarded(|core| core.run_fresh());
} else {
let all = std::sync::Arc::clone(&self.all);
self.run_steps(&all);
}
}
fn pull_named(&mut self, name: &str) -> crate::ast::Value {
if self.drive.stale {
self.begin_epoch();
} else {
self.refresh_cells();
}
let plan = std::sync::Arc::clone(&self.plan);
if let Some(order) = plan.cones.get(name) {
self.run_steps(order);
}
self.value_of(name)
}
fn pull_at(&mut self, index: usize) -> crate::ast::Value {
if self.resolved_outputs.len() <= index {
self.resolved_outputs.resize(index + 1, None);
}
if self.resolved_outputs[index].is_none() {
let name = self
.externs
.output_names()
.get(index)
.cloned()
.unwrap_or_else(|| {
panic!(
"no output at index {index}; this kernel declares {}",
self.externs.output_names().len()
)
});
let slot = self.output_map[&name];
let ty = self
.output_types
.get(&name)
.copied()
.unwrap_or(crate::ast::PortType::U64);
let cone = self
.plan
.cones
.get(&name)
.map(|c| std::sync::Arc::from(c.as_slice()));
self.resolved_outputs[index] = Some((slot, ty, cone));
}
if self.drive.stale {
self.begin_epoch();
} else {
self.refresh_cells();
}
let (slot, ty, cone) = self.resolved_outputs[index]
.clone()
.expect("resolved above");
if let Some(order) = cone {
self.run_steps(&order);
}
self.slot_value(slot, ty)
}
fn value_of(&self, name: &str) -> crate::ast::Value {
let slot = self.output_map[name];
let ty = self
.output_types
.get(name)
.copied()
.unwrap_or(crate::ast::PortType::U64);
self.slot_value(slot, ty)
}
fn slot_value(&self, slot: usize, ty: crate::ast::PortType) -> crate::ast::Value {
if self.none.get(slot).copied().unwrap_or(false) {
return crate::ast::Value::None;
}
crate::compile::marshal::decode_output(&self.buffer, slot, ty)
}
fn plan(&self) -> crate::EnginePlan {
let (native_segments, closure_steps) = self.engine_counts();
crate::EnginePlan {
native_segments,
closure_steps,
interpreted_nodes: 0,
}
}
fn invalidate_all(&mut self) {
self.clean.fill(false);
self.all_ran = false;
self.drive.stale = true;
}
}
#[inline]
fn eval_all_hybrid_steps(core: &mut HybridCore) {
core.drive.stale = true;
core.eval_all();
}
impl HybridCore {
fn engine_counts(&self) -> (usize, usize) {
let closures = self
.steps
.iter()
.filter(|s| matches!(s, HybridStep::Closure(_)))
.count();
(self.steps.len() - closures, closures)
}
fn set_extern(&mut self, name: &str, value: crate::ast::Value) -> Result<usize, String> {
let (slot, unset) = self.externs.set(name, value, &mut self.buffer)?;
self.extern_written(slot, unset);
Ok(slot)
}
fn set_extern_at(&mut self, index: usize, value: crate::ast::Value) -> Result<usize, String> {
let (slot, unset) = self.externs.set_at(index, value, &mut self.buffer)?;
self.extern_written(slot, unset);
Ok(slot)
}
fn extern_written(&mut self, slot: usize, unset: bool) {
self.none[slot] = unset;
let was = self.any_none;
self.any_none = self.externs.any_unset();
if was && !self.any_none {
self.none.fill(false);
}
self.dirty_input(slot);
self.drive.stale = true;
}
}
#[derive(Clone)]
pub struct HybridKernelRaw {
core: HybridCore,
}
impl HybridKernelRaw {
#[inline]
fn set_coords(&mut self, coords: &[u64]) {
for (i, &c) in coords.iter().enumerate().take(self.core.coord_count) {
if self.core.buffer[i] != c {
self.core.buffer[i] = c;
self.core.dirty_input(i);
}
}
}
#[inline]
pub fn eval(&mut self, coords: &[u64]) {
self.set_coords(coords);
eval_all_hybrid_steps(&mut self.core);
}
#[cfg(feature = "jit")]
fn pull_output(&mut self, name: &str) -> crate::ast::Value {
self.core.pull_named(name)
}
pub fn set_input(&mut self, name: &str, value: crate::ast::Value) -> Result<(), String> {
self.core.set_extern(name, value).map(|_| ())
}
pub fn set_input_at(&mut self, index: usize, value: crate::ast::Value) -> Result<(), String> {
self.core.set_extern_at(index, value).map(|_| ())
}
pub fn externs(&self) -> Vec<(&str, crate::ast::PortType)> {
self.core.externs.names()
}
pub fn cursor_schemas(&self) -> &[crate::iteration::source::SourceSchema] {
self.core.externs.cursor_schemas()
}
pub fn set_cursor(
&mut self,
name: &str,
partition: &crate::iteration::cursor_partition::Partition,
) -> Result<(), String> {
for (slot, value) in self.core.externs.cursor_writes(name, partition)? {
self.set_input(&slot, value)?;
}
Ok(())
}
#[inline]
pub fn eval_for_slot(&mut self, coords: &[u64], slot: usize) -> u64 {
self.core.guard_ref_slot(slot);
self.eval(coords);
self.core.buffer[slot]
}
#[inline]
pub fn get(&self, name: &str) -> u64 {
let slot = self.core.output_map[name];
self.core.guard_ref_slot(slot);
self.core.buffer[slot]
}
#[inline]
pub fn get_slot(&self, slot: usize) -> u64 {
self.core.guard_ref_slot(slot);
self.core.buffer[slot]
}
crate::compile::ref_readers!();
pub fn get_value(&self, name: &str) -> crate::ast::Value {
self.core.value_of(name)
}
pub fn coord_count(&self) -> usize {
self.core.coord_count
}
pub fn engine_counts(&self) -> (usize, usize) {
self.core.engine_counts()
}
pub fn resolve_output(&self, name: &str) -> Option<usize> {
self.core.output_map.get(name).copied()
}
pub fn retain_nodes(&mut self, nodes: Vec<Box<dyn PolydatNode>>) {
self.core._nodes = std::sync::Arc::new(nodes);
}
}
#[derive(Clone)]
pub struct HybridKernelPull {
core: HybridCore,
slot_provenance: Vec<crate::kernel::ProvMask>,
changed_mask: crate::kernel::ProvMask,
force_run: bool,
}
impl HybridKernelPull {
#[inline]
fn set_inputs(&mut self, coords: &[u64]) {
self.changed_mask.clear();
for (i, &c) in coords.iter().enumerate().take(self.core.coord_count) {
if self.core.buffer[i] != c {
self.core.buffer[i] = c;
self.changed_mask.set(i);
self.core.dirty_input(i);
}
}
}
#[inline]
pub fn eval(&mut self, coords: &[u64]) {
self.set_inputs(coords);
self.force_run = false;
eval_all_hybrid_steps(&mut self.core);
}
fn pull_output(&mut self, name: &str) -> crate::ast::Value {
self.core.pull_named(name)
}
#[inline]
pub fn eval_for_slot(&mut self, coords: &[u64], slot: usize) -> u64 {
self.core.guard_ref_slot(slot);
self.set_inputs(coords);
if !self.force_run
&& slot < self.slot_provenance.len()
&& !self.slot_provenance[slot].intersects(&self.changed_mask)
{
return self.core.buffer[slot];
}
self.force_run = false;
eval_all_hybrid_steps(&mut self.core);
self.core.buffer[slot]
}
pub fn set_input(&mut self, name: &str, value: crate::ast::Value) -> Result<(), String> {
self.core.set_extern(name, value)?;
self.force_run = true;
Ok(())
}
pub fn set_input_at(&mut self, index: usize, value: crate::ast::Value) -> Result<(), String> {
self.core.set_extern_at(index, value)?;
self.force_run = true;
Ok(())
}
pub fn externs(&self) -> Vec<(&str, crate::ast::PortType)> {
self.core.externs.names()
}
pub fn cursor_schemas(&self) -> &[crate::iteration::source::SourceSchema] {
self.core.externs.cursor_schemas()
}
pub fn set_cursor(
&mut self,
name: &str,
partition: &crate::iteration::cursor_partition::Partition,
) -> Result<(), String> {
for (slot, value) in self.core.externs.cursor_writes(name, partition)? {
self.set_input(&slot, value)?;
}
Ok(())
}
#[inline]
pub fn get(&self, name: &str) -> u64 {
let slot = self.core.output_map[name];
self.core.guard_ref_slot(slot);
self.core.buffer[slot]
}
#[inline]
pub fn get_slot(&self, slot: usize) -> u64 {
self.core.guard_ref_slot(slot);
self.core.buffer[slot]
}
crate::compile::ref_readers!();
pub fn get_value(&self, name: &str) -> crate::ast::Value {
self.core.value_of(name)
}
pub fn coord_count(&self) -> usize {
self.core.coord_count
}
pub fn engine_counts(&self) -> (usize, usize) {
self.core.engine_counts()
}
pub fn resolve_output(&self, name: &str) -> Option<usize> {
self.core.output_map.get(name).copied()
}
pub fn retain_nodes(&mut self, nodes: Vec<Box<dyn PolydatNode>>) {
self.core._nodes = std::sync::Arc::new(nodes);
}
}
#[derive(Clone)]
pub struct HybridKernelPushPull {
core: HybridCore,
slot_provenance: Vec<crate::kernel::ProvMask>,
changed_mask: crate::kernel::ProvMask,
force_run: bool,
}
impl HybridKernelPushPull {
pub fn set_input(&mut self, name: &str, value: crate::ast::Value) -> Result<(), String> {
self.core.set_extern(name, value)?;
self.force_run = true;
Ok(())
}
pub fn set_input_at(&mut self, index: usize, value: crate::ast::Value) -> Result<(), String> {
self.core.set_extern_at(index, value)?;
self.force_run = true;
Ok(())
}
pub fn externs(&self) -> Vec<(&str, crate::ast::PortType)> {
self.core.externs.names()
}
pub fn cursor_schemas(&self) -> &[crate::iteration::source::SourceSchema] {
self.core.externs.cursor_schemas()
}
pub fn set_cursor(
&mut self,
name: &str,
partition: &crate::iteration::cursor_partition::Partition,
) -> Result<(), String> {
for (slot, value) in self.core.externs.cursor_writes(name, partition)? {
self.set_input(&slot, value)?;
}
Ok(())
}
#[inline]
fn set_inputs(&mut self, coords: &[u64]) {
self.changed_mask.clear();
for (i, &c) in coords.iter().enumerate().take(self.core.coord_count) {
if self.core.buffer[i] != c {
self.core.buffer[i] = c;
self.changed_mask.set(i);
self.core.dirty_input(i);
}
}
}
#[inline]
pub fn eval(&mut self, coords: &[u64]) {
self.set_inputs(coords);
self.force_run = false;
self.core.drive.stale = true;
self.core.eval_all();
}
fn pull_output(&mut self, name: &str) -> crate::ast::Value {
self.core.pull_named(name)
}
#[inline]
pub fn eval_for_slot(&mut self, coords: &[u64], slot: usize) -> u64 {
self.core.guard_ref_slot(slot);
self.set_inputs(coords);
if !self.force_run
&& slot < self.slot_provenance.len()
&& !self.slot_provenance[slot].intersects(&self.changed_mask)
{
return self.core.buffer[slot];
}
self.force_run = false;
self.core.drive.stale = true;
self.core.eval_all();
self.core.buffer[slot]
}
#[inline]
pub fn get(&self, name: &str) -> u64 {
let slot = self.core.output_map[name];
self.core.guard_ref_slot(slot);
self.core.buffer[slot]
}
#[inline]
pub fn get_slot(&self, slot: usize) -> u64 {
self.core.guard_ref_slot(slot);
self.core.buffer[slot]
}
crate::compile::ref_readers!();
pub fn get_value(&self, name: &str) -> crate::ast::Value {
self.core.value_of(name)
}
pub fn coord_count(&self) -> usize {
self.core.coord_count
}
pub fn engine_counts(&self) -> (usize, usize) {
self.core.engine_counts()
}
pub fn resolve_output(&self, name: &str) -> Option<usize> {
self.core.output_map.get(name).copied()
}
pub fn retain_nodes(&mut self, nodes: Vec<Box<dyn PolydatNode>>) {
self.core._nodes = std::sync::Arc::new(nodes);
}
}
pub type HybridKernel = HybridKernelPushPull;
fn flatten_input_slots(
wiring: &[Vec<WireSource>],
nodes: &[Box<dyn PolydatNode>],
node_idx: usize,
port_offsets: &[Vec<usize>],
input_starts: &[usize],
input_widths: &[usize],
) -> Vec<usize> {
let mut slots = Vec::new();
for source in &wiring[node_idx] {
let (start, w) = match source {
WireSource::Input(c) => (
input_starts.get(*c).copied().unwrap_or(*c),
input_widths.get(*c).copied().unwrap_or(1),
),
WireSource::NodeOutput(u, p) => (
port_offsets[*u][*p],
nodes[*u].meta().outs[*p].typ.slot_width(),
),
};
slots.extend(start..start + w);
}
slots
}
fn flatten_ref_output_starts(
nodes: &[Box<dyn PolydatNode>],
node_idx: usize,
port_offsets: &[Vec<usize>],
) -> Vec<usize> {
nodes[node_idx]
.meta()
.outs
.iter()
.enumerate()
.filter(|(_, out)| out.typ.slot_color() == crate::ast::SlotColor::Ref2)
.map(|(p, _)| port_offsets[node_idx][p])
.collect()
}
fn flatten_output_slots(
nodes: &[Box<dyn PolydatNode>],
node_idx: usize,
port_offsets: &[Vec<usize>],
) -> Vec<usize> {
let mut slots = Vec::new();
for (p, out) in nodes[node_idx].meta().outs.iter().enumerate() {
let start = port_offsets[node_idx][p];
slots.extend(start..start + out.typ.slot_width());
}
slots
}
#[cfg(feature = "jit")]
#[allow(clippy::too_many_arguments)]
pub(crate) fn build_hybrid(
nodes: &[Box<dyn PolydatNode>],
wiring: &[Vec<WireSource>],
coord_count: usize,
total_slots: usize,
port_offsets: &[Vec<usize>],
input_starts: &[usize],
input_widths: &[usize],
output_map: HashMap<String, usize>,
ref_slots: Vec<bool>,
input_types: &[crate::ast::PortType],
externs: crate::compile::externs::Externs,
constant: Vec<bool>,
volatile: Vec<bool>,
attribution: std::sync::Arc<crate::compile::Attribution>,
) -> Result<HybridKernelPushPull, String> {
let mut steps: Vec<HybridStep> = Vec::new();
let mut scratch: Vec<crate::ast::ScratchBuf> = Vec::new();
let mut ref_scratch: Vec<(usize, usize)> = Vec::new();
let mut max_inputs = 0usize;
let mut max_outputs = 0usize;
let classifications: Vec<(JitOp, Vec<usize>, Vec<usize>)> = nodes
.iter()
.enumerate()
.map(|(node_idx, node)| {
let wire_types: Vec<crate::ast::PortType> = wiring[node_idx]
.iter()
.map(|src| match src {
WireSource::Input(c) => input_types
.get(*c)
.copied()
.unwrap_or(crate::ast::PortType::U64),
WireSource::NodeOutput(j, p) => nodes[*j].meta().outs[*p].typ,
})
.collect();
let jit_op = jit::classify_node_typed(node.as_ref(), &wire_types);
let input_slots = flatten_input_slots(
wiring,
nodes,
node_idx,
port_offsets,
input_starts,
input_widths,
);
let output_slots = flatten_output_slots(nodes, node_idx, port_offsets);
max_inputs = max_inputs.max(input_slots.len());
max_outputs = max_outputs.max(output_slots.len());
(jit_op, input_slots, output_slots)
})
.collect();
let mut classifications = classifications;
let unset = externs.unset_slots();
if !unset.is_empty() {
let mut tainted = vec![false; nodes.len()];
for node_idx in 0..nodes.len() {
tainted[node_idx] = wiring[node_idx].iter().any(|src| match src {
WireSource::Input(c) => unset.contains(&input_starts[*c]),
WireSource::NodeOutput(j, _) => tainted[*j],
});
if tainted[node_idx] {
classifications[node_idx].0 = JitOp::Fallback;
}
}
}
let mut node_step = vec![usize::MAX; nodes.len()];
let order: Vec<usize> = (0..nodes.len())
.filter(|&k| constant[k])
.chain((0..nodes.len()).filter(|&k| !constant[k]))
.collect();
let mut pos = 0;
while pos < order.len() {
let i = order[pos];
if matches!(classifications[i].0, JitOp::Fallback) {
let node = &nodes[i];
let (_, ref input_slots, ref output_slots) = classifications[i];
let scratch_start = scratch.len();
let wire_types: Vec<crate::ast::PortType> = wiring[i]
.iter()
.map(|src| match src {
WireSource::Input(c) => input_types
.get(*c)
.copied()
.unwrap_or(crate::ast::PortType::U64),
WireSource::NodeOutput(j, p) => nodes[*j].meta().outs[*p].typ,
})
.collect();
let op = if let Some(op) = node.compiled_u64() {
ClosureOp::U64(op)
} else if let Some(op) = crate::compile::assembly::identity_op(node.as_ref()) {
ClosureOp::U64(op)
} else if let Some(kit) = ref_copy_or_slot(node.as_ref(), &wire_types) {
scratch.extend(kit.scratch.iter().map(|e| crate::ast::ScratchBuf::new(*e)));
let starts = flatten_ref_output_starts(nodes, i, port_offsets);
ref_scratch.extend(crate::compile::assembly::scratch_pairs(
&node.meta().name,
&starts,
&kit.scratch,
scratch_start,
));
ClosureOp::Slot(kit.op)
} else {
return Err(format!(
"node '{}' has no compiled form and can't be JIT-compiled",
node.meta().name
));
};
node_step[i] = steps.len();
steps.push(HybridStep::Closure(ClosureStep {
op,
input_slots: input_slots.clone(),
output_slots: output_slots.clone(),
scratch_range: (scratch_start, scratch.len()),
accepts_none: node.accepts_none_inputs(),
node: i,
}));
pos += 1;
} else {
let is_side =
|k: usize| matches!(nodes[k].purity(), crate::ast::Purity::SideChannel { .. });
let batch_start = pos;
let first = order[batch_start];
while pos < order.len()
&& !matches!(classifications[order[pos]].0, JitOp::Fallback)
&& constant[order[pos]] == constant[first]
&& volatile[order[pos]] == volatile[first]
&& !is_side(order[pos])
&& !is_side(first)
{
pos += 1;
}
if pos == batch_start {
pos += 1;
}
let members: Vec<usize> = order[batch_start..pos].to_vec();
for &k in &members {
let base = scratch.len();
classifications[k].0.place_scratch(base);
let elems = classifications[k].0.scratch_elems().to_vec();
ref_scratch.extend(crate::compile::assembly::scratch_pairs(
&nodes[k].meta().name,
&flatten_ref_output_starts(nodes, k, port_offsets),
&elems,
base,
));
scratch.extend(elems.iter().map(|e| crate::ast::ScratchBuf::new(*e)));
}
let batch: Vec<(JitOp, Vec<usize>, Vec<usize>)> = members
.iter()
.map(|&k| classifications[k].clone())
.collect();
let written: std::collections::HashSet<usize> = batch
.iter()
.flat_map(|(_, _, o)| o.iter().copied())
.collect();
let mut input_slots: Vec<usize> = Vec::new();
for (_, ins, _) in &batch {
for &s in ins {
if !written.contains(&s) && !input_slots.contains(&s) {
input_slots.push(s);
}
}
}
let output_slots: Vec<usize> = batch
.iter()
.flat_map(|(_, _, o)| o.iter().copied())
.collect();
let (code_fn, code) = jit::compile_jit_entry(&batch, Some(total_slots))?;
let segment = steps.len();
for &k in &members {
node_step[k] = segment;
}
steps.push(HybridStep::Jit(JitSegment {
code_fn,
fallible: code.fallible(),
_module: code,
input_slots,
output_slots,
nodes: members,
}));
}
}
let output_types = output_types_of(nodes, port_offsets, input_starts, input_types, &output_map);
build_pushpull_from_steps(
steps,
scratch,
ref_scratch,
ref_slots,
wiring,
nodes,
coord_count,
total_slots,
output_map,
max_inputs,
max_outputs,
input_starts,
input_widths,
output_types,
externs,
constant,
volatile,
attribution,
node_step,
)
}
fn output_types_of(
nodes: &[Box<dyn PolydatNode>],
port_offsets: &[Vec<usize>],
input_starts: &[usize],
input_types: &[crate::ast::PortType],
output_map: &HashMap<String, usize>,
) -> HashMap<String, crate::ast::PortType> {
let mut slot_types: HashMap<usize, crate::ast::PortType> = HashMap::new();
for (start, ty) in input_starts.iter().zip(input_types) {
slot_types.insert(*start, *ty);
}
for (node_idx, node) in nodes.iter().enumerate() {
for (p, out) in node.meta().outs.iter().enumerate() {
slot_types.insert(port_offsets[node_idx][p], out.typ);
}
}
output_map
.iter()
.map(|(name, slot)| {
(
name.clone(),
slot_types
.get(slot)
.copied()
.unwrap_or(crate::ast::PortType::U64),
)
})
.collect()
}
#[cfg(not(feature = "jit"))]
#[allow(clippy::too_many_arguments)]
pub(crate) fn build_hybrid(
nodes: &[Box<dyn PolydatNode>],
wiring: &[Vec<WireSource>],
coord_count: usize,
total_slots: usize,
port_offsets: &[Vec<usize>],
input_starts: &[usize],
input_widths: &[usize],
output_map: HashMap<String, usize>,
ref_slots: Vec<bool>,
input_types: &[crate::ast::PortType],
externs: crate::compile::externs::Externs,
constant: Vec<bool>,
volatile: Vec<bool>,
attribution: std::sync::Arc<crate::compile::Attribution>,
) -> Result<HybridKernelPushPull, String> {
let mut steps: Vec<HybridStep> = Vec::new();
let mut scratch: Vec<crate::ast::ScratchBuf> = Vec::new();
let mut ref_scratch: Vec<(usize, usize)> = Vec::new();
let mut max_inputs = 0usize;
let mut max_outputs = 0usize;
for (node_idx, node) in nodes.iter().enumerate() {
let input_slots = flatten_input_slots(
wiring,
nodes,
node_idx,
port_offsets,
input_starts,
input_widths,
);
let output_slots = flatten_output_slots(nodes, node_idx, port_offsets);
max_inputs = max_inputs.max(input_slots.len());
max_outputs = max_outputs.max(output_slots.len());
let scratch_start = scratch.len();
let wire_types: Vec<crate::ast::PortType> = wiring[node_idx]
.iter()
.map(|src| match src {
WireSource::Input(c) => input_types
.get(*c)
.copied()
.unwrap_or(crate::ast::PortType::U64),
WireSource::NodeOutput(j, p) => nodes[*j].meta().outs[*p].typ,
})
.collect();
let op = if let Some(op) = node.compiled_u64() {
ClosureOp::U64(op)
} else if let Some(op) = crate::compile::assembly::identity_op(node.as_ref()) {
ClosureOp::U64(op)
} else if let Some(kit) = ref_copy_or_slot(node.as_ref(), &wire_types) {
scratch.extend(kit.scratch.iter().map(|e| crate::ast::ScratchBuf::new(*e)));
let starts = flatten_ref_output_starts(nodes, node_idx, port_offsets);
ref_scratch.extend(crate::compile::assembly::scratch_pairs(
&node.meta().name,
&starts,
&kit.scratch,
scratch_start,
));
ClosureOp::Slot(kit.op)
} else {
return Err(format!("node '{}' has no compiled form", node.meta().name));
};
steps.push(HybridStep::Closure(ClosureStep {
op,
input_slots,
output_slots,
scratch_range: (scratch_start, scratch.len()),
accepts_none: node.accepts_none_inputs(),
node: node_idx,
}));
}
let node_step: Vec<usize> = (0..nodes.len()).collect();
let output_types = output_types_of(nodes, port_offsets, input_starts, input_types, &output_map);
build_pushpull_from_steps(
steps,
scratch,
ref_scratch,
ref_slots,
wiring,
nodes,
coord_count,
total_slots,
output_map,
max_inputs,
max_outputs,
input_starts,
input_widths,
output_types,
externs,
constant,
volatile,
attribution,
node_step,
)
}
#[allow(clippy::too_many_arguments)]
fn build_pushpull_from_steps(
steps: Vec<HybridStep>,
scratch: Vec<crate::ast::ScratchBuf>,
ref_scratch: Vec<(usize, usize)>,
ref_slots: Vec<bool>,
wiring: &[Vec<WireSource>],
nodes: &[Box<dyn PolydatNode>],
coord_count: usize,
total_slots: usize,
output_map: HashMap<String, usize>,
max_inputs: usize,
max_outputs: usize,
_input_starts: &[usize],
input_widths: &[usize],
output_types: HashMap<String, crate::ast::PortType>,
externs: crate::compile::externs::Externs,
constant: Vec<bool>,
volatile: Vec<bool>,
attribution: std::sync::Arc<crate::compile::Attribution>,
node_step: Vec<usize>,
) -> Result<HybridKernelPushPull, String> {
let step_count = steps.len();
debug_assert_eq!(node_step.len(), nodes.len());
debug_assert!(node_step.iter().all(|&s| s < step_count));
let to_steps = |list: &[usize]| -> Vec<usize> {
let mut v: Vec<usize> = list.iter().map(|&n| node_step[n]).collect();
v.sort_unstable();
v.dedup();
v
};
let mut buffer = vec![0u64; total_slots + 1];
let mut none = vec![false; total_slots];
let any_none = externs.seed(&mut buffer, Some(&mut none));
let node_provenance = crate::kernel::PolydatProgram::compute_provenance(nodes, wiring);
let input_dependents: Vec<Vec<usize>> =
crate::kernel::PolydatProgram::compute_dependents(&node_provenance, input_widths.len())
.iter()
.map(|d| to_steps(d))
.collect();
let step_dependents: Vec<Vec<usize>> = input_widths
.iter()
.enumerate()
.flat_map(|(i, w)| {
std::iter::repeat_n(input_dependents.get(i).cloned().unwrap_or_default(), *w)
})
.collect();
let step_outs: Vec<&[usize]> = steps.iter().map(|s| s.output_slots()).collect();
let slot_provenance =
crate::compile::slot_provenance(coord_count, total_slots, &step_outs, &step_dependents);
debug_assert_eq!(constant.len(), nodes.len());
debug_assert_eq!(volatile.len(), nodes.len());
let mut step_constant = vec![true; step_count];
let mut step_volatile = vec![false; step_count];
let mut side = vec![false; step_count];
for (n, node) in nodes.iter().enumerate() {
let s = node_step[n];
step_constant[s] &= constant[n];
step_volatile[s] |= volatile[n];
side[s] |= matches!(node.purity(), crate::ast::Purity::SideChannel { .. });
}
let volatile = step_volatile;
let constants: Vec<usize> = (0..step_count).filter(|&i| step_constant[i]).collect();
let step_inputs: Vec<&[usize]> = steps.iter().map(|s| s.input_slots()).collect();
let step_outputs: Vec<&[usize]> = steps.iter().map(|s| s.output_slots()).collect();
let plan = crate::compile::Invalidation::from_provenance(
step_dependents.clone(),
&step_inputs,
&step_outputs,
&output_map,
total_slots,
);
let mut slot_step: Vec<Option<usize>> = vec![None; total_slots];
for (i, outs) in step_outputs.iter().enumerate() {
for &s in outs.iter() {
slot_step[s] = Some(i);
}
}
drop(step_inputs);
drop(step_outputs);
let dirty: Vec<Vec<usize>> = plan.input_dependents.clone();
let volatile_steps: Vec<usize> = (0..step_count).filter(|&i| volatile[i]).collect();
let mut kernel = HybridKernelPushPull {
core: HybridCore {
buffer,
coord_count,
steps: std::sync::Arc::new(steps),
output_map,
gather_buf: vec![0u64; max_inputs.max(1)],
scatter_buf: vec![0u64; max_outputs.max(1)],
scratch,
ref_slots,
ref_scratch,
output_types,
externs,
traversals: Vec::new().into(),
resolved_outputs: Vec::new(),
_nodes: std::sync::Arc::new(Vec::new()),
drive: crate::compile::Drive {
coords: Vec::new(),
stale: true,
},
none,
ran: vec![0; step_count],
epoch: 0,
all_ran: false,
clean: vec![false; step_count],
use_clean: true,
plan: std::sync::Arc::new(plan),
volatile: volatile.into(),
side: side.into(),
slot_step: slot_step.into(),
sites: attribution,
cur_step: 0,
tracker: total_slots,
all: (0..step_count).collect::<Vec<usize>>().into(),
dirty: dirty.into(),
any_none,
volatile_steps: volatile_steps.into(),
},
slot_provenance,
changed_mask: crate::kernel::ProvMask::all_below(coord_count), force_run: false,
};
kernel.core.begin_epoch();
kernel.core.run_steps(&constants);
kernel.core.drive.stale = true;
Ok(kernel)
}
fn ref_copy_or_slot(
node: &dyn PolydatNode,
wire_types: &[crate::ast::PortType],
) -> Option<crate::ast::CompiledSlotKit> {
let meta = node.meta();
if (meta.name == "identity" || meta.name.starts_with("__port_"))
&& meta.outs.len() == 1
&& meta.outs[0].typ.slot_color() == crate::ast::SlotColor::Ref2
{
return crate::compile::assembly::ref_copy_kit(meta.outs[0].typ);
}
node.compiled_slot(wire_types)
}
#[cfg(feature = "jit")]
impl HybridKernelRaw {
fn mark_all_dirty(&mut self) {}
}
impl HybridKernelPull {
fn mark_all_dirty(&mut self) {
self.changed_mask = crate::kernel::ProvMask::all_below(self.core.coord_count);
self.force_run = true;
}
}
impl HybridKernelPushPull {
fn mark_all_dirty(&mut self) {
self.core.clean.fill(false);
self.changed_mask = crate::kernel::ProvMask::all_below(self.core.coord_count);
self.force_run = true;
}
#[cfg(feature = "jit")]
pub(crate) fn into_raw(self) -> HybridKernelRaw {
let mut core = self.core;
core.set_use_clean(false);
HybridKernelRaw { core }
}
#[cfg(feature = "jit")]
pub(crate) fn into_pull(self) -> HybridKernelPull {
let mut core = self.core;
core.set_use_clean(false);
let changed_mask = crate::kernel::ProvMask::all_below(core.coord_count);
HybridKernelPull {
core,
slot_provenance: self.slot_provenance,
changed_mask,
force_run: false,
}
}
}
use crate::compile::select::{Engine, Provenance};
#[cfg(feature = "jit")]
crate::compile::impl_kernel_trait!(HybridKernelRaw, Engine::Native(Provenance::Raw));
crate::compile::impl_kernel_trait!(HybridKernelPull, Engine::Native(Provenance::Pull));
crate::compile::impl_kernel_trait!(HybridKernelPushPull, Engine::Native(Provenance::PushPull));
macro_rules! hybrid_drive {
($ty:ident, $set_coords:ident) => {
impl $ty {
fn pull_value(&mut self, name: &str) -> crate::ast::Value {
let coords = std::mem::take(&mut self.core.drive.coords);
self.$set_coords(&coords);
self.core.drive.coords = coords;
self.pull_output(name)
}
fn pull_value_at(&mut self, index: usize) -> crate::ast::Value {
let coords = std::mem::take(&mut self.core.drive.coords);
self.$set_coords(&coords);
self.core.drive.coords = coords;
self.core.pull_at(index)
}
fn eval_pending(&mut self) {
let coords = std::mem::take(&mut self.core.drive.coords);
self.eval(&coords);
self.core.drive.coords = coords;
}
}
};
}
#[cfg(feature = "jit")]
hybrid_drive!(HybridKernelRaw, set_coords);
hybrid_drive!(HybridKernelPull, set_inputs);
hybrid_drive!(HybridKernelPushPull, set_inputs);
#[inline(always)]
fn run_hybrid_step(
step: &HybridStep,
none_free: bool,
buffer: &mut [u64],
none: &mut [bool],
gather: &mut [u64],
scatter: &mut [u64],
scratch: &mut [crate::ast::ScratchBuf],
) {
if !none_free && step.input_slots().iter().any(|&s| none[s]) {
#[cfg(feature = "jit")]
if let HybridStep::Jit(_) = step {
panic!(
"a `None` reached native code in a hybrid kernel: an extern was cleared \
after the build (docs/design/engine_parity.md, A12)"
);
}
if !step.accepts_none() {
for &s in step.output_slots() {
none[s] = true;
}
return;
}
}
match step {
#[cfg(feature = "jit")]
HybridStep::Jit(seg) => {
let code_fn = seg.code_fn;
let buf_const = buffer.as_ptr();
let buf_mut = buffer.as_mut_ptr();
let sc = scratch.as_mut_ptr();
if seg.fallible {
crate::compile::jit::invoke_with_catch(move || unsafe {
(code_fn)(buf_const, buf_mut, sc);
});
} else {
unsafe { (code_fn)(buf_const, buf_mut, sc) };
}
}
HybridStep::Closure(cs) => {
for (i, &slot) in cs.input_slots.iter().enumerate() {
gather[i] = buffer[slot];
}
match &cs.op {
ClosureOp::U64(op) => op(
&gather[..cs.input_slots.len()],
&mut scatter[..cs.output_slots.len()],
),
ClosureOp::Slot(op) => op(
&gather[..cs.input_slots.len()],
&mut scatter[..cs.output_slots.len()],
&mut scratch[cs.scratch_range.0..cs.scratch_range.1],
),
}
for (i, &slot) in cs.output_slots.iter().enumerate() {
buffer[slot] = scatter[i];
}
}
}
if !none_free {
for &s in step.output_slots() {
none[s] = false;
}
}
}