diff --git a/backend-simulators/pim/pim-simulator/src/bin/pim-simulator/main.rs b/backend-simulators/pim/pim-simulator/src/bin/pim-simulator/main.rs index 506200f..38b538e 100644 --- a/backend-simulators/pim/pim-simulator/src/bin/pim-simulator/main.rs +++ b/backend-simulators/pim/pim-simulator/src/bin/pim-simulator/main.rs @@ -11,6 +11,7 @@ use pimcore::cpu::crossbar::Crossbar; use pimcore::json_to_instruction::json_to_executor; use pimcore::memory_manager::CoreMemory; use pimcore::tracing::TRACER; +use pimcore::{DiagnosticSchedulePolicy, DiagnosticScheduleTarget}; use serde_json::Value; use std::collections::HashMap; use std::fs::{self, File}; @@ -60,6 +61,38 @@ struct Args { /// Optional directory for per-iteration output dumps #[arg(long)] batch_output_dir: Option, + + /// Optional JSONL shadow provenance trace + #[arg(long)] + provenance_trace: Option, + + /// Diagnostic-only barrier between global throughput iterations + #[arg(long)] + provenance_global_barrier: bool, + + /// Diagnostic-only ready-core delay, formatted as CORE:CYCLES + #[arg(long, value_name = "CORE:CYCLES")] + provenance_core_stall: Option, + + /// Diagnostic scheduler policy; greedy is the unchanged default + #[arg(long, value_enum, default_value_t = DiagnosticSchedulePolicyArg::Greedy)] + diagnostic_schedule_policy: DiagnosticSchedulePolicyArg, + + /// Deterministic seed for the randomized diagnostic scheduler + #[arg(long, default_value_t = 0)] + diagnostic_schedule_seed: u64, + + /// Adversarial target: WCORE:WPC:RCORE:RPC:BEGIN:END[:READER_ITER:WRITER_MIN_ITER] + #[arg(long, value_name = "WCORE:WPC:RCORE:RPC:BEGIN:END[:RITER:WMIN]")] + diagnostic_schedule_target: Option, + + /// Maximum number of target-consumer deferrals + #[arg(long, default_value_t = 10_000)] + diagnostic_schedule_deferral_budget: u64, + + /// Diagnostic delay applied when a target reader reaches its PC, formatted as CORE:PC:CYCLES or CORE:PC:ITERATION:CYCLES + #[arg(long, value_name = "CORE:PC[:ITERATION]:CYCLES")] + diagnostic_target_stall: Option, } #[derive(Clone, Debug, ValueEnum)] @@ -68,6 +101,13 @@ enum ExecutionMode { Throughput, } +#[derive(Clone, Copy, Debug, ValueEnum)] +enum DiagnosticSchedulePolicyArg { + Greedy, + Randomized, + Adversarial, +} + fn main() -> Result<()> { let args = Args::parse(); @@ -89,6 +129,33 @@ fn main() -> Result<()> { } }; set_memory(&mut executor, memory); + if let Some(path) = &args.provenance_trace { + executor.enable_provenance(path)?; + } + executor.set_provenance_global_barrier(args.provenance_global_barrier); + if let Some(spec) = args.provenance_core_stall.as_deref() { + let (core, cycles) = parse_core_stall(spec)?; + executor.set_provenance_core_stall(core, cycles); + } + executor.set_diagnostic_schedule_policy(match args.diagnostic_schedule_policy { + DiagnosticSchedulePolicyArg::Greedy => DiagnosticSchedulePolicy::Greedy, + DiagnosticSchedulePolicyArg::Randomized => DiagnosticSchedulePolicy::Randomized, + DiagnosticSchedulePolicyArg::Adversarial => DiagnosticSchedulePolicy::Adversarial, + }); + executor.set_diagnostic_schedule_seed(args.diagnostic_schedule_seed); + executor.set_diagnostic_schedule_deferral_budget(args.diagnostic_schedule_deferral_budget); + if let Some(spec) = args.diagnostic_target_stall.as_deref() { + let (core, pc, iteration, cycles) = parse_target_stall(spec)?; + executor.set_diagnostic_target_stall(core, pc, iteration, cycles); + } + if let Some(spec) = args.diagnostic_schedule_target.as_deref() { + executor.set_diagnostic_schedule_target(parse_schedule_target(spec)?); + } else if matches!( + args.diagnostic_schedule_policy, + DiagnosticSchedulePolicyArg::Adversarial + ) { + bail!("adversarial scheduling requires --diagnostic-schedule-target"); + } TRACER .lock() .unwrap() @@ -107,6 +174,75 @@ fn main() -> Result<()> { Ok(()) } +fn parse_core_stall(spec: &str) -> Result<(usize, u64)> { + let (core, cycles) = spec + .split_once(':') + .context("--provenance-core-stall must be CORE:CYCLES")?; + let core = core.parse().context("invalid stalled core")?; + let cycles = cycles.parse().context("invalid stall cycle count")?; + if cycles == 0 { + bail!("--provenance-core-stall cycles must be positive"); + } + Ok((core, cycles)) +} + +fn parse_schedule_target(spec: &str) -> Result { + let values: Vec = spec + .split(':') + .map(|value| { + value + .parse() + .with_context(|| format!("invalid schedule target field: {value}")) + }) + .collect::>()?; + if values.len() != 6 && values.len() != 8 { + bail!( + "--diagnostic-schedule-target requires WCORE:WPC:RCORE:RPC:BEGIN:END with optional RITER:WMIN" + ); + } + if values[5] <= values[4] { + bail!("schedule target address end must be greater than begin"); + } + Ok(DiagnosticScheduleTarget { + writer_core: values[0], + writer_pc: values[1], + reader_core: values[2], + reader_pc: values[3], + address_begin: values[4], + address_end: values[5], + reader_iteration: (values.len() == 8).then_some(values[6] as u32), + writer_min_iteration: (values.len() == 8).then_some(values[7] as u32), + }) +} + +fn parse_target_stall(spec: &str) -> Result<(usize, usize, Option, u64)> { + let values: Vec<&str> = spec.split(':').collect(); + if values.len() != 3 && values.len() != 4 { + bail!("--diagnostic-target-stall must be CORE:PC:CYCLES or CORE:PC:ITERATION:CYCLES"); + } + let core = values[0].parse().context("invalid target-stall core")?; + let pc = values[1].parse().context("invalid target-stall PC")?; + let (iteration, cycle_field) = if values.len() == 4 { + ( + Some( + values[2] + .parse() + .context("invalid target-stall iteration")?, + ), + values[3], + ) + } else { + (None, values[2]) + }; + let cycles = cycle_field + .parse() + .context("invalid target-stall cycle count")?; + if cycles == 0 { + bail!("--diagnostic-target-stall cycles must be positive"); + } + Ok((core, pc, iteration, cycles)) +} + fn batch_size(args: &Args) -> Result { match (&args.mode, args.batch_size) { (ExecutionMode::Latency, None | Some(1)) => Ok(1), diff --git a/backend-simulators/pim/pim-simulator/src/lib/cpu/mod.rs b/backend-simulators/pim/pim-simulator/src/lib/cpu/mod.rs index 0ad723f..e9c241f 100644 --- a/backend-simulators/pim/pim-simulator/src/lib/cpu/mod.rs +++ b/backend-simulators/pim/pim-simulator/src/lib/cpu/mod.rs @@ -2,11 +2,14 @@ use crate::utility::AddressArg; use anyhow::{Context, Result, ensure}; use std::{collections::HashMap, fmt::Debug}; +use super::{DiagnosticSchedulePolicy, DiagnosticScheduleTarget}; use crate::{ cpu::crossbar::Crossbar, instruction_set::Instructions, memory_manager::{CoreMemory, MemoryStorable, type_traits::TryToUsize}, + provenance::ProvenanceTracker, }; +use serde_json::json; pub mod crossbar; @@ -14,6 +17,7 @@ pub mod crossbar; pub struct CPU<'a> { cores: Box<[Core<'a>]>, batch_outputs: Option, + provenance: Option, } #[derive(Debug, Clone)] @@ -53,9 +57,265 @@ impl<'a> CPU<'a> { Self { cores: cores.into(), batch_outputs: None, + provenance: None, } } + pub(crate) fn enable_provenance( + &mut self, + path: impl AsRef, + ) -> std::io::Result<()> { + self.provenance = Some(ProvenanceTracker::new(self.cores.len(), path)?); + Ok(()) + } + + pub(crate) fn begin_provenance_batch(&mut self, batch_size: usize) { + if let Some(provenance) = &mut self.provenance { + provenance.begin_batch(batch_size); + } + } + + pub(crate) fn set_execution_context( + &mut self, + cycle: u64, + core: usize, + pc: usize, + iteration: u32, + ) { + if let Some(provenance) = &mut self.provenance { + provenance.set_context(cycle, core, pc, iteration); + } + } + + pub(crate) fn provenance_input_store(&mut self, address: usize, size: usize, sample: u32) { + if let Some(provenance) = &mut self.provenance { + provenance.input_store(address, size, sample); + } + } + + pub(crate) fn provenance_global_store_from_local( + &mut self, + core: usize, + global_address: usize, + local_address: usize, + size: usize, + ) { + if let Some(provenance) = &mut self.provenance { + provenance.global_store_from_local(core, global_address, local_address, size); + } + } + + pub(crate) fn provenance_global_load_to_local( + &mut self, + core: usize, + global_address: usize, + local_address: usize, + size: usize, + ) { + if let Some(provenance) = &mut self.provenance { + provenance.global_load_to_local(core, global_address, local_address, size); + } + } + + pub(crate) fn provenance_local_copy( + &mut self, + core: usize, + destination: usize, + source: usize, + size: usize, + operation: &'static str, + ) { + if let Some(provenance) = &mut self.provenance { + provenance.local_copy(core, destination, source, size, operation); + } + } + + pub(crate) fn provenance_local_strided_copy( + &mut self, + core: usize, + destination: usize, + source: usize, + element_size: usize, + stride: usize, + element_count: usize, + operation: &'static str, + ) { + if let Some(provenance) = &mut self.provenance { + provenance.local_strided_copy( + core, + destination, + source, + element_size, + stride, + element_count, + operation, + ); + } + } + + pub(crate) fn provenance_local_transform( + &mut self, + core: usize, + destination: usize, + sources: &[(usize, usize)], + output_size: usize, + operation: &'static str, + ) { + if let Some(provenance) = &mut self.provenance { + provenance.local_transform(core, destination, sources, output_size, operation); + } + } + + pub(crate) fn provenance_local_broadcast_transform( + &mut self, + core: usize, + destination: usize, + source: usize, + source_size: usize, + output_size: usize, + operation: &'static str, + ) { + if let Some(provenance) = &mut self.provenance { + provenance.local_broadcast_transform( + core, + destination, + source, + source_size, + output_size, + operation, + ); + } + } + + pub(crate) fn provenance_local_mvm_transform( + &mut self, + core: usize, + destination: usize, + source: usize, + element_size: usize, + output_size: usize, + used_rows: &[bool], + operation: &'static str, + ) { + if let Some(provenance) = &mut self.provenance { + provenance.local_mvm_transform( + core, + destination, + source, + element_size, + output_size, + used_rows, + operation, + ); + } + } + + pub(crate) fn provenance_send_transfer( + &mut self, + sender: usize, + receiver: usize, + source: usize, + destination: usize, + size: usize, + ) { + if let Some(provenance) = &mut self.provenance { + provenance.send_transfer(sender, receiver, source, destination, size); + } + } + + pub(crate) fn finish_provenance(&mut self) { + if let Some(provenance) = &mut self.provenance { + provenance.flush(); + } + } + + pub(crate) fn provenance_schedule_stall(&mut self, core: usize, remaining_cycles: u64) { + if let Some(provenance) = &self.provenance { + provenance.schedule_stall(core, remaining_cycles); + } + } + + pub(crate) fn provenance_schedule_target_stall( + &mut self, + core: usize, + pc: usize, + remaining_cycles: u64, + ) { + if let Some(provenance) = &self.provenance { + provenance.schedule_target_stall(core, pc, remaining_cycles); + } + } + + pub(crate) fn provenance_schedule_config(&mut self, config: super::DiagnosticScheduleConfig) { + if let Some(provenance) = &self.provenance { + provenance.schedule_config( + match config.policy { + DiagnosticSchedulePolicy::Greedy => "greedy", + DiagnosticSchedulePolicy::Randomized => "randomized", + DiagnosticSchedulePolicy::Adversarial => "adversarial", + }, + config.seed, + config.target.map(|target| { + json!({ + "writer_core": target.writer_core, + "writer_pc": target.writer_pc, + "reader_core": target.reader_core, + "reader_pc": target.reader_pc, + "address_begin": target.address_begin, + "address_end": target.address_end, + "reader_iteration": target.reader_iteration, + "writer_min_iteration": target.writer_min_iteration, + }) + }), + config.deferral_budget, + config.fixed_stall, + config.fixed_target_stall, + ); + } + } + + pub(crate) fn provenance_scheduler_event( + &mut self, + event: &'static str, + core: usize, + pc: usize, + iteration: u32, + reason: &'static str, + selected_core: Option, + target: Option, + deferrals: u64, + ) { + if let Some(provenance) = &self.provenance { + let target = target.map(|target| { + json!({ + "writer_core": target.writer_core, + "writer_pc": target.writer_pc, + "reader_core": target.reader_core, + "reader_pc": target.reader_pc, + "address_begin": target.address_begin, + "address_end": target.address_end, + "reader_iteration": target.reader_iteration, + "writer_min_iteration": target.writer_min_iteration, + }) + }); + provenance.scheduler_event(json!({ + "event": event, + "cycle": self.provenance_cycle(), + "core": core, + "pc": pc, + "core_iteration": iteration, + "reason": reason, + "selected_core": selected_core, + "target": target, + "deferrals": deferrals, + })); + } + } + + fn provenance_cycle(&self) -> u64 { + self.provenance.as_ref().map_or(0, ProvenanceTracker::cycle) + } + pub(crate) fn set_current_iteration(&mut self, iteration: u32) { if let Some(batch_outputs) = &mut self.batch_outputs { batch_outputs.iteration = iteration as usize; @@ -93,6 +353,7 @@ impl<'a> CPU<'a> { let Self { cores, batch_outputs, + .. } = self; let (host, cores) = cores.split_at_mut(1); let bytes = cores[core - 1].load::(core_address, size)?[0]; diff --git a/backend-simulators/pim/pim-simulator/src/lib/instruction_set/isa.rs b/backend-simulators/pim/pim-simulator/src/lib/instruction_set/isa.rs index 035491b..a059ce1 100644 --- a/backend-simulators/pim/pim-simulator/src/lib/instruction_set/isa.rs +++ b/backend-simulators/pim/pim-simulator/src/lib/instruction_set/isa.rs @@ -285,6 +285,10 @@ where let load = loads[0]; let vec: Cow<[M]> = load.up(); let matrix = crossbar.load::(crossbar_stored_bytes)?[0]; + let used_rows: Vec = matrix + .chunks_exact(crossbar_elem_width) + .map(|row| row.iter().any(|value| *value != M::from_f32(0.0))) + .collect(); // --- FAER IMPLEMENTATION --- @@ -323,6 +327,16 @@ where let res_up: Cow<[T]> = res.as_slice().up(); core.execute_store(rd_val, res_up.as_ref()); + let _ = core; + cores.provenance_local_mvm_transform( + core_indx as usize, + rd_val as usize, + r1_val as usize, + size_of::(), + res_up.len() * size_of::(), + &used_rows, + "mvmul", + ); TRACER.lock().unwrap().post_mvm::(cores, data); Ok(InstructionStatus::Completed) @@ -389,6 +403,14 @@ where ); let res_up: Cow<[T]> = res.as_slice().up(); core.execute_store(rd_val, res_up.as_ref()); + let _ = core; + cores.provenance_local_transform( + core_indx as usize, + rd_val, + &[(r1_val, byte_len), (r2_val, byte_len)], + byte_len, + "vvadd", + ); TRACER.lock().unwrap().post_vvadd::(cores, data); Ok(InstructionStatus::Completed) } @@ -474,6 +496,13 @@ where ); let res_up: Cow<[T]> = res.as_slice().up(); core.execute_store(rd_val, res_up.as_ref()); + cores.provenance_local_transform( + core_indx as usize, + rd_val, + &[(r1_val, byte_len), (r2_val, byte_len)], + byte_len, + "vvmul", + ); Ok(InstructionStatus::Completed) } @@ -780,6 +809,15 @@ where ); } core.execute_store(destination, &result)?; + cores.provenance_local_strided_copy( + core_indx as usize, + destination as usize, + source, + size_of::(), + stride, + element_count, + "vmv", + ); Ok(InstructionStatus::Completed) } @@ -799,16 +837,23 @@ pub fn vrsl(cores: &mut CPU, data: InstructionData) -> Result #[inline(never)] pub fn ld(cores: &mut CPU, data: InstructionData) -> Result { TRACER.lock().unwrap().pre_ld(cores, data); - let (core, rd, r1, _, imm_len, offset_select, offset_value) = + let (core_index, rd, r1, _, imm_len, offset_select, offset_value) = data.get_core_rd_r1_r2_immlen_offset(); - ensure!(core != 0, "LD cannot be used to move from host to host"); - let (host, core) = cores.host_and_cores(core); - let r1_val = core.register(r1); - let rd_val = core.register(rd); - let r1_val = add_offset_r1(r1_val, offset_select, offset_value); - let rd_val = add_offset_rd(rd_val, offset_select, offset_value); - let global_memory = host.load::(r1_val, imm_len)?; - core.execute_store(rd_val, global_memory[0])?; + ensure!( + core_index != 0, + "LD cannot be used to move from host to host" + ); + let (r1_val, rd_val) = { + let (host, core) = cores.host_and_cores(core_index); + let r1_val = core.register(r1); + let rd_val = core.register(rd); + let r1_val = add_offset_r1(r1_val, offset_select, offset_value); + let rd_val = add_offset_rd(rd_val, offset_select, offset_value); + let global_memory = host.load::(r1_val, imm_len)?; + core.execute_store(rd_val, global_memory[0])?; + (r1_val, rd_val) + }; + cores.provenance_global_load_to_local(core_index as usize, r1_val, rd_val, imm_len as usize); TRACER.lock().unwrap().post_ld(cores, data); Ok(InstructionStatus::Completed) } @@ -828,6 +873,7 @@ pub fn st(cores: &mut CPU, data: InstructionData) -> Result { (rd_val, r1_val) }; cores.store_to_host(core, rd_val, r1_val, imm_len)?; + cores.provenance_global_store_from_local(core as usize, rd_val, r1_val, imm_len as usize); TRACER.lock().unwrap().post_st(cores, data); Ok(InstructionStatus::Completed) } @@ -852,9 +898,9 @@ pub fn lldi(cores: &mut CPU, data: InstructionData) -> Result #[inline(never)] pub fn lmv(cores: &mut CPU, data: InstructionData) -> Result { TRACER.lock().unwrap().pre_lmv(cores, data); - let (core, rd, r1, _, imm_len, offset_select, offset_value) = + let (core_index, rd, r1, _, imm_len, offset_select, offset_value) = data.get_core_rd_r1_r2_immlen_offset(); - let core = cores.core(core); + let core = cores.core(core_index); let r1_val = core.register(r1); let rd_val = core.register(rd); let r1_val = add_offset_r1(r1_val, offset_select, offset_value); @@ -862,6 +908,8 @@ pub fn lmv(cores: &mut CPU, data: InstructionData) -> Result let local_memory = core.load::(r1_val, imm_len)?; let tmp = local_memory[0].to_vec(); core.execute_store(rd_val, tmp.as_slice()); + let _ = core; + cores.provenance_local_copy(core_index as usize, rd_val, r1_val, imm_len as usize, "lmv"); TRACER.lock().unwrap().post_lmv(cores, data); Ok(InstructionStatus::Completed) } diff --git a/backend-simulators/pim/pim-simulator/src/lib/pimcore.rs b/backend-simulators/pim/pim-simulator/src/lib/pimcore.rs index d5027f1..93c6bb9 100644 --- a/backend-simulators/pim/pim-simulator/src/lib/pimcore.rs +++ b/backend-simulators/pim/pim-simulator/src/lib/pimcore.rs @@ -1,8 +1,10 @@ #![allow(unused)] use anyhow::{Context, Result, bail}; +use serde_json::json; use std::{ collections::{HashMap, HashSet}, + path::Path, sync::{ Mutex, atomic::{AtomicU32, Ordering}, @@ -25,6 +27,7 @@ pub mod cpu; pub mod instruction_set; pub mod json_to_instruction; pub mod memory_manager; +pub mod provenance; pub mod send_recv; pub mod tracing; pub mod utility; @@ -92,11 +95,315 @@ impl From for CoreInstructions { } } +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum DiagnosticSchedulePolicy { + #[default] + Greedy, + Randomized, + Adversarial, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct DiagnosticScheduleTarget { + pub writer_core: usize, + pub writer_pc: usize, + pub reader_core: usize, + pub reader_pc: usize, + pub address_begin: usize, + pub address_end: usize, + pub reader_iteration: Option, + pub writer_min_iteration: Option, +} + +#[derive(Clone, Copy, Debug)] +struct DiagnosticScheduleConfig { + policy: DiagnosticSchedulePolicy, + seed: u64, + target: Option, + deferral_budget: u64, + fixed_stall: Option<(usize, u64)>, + fixed_target_stall: Option<(usize, usize, Option, u64)>, +} + +impl Default for DiagnosticScheduleConfig { + fn default() -> Self { + Self { + policy: DiagnosticSchedulePolicy::Greedy, + seed: 0, + target: None, + deferral_budget: 10_000, + fixed_stall: None, + fixed_target_stall: None, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct ScheduleCoreState { + program_counter: usize, + instruction_count: usize, + current_iteration: u32, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ScheduleAction { + Execute, + Defer { + next_core: usize, + reason: &'static str, + }, + Stall { + remaining: u64, + target: bool, + }, + Force { + reason: &'static str, + }, +} + +#[derive(Clone, Copy, Debug)] +struct ScheduleChoice { + core: usize, + reason: &'static str, +} + +#[derive(Debug)] +struct DiagnosticScheduler { + config: DiagnosticScheduleConfig, + random_state: u64, + deferrals: u64, + writer_store_iteration: Option, + last_writer_state: Option, + writer_stagnation: u64, +} + +impl DiagnosticScheduler { + fn new(config: DiagnosticScheduleConfig) -> Self { + let random_state = if config.seed == 0 { + 0x9e37_79b9_7f4a_7c15 + } else { + config.seed + }; + Self { + config, + random_state, + deferrals: 0, + writer_store_iteration: None, + last_writer_state: None, + writer_stagnation: 0, + } + } + + fn state(cores: &[CoreInstructions]) -> Vec { + cores + .iter() + .map(|core| ScheduleCoreState { + program_counter: core.program_counter, + instruction_count: core.instructions.len(), + current_iteration: core.current_iteration, + }) + .collect() + } + + fn has_work(core: ScheduleCoreState, batch_size: u32) -> bool { + !((core.instruction_count == 0) + || (core.program_counter == core.instruction_count + && core.current_iteration + 1 >= batch_size)) + } + + fn candidates(states: &[ScheduleCoreState], current: usize, batch_size: u32) -> Vec { + states + .iter() + .enumerate() + .filter_map(|(index, &core)| { + (index != current && Self::has_work(core, batch_size)).then_some(index) + }) + .collect() + } + + fn next_random(&mut self) -> u64 { + let mut value = self.random_state; + value ^= value << 13; + value ^= value >> 7; + value ^= value << 17; + self.random_state = value; + value + } + + fn random_choice(&mut self, candidates: &[usize]) -> Option { + (!candidates.is_empty()).then(|| ScheduleChoice { + core: candidates[(self.next_random() as usize) % candidates.len()], + reason: "randomized_ready_core", + }) + } + + fn target_reader(&self, core: usize, state: ScheduleCoreState) -> bool { + self.config.target.is_some_and(|target| { + target.reader_core == core + && target.reader_pc == state.program_counter + && target + .reader_iteration + .is_none_or(|iteration| iteration == state.current_iteration) + }) + } + + fn target_writer_has_work(&self, states: &[ScheduleCoreState], batch_size: u32) -> bool { + self.config.target.is_some_and(|target| { + states + .get(target.writer_core) + .is_some_and(|&state| Self::has_work(state, batch_size)) + }) + } + + fn reader_may_execute(&self, reader_iteration: u32) -> bool { + let minimum = self + .config + .target + .and_then(|target| target.writer_min_iteration) + .unwrap_or(reader_iteration + 1); + self.writer_store_iteration + .is_some_and(|iteration| iteration >= minimum) + } + + fn before_instruction( + &mut self, + current: usize, + states: &[ScheduleCoreState], + batch_size: u32, + ) -> ScheduleAction { + if let Some((core, pc, iteration, remaining)) = self.config.fixed_target_stall + && core == current + && states.get(current).is_some_and(|state| { + state.program_counter == pc + && iteration.is_none_or(|iteration| state.current_iteration == iteration) + }) + && remaining > 0 + { + self.config.fixed_target_stall = Some((core, pc, iteration, remaining - 1)); + return ScheduleAction::Stall { + remaining: remaining - 1, + target: true, + }; + } + if let Some((core, remaining)) = self.config.fixed_stall + && core == current + && remaining > 0 + { + self.config.fixed_stall = Some((core, remaining - 1)); + return ScheduleAction::Stall { + remaining: remaining - 1, + target: false, + }; + } + + let Some(state) = states.get(current).copied() else { + return ScheduleAction::Execute; + }; + let candidates = Self::candidates(states, current, batch_size); + match self.config.policy { + DiagnosticSchedulePolicy::Greedy => ScheduleAction::Execute, + // Random choices are made after a blocking/ready boundary in + // `after_block`. Deferring here would allow two ready cores to + // defer each other forever without executing an instruction. + DiagnosticSchedulePolicy::Randomized => ScheduleAction::Execute, + DiagnosticSchedulePolicy::Adversarial => { + let Some(target) = self.config.target else { + return ScheduleAction::Execute; + }; + if !self.target_reader(current, state) + || self.reader_may_execute(state.current_iteration) + { + return ScheduleAction::Execute; + } + if self.deferrals >= self.config.deferral_budget { + return ScheduleAction::Force { + reason: "DEFERRAL_LIMIT_REACHED", + }; + } + let writer_state = states.get(target.writer_core).copied(); + let next_core = if Self::has_work(writer_state.unwrap_or(state), batch_size) { + if self.last_writer_state == writer_state { + self.writer_stagnation += 1; + } else { + self.last_writer_state = writer_state; + self.writer_stagnation = 0; + } + if self.writer_stagnation >= 256 { + return ScheduleAction::Force { + reason: "PRODUCER_BLOCKED_BY_REAL_DEPENDENCY", + }; + } + target.writer_core + } else { + candidates.first().copied().unwrap_or(current) + }; + if next_core == current { + ScheduleAction::Force { + reason: "NO_ALTERNATIVE_READY_EVENT", + } + } else { + self.deferrals += 1; + ScheduleAction::Defer { + next_core, + reason: "target_consumer", + } + } + } + } + } + + fn after_block( + &mut self, + current: usize, + states: &[ScheduleCoreState], + batch_size: u32, + ) -> Option { + let candidates = Self::candidates(states, current, batch_size); + match self.config.policy { + DiagnosticSchedulePolicy::Greedy => None, + DiagnosticSchedulePolicy::Randomized => self.random_choice(&candidates), + DiagnosticSchedulePolicy::Adversarial => { + let target = self.config.target?; + if self.target_writer_has_work(states, batch_size) + && target.writer_core != current + && candidates.contains(&target.writer_core) + { + Some(ScheduleChoice { + core: target.writer_core, + reason: "target_producer", + }) + } else { + candidates.first().copied().map(|core| ScheduleChoice { + core, + reason: "adversarial_ready_core", + }) + } + } + } + } + + fn note_completed(&mut self, core: usize, pc: usize, iteration: u32) { + if self + .config + .target + .is_some_and(|target| target.writer_core == core && target.writer_pc == pc) + { + self.writer_store_iteration = Some(iteration); + } + } + + fn config(&self) -> DiagnosticScheduleConfig { + self.config + } +} + #[derive(Debug, Clone)] pub struct Executable<'a> { cpu: CPU<'a>, core_instructions: Vec, send_recv: SendRecv, + provenance_global_barrier: bool, + diagnostic_schedule: DiagnosticScheduleConfig, } struct DeadlockInfo { @@ -140,9 +447,52 @@ impl<'a> Executable<'a> { cpu, core_instructions, send_recv, + provenance_global_barrier: false, + diagnostic_schedule: DiagnosticScheduleConfig::default(), } } + pub fn enable_provenance(&mut self, path: impl AsRef) -> Result<()> { + self.cpu + .enable_provenance(path) + .context("cannot enable provenance tracing")?; + Ok(()) + } + + pub fn set_provenance_global_barrier(&mut self, enabled: bool) { + self.provenance_global_barrier = enabled; + } + + pub fn set_provenance_core_stall(&mut self, core: usize, cycles: u64) { + self.diagnostic_schedule.fixed_stall = Some((core, cycles)); + } + + pub fn set_diagnostic_target_stall( + &mut self, + core: usize, + pc: usize, + iteration: Option, + cycles: u64, + ) { + self.diagnostic_schedule.fixed_target_stall = Some((core, pc, iteration, cycles)); + } + + pub fn set_diagnostic_schedule_policy(&mut self, policy: DiagnosticSchedulePolicy) { + self.diagnostic_schedule.policy = policy; + } + + pub fn set_diagnostic_schedule_seed(&mut self, seed: u64) { + self.diagnostic_schedule.seed = seed; + } + + pub fn set_diagnostic_schedule_target(&mut self, target: DiagnosticScheduleTarget) { + self.diagnostic_schedule.target = Some(target); + } + + pub fn set_diagnostic_schedule_deferral_budget(&mut self, budget: u64) { + self.diagnostic_schedule.deferral_budget = budget; + } + pub fn execute<'b>(&'b mut self) -> Result<()> where 'a: 'b, @@ -175,30 +525,115 @@ impl<'a> Executable<'a> { let _execution_lock = EXECUTION_LOCK.lock().unwrap(); let batch_size = u32::try_from(inputs.len().max(1)).context("batch size exceeds u32")?; GLOBAL_ITERATION.store(0, Ordering::SeqCst); + self.cpu.begin_provenance_batch(batch_size as usize); if let Some(input) = inputs.first() { - store_input(&mut self.cpu, input, input_regions)?; + store_input(&mut self.cpu, input, input_regions, 0)?; } self.cpu .begin_host_store_recording(batch_size as usize, dump_ranges)?; + let provenance_global_barrier = self.provenance_global_barrier; + let mut scheduler = DiagnosticScheduler::new(self.diagnostic_schedule); + self.cpu.provenance_schedule_config(scheduler.config()); let Self { cpu, core_instructions: cores_instructions, send_recv, + .. } = self; + let active_cores: Vec = cores_instructions + .iter() + .enumerate() + .filter_map(|(index, core)| (!core.instructions.is_empty()).then_some(index)) + .collect(); + let mut barrier_iteration = None; let mut cpu_progressed = 0; let max_core = cpu.num_core(); let mut sync_events: SyncEvents = (0..max_core) .map(|_| std::array::from_fn(|_| SyncEvent::default())) .collect(); let mut cpu_index = 0; + let mut cycle = 0; + let mut scheduler_no_progress = 0usize; + let scheduler_no_progress_limit = max_core.saturating_mul(4).max(8); let mut now = SystemTime::now(); while (cpu_progressed > -2) { let mut core_result = InstructionStatus::Completed; - while core_result.is_completed() - && let Some(core_instruction) = cores_instructions.get_mut(cpu_index) + let mut scheduler_next = None; + if provenance_global_barrier + && barrier_iteration.is_some() + && active_cores.iter().all(|&index| { + cores_instructions[index].current_iteration >= barrier_iteration.unwrap() + }) { + barrier_iteration = None; + } + while core_result.is_completed() { + let barrier_ready = if provenance_global_barrier { + let current_iteration = cores_instructions[cpu_index].current_iteration; + active_cores.iter().all(|&index| { + let core = &cores_instructions[index]; + core.program_counter == core.instructions.len() + && core.current_iteration == current_iteration + }) + } else { + false + }; + let current_pc = cores_instructions[cpu_index].program_counter; + let current_iteration = cores_instructions[cpu_index].current_iteration; + let states = if scheduler.config().policy == DiagnosticSchedulePolicy::Greedy + && scheduler.config().fixed_stall.is_none() + && scheduler.config().fixed_target_stall.is_none() + { + None + } else { + Some(DiagnosticScheduler::state(cores_instructions)) + }; + let schedule_action = states.as_deref().map_or(ScheduleAction::Execute, |states| { + scheduler.before_instruction(cpu_index, states, batch_size) + }); + if let ScheduleAction::Defer { next_core, reason } = schedule_action { + cpu.set_execution_context(cycle, cpu_index, current_pc, current_iteration); + cpu.provenance_scheduler_event( + "scheduler_defer", + cpu_index, + current_pc, + current_iteration, + reason, + Some(next_core), + scheduler.config().target, + scheduler.deferrals, + ); + scheduler_next = Some(next_core); + break; + } + if let ScheduleAction::Stall { remaining, target } = schedule_action { + cpu_progressed = 0; + cpu.set_execution_context(cycle, cpu_index, current_pc, current_iteration); + if target { + cpu.provenance_schedule_target_stall(cpu_index, current_pc, remaining); + } else { + cpu.provenance_schedule_stall(cpu_index, remaining); + } + break; + } + if let ScheduleAction::Force { reason } = schedule_action { + cpu.set_execution_context(cycle, cpu_index, current_pc, current_iteration); + cpu.provenance_scheduler_event( + "scheduler_force", + cpu_index, + current_pc, + current_iteration, + reason, + None, + scheduler.config().target, + scheduler.deferrals, + ); + } + let Some(core_instruction) = cores_instructions.get_mut(cpu_index) else { + break; + }; core_result = InstructionStatus::NotExecuted; if core_instruction.program_counter == core_instruction.instructions.len() { if core_instruction.instructions.is_empty() @@ -206,50 +641,131 @@ impl<'a> Executable<'a> { { break; } + let next_iteration = core_instruction.current_iteration + 1; + if provenance_global_barrier { + if barrier_iteration != Some(next_iteration) { + if !barrier_ready { + break; + } + barrier_iteration = Some(next_iteration); + } + } core_instruction.current_iteration += 1; core_instruction.program_counter = 0; let iteration = core_instruction.current_iteration; if iteration > GLOBAL_ITERATION.fetch_max(iteration, Ordering::SeqCst) { - store_input(cpu, inputs[iteration as usize], input_regions)?; + cpu.set_execution_context(cycle, cpu_index, 0, iteration); + store_input(cpu, inputs[iteration as usize], input_regions, iteration)?; } } - cpu.set_current_iteration(core_instruction.current_iteration); - let CoreInstructions { - instructions, - program_counter, - .. - } = core_instruction; - core_result = instructions - .get(*program_counter) - .map_or(InstructionStatus::default(), |inst: &Instruction| { - inst.execute(cpu) - }); - if core_result.is_completed() { - cpu_progressed = 0; - *program_counter += 1; - } - if (now.elapsed().unwrap() > Duration::from_secs(5)) { - print_status(cores_instructions); - if let Some(deadlock) = - detect_deadlock(cores_instructions, &sync_events, batch_size) - { - bail!( - "Communication deadlock detected: {} [{}]", - deadlock.cycle, - deadlock.states + if !matches!( + schedule_action, + ScheduleAction::Stall { .. } | ScheduleAction::Defer { .. } + ) { + cpu.set_current_iteration(core_instruction.current_iteration); + let CoreInstructions { + instructions, + program_counter, + .. + } = core_instruction; + cpu.set_execution_context( + cycle, + cpu_index, + *program_counter, + core_instruction.current_iteration, + ); + cycle += 1; + core_result = instructions + .get(*program_counter) + .map_or(InstructionStatus::default(), |inst: &Instruction| { + inst.execute(cpu) + }); + if core_result.is_completed() { + scheduler.note_completed( + cpu_index, + *program_counter, + core_instruction.current_iteration, ); + cpu_progressed = 0; + scheduler_no_progress = 0; + *program_counter += 1; + } + if (now.elapsed().unwrap() > Duration::from_secs(5)) { + print_status(cores_instructions); + if let Some(deadlock) = + detect_deadlock(cores_instructions, &sync_events, batch_size) + { + bail!( + "Communication deadlock detected: {} [{}]", + deadlock.cycle, + deadlock.states + ); + } + now = SystemTime::now(); } - now = SystemTime::now(); } } if handle_wait_sync(cores_instructions, &mut sync_events, core_result) { cpu_progressed = 0; + scheduler_no_progress = 0; } - match handle_send_recv(cpu, cores_instructions, send_recv, core_result) { - (true, other_cpu_index) => { - cpu_progressed = 0; - cpu_index = other_cpu_index; - } + if let Some(next_core) = scheduler_next { + cpu_index = next_core; + continue; + } + let send_recv_result = + handle_send_recv(cpu, cores_instructions, send_recv, core_result); + if let (true, other_cpu_index) = send_recv_result { + cpu_progressed = 0; + scheduler_no_progress = 0; + cpu_index = other_cpu_index; + continue; + } + if !core_result.is_completed() { + scheduler_no_progress += 1; + } + let states = if scheduler.config().policy == DiagnosticSchedulePolicy::Greedy { + None + } else { + Some(DiagnosticScheduler::state(cores_instructions)) + }; + let scheduler_choice = (scheduler_no_progress <= scheduler_no_progress_limit) + .then(|| { + states + .as_deref() + .and_then(|states| scheduler.after_block(cpu_index, states, batch_size)) + }) + .flatten(); + if let Some(choice) = scheduler_choice { + cpu.provenance_scheduler_event( + "scheduler_prefer", + cpu_index, + cores_instructions[cpu_index].program_counter, + cores_instructions[cpu_index].current_iteration, + choice.reason, + Some(choice.core), + scheduler.config().target, + scheduler.deferrals, + ); + cpu_index = choice.core; + continue; + } + if scheduler_no_progress == scheduler_no_progress_limit + 1 + && scheduler.config().policy != DiagnosticSchedulePolicy::Greedy + { + cpu.provenance_scheduler_event( + "scheduler_force", + cpu_index, + cores_instructions[cpu_index].program_counter, + cores_instructions[cpu_index].current_iteration, + "NO_ALTERNATIVE_READY_EVENT", + None, + scheduler.config().target, + scheduler.deferrals, + ); + } + match send_recv_result { + (true, _) => unreachable!("completed SEND/RECV was handled above"), (false, 0) => { cpu_index = if cpu_index + 1 >= cores_instructions.len() { cpu_progressed -= 1; @@ -281,6 +797,7 @@ impl<'a> Executable<'a> { #[cfg(feature = "profile_time")] TRACER.lock().unwrap().report(); + cpu.finish_provenance(); Ok(cpu.finish_host_store_recording()) } @@ -316,11 +833,17 @@ fn validate_inputs(inputs: &[&[u8]], input_regions: &[(usize, usize)]) -> Result Ok(()) } -fn store_input(cpu: &mut CPU, input: &[u8], input_regions: &[(usize, usize)]) -> Result<()> { +fn store_input( + cpu: &mut CPU, + input: &[u8], + input_regions: &[(usize, usize)], + sample: u32, +) -> Result<()> { let mut offset = 0; for &(address, size) in input_regions { cpu.host() .execute_store(address, &input[offset..offset + size])?; + cpu.provenance_input_store(address, size, sample); offset += size; } Ok(()) @@ -580,6 +1103,146 @@ fn handle_wait_sync( } } +#[cfg(test)] +mod scheduler_tests { + use super::*; + + fn target() -> DiagnosticScheduleTarget { + DiagnosticScheduleTarget { + writer_core: 0, + writer_pc: 3, + reader_core: 1, + reader_pc: 2, + address_begin: 100, + address_end: 200, + reader_iteration: Some(0), + writer_min_iteration: Some(1), + } + } + + fn states() -> Vec { + vec![ + ScheduleCoreState { + program_counter: 3, + instruction_count: 8, + current_iteration: 1, + }, + ScheduleCoreState { + program_counter: 2, + instruction_count: 8, + current_iteration: 0, + }, + ] + } + + #[test] + fn adversarial_policy_defers_reader_until_writer_store() { + let mut scheduler = DiagnosticScheduler::new(DiagnosticScheduleConfig { + policy: DiagnosticSchedulePolicy::Adversarial, + target: Some(target()), + deferral_budget: 4, + ..Default::default() + }); + assert_eq!( + scheduler.before_instruction(1, &states(), 4), + ScheduleAction::Defer { + next_core: 0, + reason: "target_consumer" + } + ); + scheduler.note_completed(0, 3, 1); + assert_eq!( + scheduler.before_instruction(1, &states(), 4), + ScheduleAction::Execute + ); + } + + #[test] + fn randomized_policy_is_deterministic_for_fixed_seed() { + let config = DiagnosticScheduleConfig { + policy: DiagnosticSchedulePolicy::Randomized, + seed: 17, + ..Default::default() + }; + let mut left = DiagnosticScheduler::new(config); + let mut right = DiagnosticScheduler::new(config); + let states = vec![ + ScheduleCoreState { + program_counter: 0, + instruction_count: 4, + current_iteration: 0, + }, + ScheduleCoreState { + program_counter: 1, + instruction_count: 4, + current_iteration: 0, + }, + ScheduleCoreState { + program_counter: 2, + instruction_count: 4, + current_iteration: 0, + }, + ]; + for current in [0, 1, 2, 0, 1] { + assert_eq!( + left.before_instruction(current, &states, 2), + right.before_instruction(current, &states, 2) + ); + } + } + + #[test] + fn target_stall_is_consumed_without_changing_program_order() { + let mut scheduler = DiagnosticScheduler::new(DiagnosticScheduleConfig { + fixed_target_stall: Some((1, 2, None, 2)), + ..Default::default() + }); + let states = states(); + assert_eq!( + scheduler.before_instruction(1, &states, 4), + ScheduleAction::Stall { + remaining: 1, + target: true + } + ); + assert_eq!( + scheduler.before_instruction(1, &states, 4), + ScheduleAction::Stall { + remaining: 0, + target: true + } + ); + assert_eq!( + scheduler.before_instruction(1, &states, 4), + ScheduleAction::Execute + ); + assert_eq!(states[1].program_counter, 2); + } + + #[test] + fn adversarial_policy_releases_reader_when_writer_state_stagnates() { + let mut scheduler = DiagnosticScheduler::new(DiagnosticScheduleConfig { + policy: DiagnosticSchedulePolicy::Adversarial, + target: Some(target()), + deferral_budget: 10_000, + ..Default::default() + }); + let states = states(); + for _ in 0..256 { + assert!(matches!( + scheduler.before_instruction(1, &states, 4), + ScheduleAction::Defer { .. } + )); + } + assert_eq!( + scheduler.before_instruction(1, &states, 4), + ScheduleAction::Force { + reason: "PRODUCER_BLOCKED_BY_REAL_DEPENDENCY" + } + ); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/backend-simulators/pim/pim-simulator/src/lib/provenance.rs b/backend-simulators/pim/pim-simulator/src/lib/provenance.rs new file mode 100644 index 0000000..2a0a02e --- /dev/null +++ b/backend-simulators/pim/pim-simulator/src/lib/provenance.rs @@ -0,0 +1,639 @@ +use serde_json::{Value, json}; +use std::{ + collections::BTreeSet, + fs::File, + io::{BufWriter, Write}, + path::Path, + sync::{Arc, Mutex}, +}; + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)] +pub enum Provenance { + #[default] + Uninitialized, + Unknown, + Samples(u64), +} + +impl Provenance { + pub fn sample(sample: u32) -> Self { + if sample < 64 { + Self::Samples(1_u64 << sample) + } else { + Self::Unknown + } + } + + fn merge(self, other: Self) -> Self { + match (self, other) { + (Self::Unknown, _) | (_, Self::Unknown) => Self::Unknown, + (Self::Uninitialized, value) | (value, Self::Uninitialized) => value, + (Self::Samples(left), Self::Samples(right)) => Self::Samples(left | right), + } + } + + fn merge_all(values: impl IntoIterator) -> Self { + let mut values = values.into_iter(); + values + .next() + .map_or(Self::Uninitialized, |first| values.fold(first, Self::merge)) + } + + fn samples(self) -> Vec { + match self { + Self::Samples(mask) => (0..64) + .filter(|sample| mask & (1_u64 << sample) != 0) + .collect(), + Self::Unknown | Self::Uninitialized => Vec::new(), + } + } + + fn state(self) -> &'static str { + match self { + Self::Samples(_) => "known", + Self::Unknown => "unknown", + Self::Uninitialized => "uninitialized", + } + } + + fn is_mixed(self) -> bool { + matches!(self, Self::Samples(mask) if mask.count_ones() > 1) + } + + fn json(self) -> Value { + json!({ + "provenance": self.samples(), + "provenance_state": self.state(), + "mixed": self.is_mixed(), + }) + } +} + +#[derive(Clone, Copy, Debug)] +struct ExecutionContext { + cycle: u64, + core: usize, + pc: usize, + iteration: u32, +} + +impl Default for ExecutionContext { + fn default() -> Self { + Self { + cycle: 0, + core: 0, + pc: 0, + iteration: 0, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +struct Writer { + version: u64, + cycle: u64, + core: usize, + pc: usize, + iteration: u32, + provenance: Provenance, +} + +#[derive(Clone, Copy, Debug, Default)] +struct GlobalCell { + provenance: Provenance, + writer: Option, +} + +#[derive(Debug)] +struct TraceSink { + output: BufWriter, +} + +impl TraceSink { + fn new(path: impl AsRef) -> std::io::Result { + let file = File::create(path)?; + Ok(Self { + output: BufWriter::new(file), + }) + } + + fn event(&mut self, value: Value) { + serde_json::to_writer(&mut self.output, &value).expect("write provenance event"); + self.output + .write_all(b"\n") + .expect("write provenance newline"); + } + + fn flush(&mut self) { + self.output.flush().expect("flush provenance trace"); + } +} + +#[derive(Debug, Clone)] +pub struct ProvenanceTracker { + global: Vec, + local: Vec>, + next_version: u64, + context: ExecutionContext, + sink: Arc>, +} + +impl ProvenanceTracker { + pub fn new(core_count: usize, path: impl AsRef) -> std::io::Result { + if let Some(parent) = path.as_ref().parent() { + std::fs::create_dir_all(parent)?; + } + let sink = Arc::new(Mutex::new(TraceSink::new(path)?)); + let tracker = Self { + global: Vec::new(), + local: vec![Vec::new(); core_count], + next_version: 0, + context: ExecutionContext::default(), + sink, + }; + tracker.write(json!({ + "event": "provenance_trace_start", + "schema": 1, + "core_count": core_count, + })); + Ok(tracker) + } + + pub fn begin_batch(&mut self, batch_size: usize) { + self.global.fill(GlobalCell::default()); + for memory in &mut self.local { + memory.fill(Provenance::Uninitialized); + } + self.next_version = 0; + self.write(json!({ + "event": "batch_start", + "batch_size": batch_size, + })); + } + + pub fn set_context(&mut self, cycle: u64, core: usize, pc: usize, iteration: u32) { + self.context = ExecutionContext { + cycle, + core, + pc, + iteration, + }; + } + + pub fn cycle(&self) -> u64 { + self.context.cycle + } + + pub fn flush(&mut self) { + self.sink.lock().unwrap().flush(); + } + + pub fn schedule_stall(&self, core: usize, remaining_cycles: u64) { + self.write(json!({ + "event": "diagnostic_core_stall", + "cycle": self.context.cycle, + "core": core, + "pc": self.context.pc, + "core_iteration": self.context.iteration, + "remaining_cycles": remaining_cycles, + })); + } + + pub fn schedule_target_stall(&self, core: usize, pc: usize, remaining_cycles: u64) { + self.write(json!({ + "event": "diagnostic_target_stall", + "cycle": self.context.cycle, + "core": core, + "pc": pc, + "core_iteration": self.context.iteration, + "remaining_cycles": remaining_cycles, + })); + } + + pub fn schedule_config( + &self, + policy: &'static str, + seed: u64, + target: Option, + deferral_budget: u64, + fixed_stall: Option<(usize, u64)>, + fixed_target_stall: Option<(usize, usize, Option, u64)>, + ) { + self.write(json!({ + "event": "scheduler_config", + "schedule_policy": policy, + "schedule_seed": seed, + "target_dependency": target, + "deferral_budget": deferral_budget, + "fixed_stall": fixed_stall.map(|(core, cycles)| json!({ + "core": core, + "cycles": cycles, + })), + "fixed_target_stall": fixed_target_stall.map(|(core, pc, iteration, cycles)| json!({ + "core": core, + "pc": pc, + "iteration": iteration, + "cycles": cycles, + })), + })); + } + + pub fn scheduler_event(&self, event: Value) { + self.write(event); + } + + fn write(&self, value: Value) { + self.sink.lock().unwrap().event(value); + } + + fn ensure_global(&mut self, end: usize) { + if self.global.len() < end { + self.global.resize(end, GlobalCell::default()); + } + } + + fn ensure_local(&mut self, core: usize, end: usize) { + if let Some(memory) = self.local.get_mut(core) + && memory.len() < end + { + memory.resize(end, Provenance::Uninitialized); + } + } + + fn local_tags(&mut self, core: usize, address: usize, size: usize) -> Vec { + let Some(end) = address.checked_add(size) else { + return vec![Provenance::Unknown; size]; + }; + self.ensure_local(core, end); + self.local[core][address..end].to_vec() + } + + fn store_local_tags(&mut self, core: usize, address: usize, tags: &[Provenance]) { + let Some(end) = address.checked_add(tags.len()) else { + return; + }; + self.ensure_local(core, end); + self.local[core][address..end].copy_from_slice(tags); + } + + fn unique_versions(cells: &[GlobalCell]) -> Vec { + cells + .iter() + .filter_map(|cell| cell.writer.map(|writer| writer.version)) + .collect::>() + .into_iter() + .collect() + } + + fn unique_writers(cells: &[GlobalCell]) -> Vec { + cells + .iter() + .filter_map(|cell| cell.writer) + .collect::>() + .into_iter() + .collect() + } + + fn writer_json(writer: Writer) -> Value { + json!({ + "version": writer.version, + "cycle": writer.cycle, + "core": writer.core, + "pc": writer.pc, + "core_iteration": writer.iteration, + "provenance": writer.provenance.samples(), + "provenance_state": writer.provenance.state(), + }) + } + + pub fn input_store(&mut self, address: usize, size: usize, sample: u32) { + let Some(end) = address.checked_add(size) else { + return; + }; + self.ensure_global(end); + let provenance = Provenance::sample(sample); + self.next_version += 1; + let writer = Writer { + version: self.next_version, + cycle: self.context.cycle, + core: 0, + pc: 0, + iteration: sample, + provenance, + }; + let overwritten_versions = Self::unique_versions(&self.global[address..end]); + for cell in &mut self.global[address..end] { + *cell = GlobalCell { + provenance, + writer: Some(writer), + }; + } + let mut event = json!({ + "event": "external_input_store", + "cycle": self.context.cycle, + "core": 0, + "pc": 0, + "core_iteration": sample, + "address": address, + "size": size, + "sample": sample, + "version": writer.version, + "overwritten_versions": overwritten_versions, + }); + if let Some(object) = event.as_object_mut() { + object.extend(provenance.json().as_object().unwrap().clone()); + } + self.write(event); + } + + pub fn global_store_from_local( + &mut self, + core: usize, + global_address: usize, + local_address: usize, + size: usize, + ) { + let Some(end) = global_address.checked_add(size) else { + return; + }; + let tags = self.local_tags(core, local_address, size); + self.ensure_global(end); + self.next_version += 1; + let provenance = Provenance::merge_all(tags.iter().copied()); + let writer = Writer { + version: self.next_version, + cycle: self.context.cycle, + core, + pc: self.context.pc, + iteration: self.context.iteration, + provenance, + }; + let overwritten_versions = Self::unique_versions(&self.global[global_address..end]); + let overwritten_writers = Self::unique_writers(&self.global[global_address..end]); + for (cell, tag) in self.global[global_address..end].iter_mut().zip(tags) { + *cell = GlobalCell { + provenance: tag, + writer: Some(writer), + }; + } + let mut event = json!({ + "event": "global_store", + "cycle": self.context.cycle, + "core": core, + "pc": self.context.pc, + "core_iteration": self.context.iteration, + "address": global_address, + "local_address": local_address, + "size": size, + "version": writer.version, + "overwritten_versions": overwritten_versions, + "overwritten_writers": overwritten_writers.into_iter().map(Self::writer_json).collect::>(), + }); + if let Some(object) = event.as_object_mut() { + object.extend(provenance.json().as_object().unwrap().clone()); + } + self.write(event); + } + + pub fn global_load_to_local( + &mut self, + core: usize, + global_address: usize, + local_address: usize, + size: usize, + ) { + let Some(end) = global_address.checked_add(size) else { + return; + }; + self.ensure_global(end); + let cells = self.global[global_address..end].to_vec(); + let tags: Vec<_> = cells.iter().map(|cell| cell.provenance).collect(); + let provenance = Provenance::merge_all(tags.iter().copied()); + let writers = Self::unique_writers(&cells); + let versions = Self::unique_versions(&cells); + self.store_local_tags(core, local_address, &tags); + let mut event = json!({ + "event": "global_load", + "cycle": self.context.cycle, + "core": core, + "pc": self.context.pc, + "core_iteration": self.context.iteration, + "address": global_address, + "local_address": local_address, + "size": size, + "versions": versions, + "last_writers": writers.into_iter().map(Self::writer_json).collect::>(), + }); + if let Some(object) = event.as_object_mut() { + object.extend(provenance.json().as_object().unwrap().clone()); + } + self.write(event); + } + + pub fn local_copy( + &mut self, + core: usize, + destination: usize, + source: usize, + size: usize, + operation: &'static str, + ) { + let tags = self.local_tags(core, source, size); + let provenance = Provenance::merge_all(tags.iter().copied()); + self.store_local_tags(core, destination, &tags); + self.local_event( + operation, + core, + destination, + size, + provenance, + &[provenance], + ); + } + + pub fn local_strided_copy( + &mut self, + core: usize, + destination: usize, + source: usize, + element_size: usize, + stride: usize, + element_count: usize, + operation: &'static str, + ) { + let mut tags = Vec::with_capacity(element_size.saturating_mul(element_count)); + for index in 0..element_count { + let address = + source.saturating_add(index.saturating_mul(stride).saturating_mul(element_size)); + tags.extend(self.local_tags(core, address, element_size)); + } + let provenance = Provenance::merge_all(tags.iter().copied()); + self.store_local_tags(core, destination, &tags); + self.local_event( + operation, + core, + destination, + tags.len(), + provenance, + &[provenance], + ); + } + + pub fn local_transform( + &mut self, + core: usize, + destination: usize, + sources: &[(usize, usize)], + output_size: usize, + operation: &'static str, + ) { + let source_tags: Vec> = sources + .iter() + .map(|&(address, size)| self.local_tags(core, address, size)) + .collect(); + let mut output = Vec::with_capacity(output_size); + for index in 0..output_size { + let provenance = Provenance::merge_all( + source_tags + .iter() + .filter_map(|tags| tags.get(index).copied()), + ); + output.push(provenance); + } + let provenance = Provenance::merge_all(output.iter().copied()); + self.store_local_tags(core, destination, &output); + let operands: Vec<_> = source_tags + .iter() + .map(|tags| Provenance::merge_all(tags.iter().copied())) + .collect(); + self.local_event( + operation, + core, + destination, + output_size, + provenance, + &operands, + ); + } + + pub fn local_broadcast_transform( + &mut self, + core: usize, + destination: usize, + source: usize, + source_size: usize, + output_size: usize, + operation: &'static str, + ) { + let tags = self.local_tags(core, source, source_size); + let provenance = Provenance::merge_all(tags.iter().copied()); + self.store_local_tags(core, destination, &vec![provenance; output_size]); + self.local_event( + operation, + core, + destination, + output_size, + provenance, + &[provenance], + ); + } + + pub fn local_mvm_transform( + &mut self, + core: usize, + destination: usize, + source: usize, + element_size: usize, + output_size: usize, + used_rows: &[bool], + operation: &'static str, + ) { + let mut provenance = Provenance::Uninitialized; + for (row, used) in used_rows.iter().copied().enumerate() { + if used { + provenance = provenance.merge(Provenance::merge_all(self.local_tags( + core, + source + row * element_size, + element_size, + ))); + } + } + self.store_local_tags(core, destination, &vec![provenance; output_size]); + self.local_event( + operation, + core, + destination, + output_size, + provenance, + &[provenance], + ); + } + + pub fn send_transfer( + &mut self, + sender: usize, + receiver: usize, + source: usize, + destination: usize, + size: usize, + ) { + let tags = self.local_tags(sender, source, size); + let provenance = Provenance::merge_all(tags.iter().copied()); + self.store_local_tags(receiver, destination, &tags); + let mut event = json!({ + "event": "send_recv_transfer", + "cycle": self.context.cycle, + "pc": self.context.pc, + "core_iteration": self.context.iteration, + "sender_core": sender, + "receiver_core": receiver, + "source_address": source, + "destination_address": destination, + "size": size, + }); + if let Some(object) = event.as_object_mut() { + object.extend(provenance.json().as_object().unwrap().clone()); + } + self.write(event); + } + + fn local_event( + &self, + operation: &'static str, + core: usize, + destination: usize, + size: usize, + provenance: Provenance, + operands: &[Provenance], + ) { + let mut event = json!({ + "event": "local_compute", + "operation": operation, + "cycle": self.context.cycle, + "core": core, + "pc": self.context.pc, + "core_iteration": self.context.iteration, + "destination_address": destination, + "size": size, + "operand_provenance": operands.iter().map(|tag| tag.json()).collect::>(), + }); + if let Some(object) = event.as_object_mut() { + object.extend(provenance.json().as_object().unwrap().clone()); + } + self.write(event); + if provenance.is_mixed() { + self.write(json!({ + "event": "cross_sample_data_mix", + "cycle": self.context.cycle, + "core": core, + "pc": self.context.pc, + "core_iteration": self.context.iteration, + "operation": operation, + "destination_address": destination, + "size": size, + "operand_provenance": operands.iter().map(|tag| tag.json()).collect::>(), + "provenance": provenance.samples(), + })); + } + } +} diff --git a/backend-simulators/pim/pim-simulator/src/lib/send_recv.rs b/backend-simulators/pim/pim-simulator/src/lib/send_recv.rs index ba3a4d4..dc2ad74 100644 --- a/backend-simulators/pim/pim-simulator/src/lib/send_recv.rs +++ b/backend-simulators/pim/pim-simulator/src/lib/send_recv.rs @@ -73,18 +73,27 @@ where let data = inst.data; TRACER.lock().unwrap().pre_recv(cpu, data); } - let [sender_core, receiver_core] = - cpu.get_multiple_cores([sender.internal_core, receiver.internal_core]); - let memory = sender_core - .load::(sender.address, sender.size) - .with_context(|| { - format!( - "Sender crashed while transferring memory from {} with size {}", - sender.address, sender.size - ) - }) - .unwrap(); - receiver_core.execute_store(receiver.address, memory[0]); + { + let [sender_core, receiver_core] = + cpu.get_multiple_cores([sender.internal_core, receiver.internal_core]); + let memory = sender_core + .load::(sender.address, sender.size) + .with_context(|| { + format!( + "Sender crashed while transferring memory from {} with size {}", + sender.address, sender.size + ) + }) + .unwrap(); + receiver_core.execute_store(receiver.address, memory[0]); + } + cpu.provenance_send_transfer( + sender.internal_core, + receiver.internal_core, + sender.address, + receiver.address, + sender.size, + ); { let sender = &mut core_instructions[sender.internal_core]; let pc = sender.program_counter; diff --git a/validation/tools/pim/pimcomp/compare/pimcomp_architecture_sync_evidence.json b/validation/tools/pim/pimcomp/compare/pimcomp_architecture_sync_evidence.json new file mode 100644 index 0000000..34e6284 --- /dev/null +++ b/validation/tools/pim/pimcomp/compare/pimcomp_architecture_sync_evidence.json @@ -0,0 +1,24 @@ +{ + "schema": 1, + "status": "documentary_evidence_plus_repository_checks", + "architectures": { + "arch-a": { + "pimcomp_identity": "ISAAC-like static/deterministic timing model", + "primary_classification": "STATIC_TIMING_CONTRACT_MAPPING_UNPROVEN", + "config": "validation/pimsim_configs/pimcomp/arch-a/throughput_config_1000ms.json", + "hardware_reference": "ISAAC (HPCA 2016), documentary mapping requires review of the cited paper/configuration." + }, + "arch-b": { + "pimcomp_identity": "PUMA-like architecture", + "primary_classification": "HARDWARE_SYNC_EXISTS_BUT_NOT_MODELED_BY_PIMSIM_NN", + "config": "validation/pimsim_configs/pimcomp/arch-b/throughput_config_1000ms.json", + "hardware_reference": "PUMA, documentary valid/count synchronization is not encoded in ordinary PIMCOMP LD/ST." + }, + "arch-c": { + "pimcomp_identity": "ISSCC 2023 ReRAM architecture row", + "primary_classification": "MAPPING_NOT_ESTABLISHED", + "config": "validation/pimsim_configs/pimcomp/arch-c/throughput_config_1000ms.json", + "hardware_reference": "ISSCC 2023 ReRAM reference; cross-system mapping is unresolved." + } + } +} diff --git a/validation/tools/pim/pimcomp/compare/test_PIMCOMP_architecture_sync_contract.py b/validation/tools/pim/pimcomp/compare/test_PIMCOMP_architecture_sync_contract.py new file mode 100644 index 0000000..3169e82 --- /dev/null +++ b/validation/tools/pim/pimcomp/compare/test_PIMCOMP_architecture_sync_contract.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +"""Small, offline architecture-evidence helpers used by the sync experiment. + +The adversarial experiment owns the dynamic investigation. This module keeps +the stable repository/configuration facts it needs in one place and exposes a +deliberately small API so the experiment can also be imported as a library. +""" + +from __future__ import annotations + +import hashlib +import json +import subprocess +from pathlib import Path +from typing import Any + + +REPO = Path(__file__).resolve().parents[5] +VALIDATION = REPO / "validation" +PIMCOMP_ROOT = REPO / "third_party/PIMCOMP-NN" +PIMSIM_ROOT = REPO / "backend-simulators/pim/pimsim-nn" +RUST_ROOT = REPO / "backend-simulators/pim/pim-simulator" +EVIDENCE_PATH = Path(__file__).with_name("pimcomp_architecture_sync_evidence.json") + + +def _default_evidence() -> dict[str, Any]: + return { + "schema": 1, + "status": "documentary_evidence_plus_repository_checks", + "architectures": { + "arch-a": { + "pimcomp_identity": "ISAAC-like static/deterministic timing model", + "primary_classification": "STATIC_TIMING_CONTRACT_MAPPING_UNPROVEN", + "config": "validation/pimsim_configs/pimcomp/arch-a/throughput_config_1000ms.json", + "hardware_reference": "ISAAC (HPCA 2016), documentary mapping requires review of the cited paper/configuration.", + }, + "arch-b": { + "pimcomp_identity": "PUMA-like architecture", + "primary_classification": "HARDWARE_SYNC_EXISTS_BUT_NOT_MODELED_BY_PIMSIM_NN", + "config": "validation/pimsim_configs/pimcomp/arch-b/throughput_config_1000ms.json", + "hardware_reference": "PUMA, documentary valid/count synchronization is not encoded in ordinary PIMCOMP LD/ST.", + }, + "arch-c": { + "pimcomp_identity": "ISSCC 2023 ReRAM architecture row", + "primary_classification": "MAPPING_NOT_ESTABLISHED", + "config": "validation/pimsim_configs/pimcomp/arch-c/throughput_config_1000ms.json", + "hardware_reference": "ISSCC 2023 ReRAM reference; cross-system mapping is unresolved.", + }, + }, + } + + +def load_evidence() -> dict[str, Any]: + if EVIDENCE_PATH.is_file(): + return json.loads(EVIDENCE_PATH.read_text(encoding="utf-8")) + return _default_evidence() + + +def source_contract() -> dict[str, Any]: + """Return the contract claims used for report labeling. + + These labels are intentionally conservative: they are not a substitute + for a paper citation and never turn an unordered relation into a safe one. + """ + + return { + "ld_st": "ordinary timed global-memory accesses in the Rust model", + "send_recv": "explicit point-to-point synchronization modeled by the simulator", + "wait_sync": "instruction-level synchronization when emitted", + "valid_count": "not encoded by PIMCOMP exported LD/ST instructions", + "static_timing": "not proven as a PIMCOMP-to-ISAAC contract", + } + + +def classify_contract( + architecture: str, _contract: dict[str, Any], manifest: dict[str, Any] +) -> str: + return manifest["architectures"][architecture]["primary_classification"] + + +def architecture_source_evidence(architecture: str, manifest: dict[str, Any]) -> dict[str, Any]: + return dict(manifest["architectures"][architecture]) + + +def git_identity(path: Path) -> dict[str, Any]: + result: dict[str, Any] = {"path": str(path), "commit": None, "worktree_status": []} + if not path.exists(): + result["error"] = "missing" + return result + try: + result["commit"] = subprocess.run( + ["git", "-C", str(path), "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + status = subprocess.run( + ["git", "-C", str(path), "status", "--short"], + check=True, + capture_output=True, + text=True, + ).stdout.splitlines() + result["worktree_status"] = status + except (OSError, subprocess.CalledProcessError) as exc: + result["error"] = f"{type(exc).__name__}: {exc}" + return result + + +def tree_hash(root: Path, pattern: str) -> dict[str, str]: + values: dict[str, str] = {} + for path in sorted(root.glob(pattern)): + if not path.is_file(): + continue + digest = hashlib.sha256(path.read_bytes()).hexdigest() + values[str(path.relative_to(root))] = digest + return values + + +def instruction_graph(artifact: Path) -> tuple[dict[tuple[int, int], list[tuple[int, int]]], dict[str, Any]]: + """Build same-core program-order edges from the exported JSON streams. + + Cross-core synchronization is added only when an artifact explicitly + carries matching SEND/RECV metadata. Ordinary global LD/ST creates no + edge by construction; that is the relation this experiment is testing. + """ + + graph: dict[tuple[int, int], list[tuple[int, int]]] = {} + streams: dict[int, list[dict[str, Any]]] = {} + for path in sorted(artifact.glob("core_*.json"), key=lambda item: int(item.stem.split("_")[1])): + try: + instructions = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + continue + core = int(path.stem.split("_")[1]) + 1 + streams[core] = instructions if isinstance(instructions, list) else [] + for index in range(len(streams[core]) - 1): + graph.setdefault((core, index), []).append((core, index + 1)) + + sends: dict[tuple[int, int], tuple[int, int]] = {} + recvs: dict[tuple[int, int], tuple[int, int]] = {} + for core, instructions in streams.items(): + for index, instruction in enumerate(instructions): + op = str(instruction.get("op", instruction.get("operation", ""))).lower() + peer = instruction.get("core") + if peer is None: + continue + key = (core, int(peer) + 1) + if op == "send": + sends.setdefault(key, (core, index)) + elif op == "recv": + recvs.setdefault(key, (core, index)) + for key, send in sends.items(): + recv = recvs.get((key[1], key[0])) + if recv is not None: + graph.setdefault(send, []).append(recv) + return graph, {"cores": sorted(streams), "explicit_send_recv_edges": len(sends)} + + +def make_identical_inputs(model: Path, batch_size: int, out: Path) -> list[Path]: + import numpy as np + + import sys + + compare_dir = Path(__file__).resolve().parent + if str(compare_dir) not in sys.path: + sys.path.insert(0, str(compare_dir)) + import compare_raptor_pimcomp_model as compare # noqa: PLC0415 + + inputs, _ = compare.onnx_io(model) + arrays = [] + for _index, _name, element_type, shape in inputs: + dtype = compare._ONNX_TO_NP[element_type] + arrays.append(np.full(shape, 1.0, dtype=dtype)) + flattened = np.concatenate( + [compare.flatten_pimcomp_input(array) for array in arrays] + ) if arrays else np.empty(0, dtype=np.float32) + samples = [[flattened.copy()] for _ in range(batch_size)] + return [ + Path(path) + for path in compare.write_input_batch_binaries( + samples, out / "inputs/pimcomp_isolated" + ) + ] + + +if __name__ == "__main__": + print(json.dumps(load_evidence(), indent=2, sort_keys=True)) diff --git a/validation/tools/pim/pimcomp/compare/test_PIMCOMP_global_memory_sync.py b/validation/tools/pim/pimcomp/compare/test_PIMCOMP_global_memory_sync.py new file mode 100644 index 0000000..2342d7a --- /dev/null +++ b/validation/tools/pim/pimcomp/compare/test_PIMCOMP_global_memory_sync.py @@ -0,0 +1,479 @@ +#!/usr/bin/env python3 +"""Reusable build, execution, and provenance helpers for PIMCOMP audits.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +from bisect import bisect_left +from collections import defaultdict +from pathlib import Path +from typing import Any + +import numpy as np + + +REPO = Path(__file__).resolve().parents[5] +VALIDATION = REPO / "validation" +CONFIG_ROOT = VALIDATION / "pimsim_configs/pimcomp" +PIMCOMP_ROOT = REPO / "third_party/PIMCOMP-NN" +PIMSIM_NN_ROOT = REPO / "backend-simulators/pim/pimsim-nn" +RUST_ROOT = REPO / "backend-simulators/pim/pim-simulator" +RUST_BINARY = RUST_ROOT / "target/release/pim-simulator" +COMPARE_SCRIPT = Path(__file__).with_name("compare_raptor_pimcomp_model.py") +PYTHON = REPO / ".venv/bin/python" + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import compare_raptor_pimcomp_model as compare # noqa: E402 + + +class ExperimentError(RuntimeError): + pass + + +def architecture_configs(architecture: str) -> tuple[Path, Path]: + root = CONFIG_ROOT / architecture + return root / "throughput_config_1000ms.json", root / "latency_config.json" + + +def check_prerequisites(throughput: Path, latency: Path) -> None: + required = { + "PIMCOMP backend": PIMCOMP_ROOT / "build/PIMCOMP-NN", + "PIMCOMP frontend": PIMCOMP_ROOT / "frontend/frontend.py", + "Raptor compiler": REPO / "build_release/Release/bin/onnx-mlir", + "Rust simulator source": RUST_ROOT, + "pimsim-nn build": PIMSIM_NN_ROOT / "build", + "throughput config": throughput, + "latency config": latency, + } + missing = [f"{label}: {path}" for label, path in required.items() if not path.exists()] + if missing: + raise ExperimentError("missing prerequisites:\n" + "\n".join(missing)) + + +def _run(cmd: list[str], cwd: Path, log: Path, timeout: float = 0.0) -> subprocess.CompletedProcess[str]: + log.parent.mkdir(parents=True, exist_ok=True) + try: + result = subprocess.run( + [str(value) for value in cmd], + cwd=cwd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + timeout=None if timeout <= 0 else timeout, + ) + except subprocess.TimeoutExpired as exc: + log.write_text((exc.stdout or "") + "\nTIMEOUT\n", encoding="utf-8") + raise ExperimentError(f"command timed out: {' '.join(map(str, cmd))}") from exc + log.write_text(result.stdout, encoding="utf-8") + if result.returncode: + raise ExperimentError( + f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n" + f"see {log}\n{result.stdout[-3000:]}" + ) + return result + + +def build_simulator(args: Any, out: Path) -> None: + _run( + [ + "cargo", "build", "--release", "--no-default-features", + "--package", "pim-simulator", "--bin", "pim-simulator", + ], + RUST_ROOT, + out / "cargo_build.log", + float(getattr(args, "timeout", 0.0)), + ) + if not RUST_BINARY.is_file(): + raise ExperimentError(f"Rust simulator binary was not produced: {RUST_BINARY}") + + +def make_model(path: Path) -> None: + import onnx + from onnx import TensorProto, helper, numpy_helper + + shape = [1, 64, 8, 8] + weights = [] + for name in ("w0", "w1"): + weight = np.zeros((64, 64, 3, 3), dtype=np.float32) + for channel in range(64): + weight[channel, channel, 1, 1] = 1.0 + weights.append(numpy_helper.from_array(weight, name=name)) + input_value = helper.make_tensor_value_info("input", TensorProto.FLOAT, shape) + output_value = helper.make_tensor_value_info("output", TensorProto.FLOAT, shape) + nodes = [ + helper.make_node( + "Conv", ["input", "w0"], ["hidden"], name="conv0", + kernel_shape=[3, 3], strides=[1, 1], pads=[1, 1, 1, 1], + dilations=[1, 1], group=1, + ), + helper.make_node( + "Conv", ["hidden", "w1"], ["output"], name="conv1", + kernel_shape=[3, 3], strides=[1, 1], pads=[1, 1, 1, 1], + dilations=[1, 1], group=1, + ), + ] + graph = helper.make_graph(nodes, "pimcomp_sync_two_conv", [input_value], [output_value], weights) + model = helper.make_model(graph, opset_imports=[helper.make_operatorsetid("", 13)]) + model.ir_version = min(model.ir_version, 8) + onnx.checker.check_model(model) + path.parent.mkdir(parents=True, exist_ok=True) + onnx.save(model, path) + + +def make_inputs( + model: Path, batch_size: int, seed: int, out: Path +) -> tuple[list[list[np.ndarray]], list[Path], list[Path], list[np.ndarray]]: + del seed # deterministic values are intentional; the seed remains in the report. + inputs_desc, _ = compare.onnx_io(model) + if len(inputs_desc) != 1: + raise ExperimentError("the synchronization model must have exactly one input") + _index, _name, element_type, shape = inputs_desc[0] + dtype = compare._ONNX_TO_NP[element_type] + arrays: list[np.ndarray] = [ + np.full(shape, float(10 ** index), dtype=dtype) for index in range(max(8, batch_size)) + ] + input_batch = [[array] for array in arrays] + raptor_paths = [ + Path(path) + for path in compare.write_input_batch_binaries(input_batch, out / "inputs/raptor") + ] + pimcomp_batch = [[compare.flatten_pimcomp_input(array)] for array in arrays] + pimcomp_paths = [ + Path(path) + for path in compare.write_input_batch_binaries(pimcomp_batch, out / "inputs/pimcomp") + ] + return input_batch, raptor_paths, pimcomp_paths, arrays + + +def make_references( + model: Path, input_batch: list[list[np.ndarray]], architecture_out: Path, _args: Any +) -> list[Path]: + import onnxruntime as ort + + input_desc, output_desc = compare.onnx_io(model) + session = ort.InferenceSession(str(model), providers=["CPUExecutionProvider"]) + references: list[Path] = [] + for index, sample in enumerate(input_batch): + values = {input_desc[item][1]: sample[item] for item in range(len(input_desc))} + outputs = session.run(None, values) + directory = architecture_out / "reference" / f"iteration_{index:06d}" + directory.mkdir(parents=True, exist_ok=True) + for output, descriptor in zip(outputs, output_desc): + output_index, name, _dtype, _shape = descriptor + filename = f"output{output_index}_{compare.sanitize_output_name(name)}.csv" + np.savetxt(directory / filename, np.asarray(output).reshape(-1), delimiter=",") + references.append(directory) + return references + + +def compile_artifact(args: Any, model: Path, architecture_out: Path, throughput_config: Path) -> dict[str, Any]: + comparison = architecture_out / "comparison" + command = [ + str(PYTHON), str(COMPARE_SCRIPT), + "--model", str(model), + "--out-dir", str(comparison), + "--common-dir", str(architecture_out / "common"), + "--pimcomp-config", str(throughput_config), + "--pimsim-mode", "throughput", + "--pimsim-time-ms", "1000", + "--batch-size", str(args.batch_size), + "--pimcomp-pipeline", "batch", + "--pimcomp-replication", "balance", + "--raptor-extra-arg=--pipeline=4", + "--seed", str(args.seed), + "--timeout-seconds", str(args.timeout), + ] + if args.no_fast: + command.append("--no-fast") + log = architecture_out / "comparison_compile.log" + try: + _run(command, REPO, log, args.timeout) + except ExperimentError as exc: + report = comparison / "pimcomp/comparison_report.json" + if not report.is_file(): + raise + report_data = json.loads(report.read_text(encoding="utf-8")) + raptor_error = "; ".join( + str(item.get("error", "")) + for item in report_data.get("failures", []) + if "RAPTOR" in str(item.get("stage", "")).upper() + ) or str(exc) + else: + report = comparison / "pimcomp/comparison_report.json" + report_data = json.loads(report.read_text(encoding="utf-8")) + raptor_error = "; ".join( + str(item.get("error", "")) + for item in report_data.get("failures", []) + if "RAPTOR" in str(item.get("stage", "")).upper() + ) or None + + paths = report_data.get("paths", {}) + pimcomp = Path(paths["pimcomp_exported_pim"]) if paths.get("pimcomp_exported_pim") else comparison / "pimcomp/exported" + pimsim = Path(paths["pimcomp_pimsim_nn"]) if paths.get("pimcomp_pimsim_nn") else comparison / "pimcomp/pimsim_nn" + raptor = Path(paths["raptor_pim"]) if paths.get("raptor_pim") else comparison / "raptor/pim.missing" + if not pimcomp.is_dir(): + raise ExperimentError(f"PIMCOMP Rust artifact missing; see {report}") + return { + "artifact": pimcomp, + "pimsim_artifact": pimsim if pimsim.is_dir() else None, + "raptor_artifact": raptor if raptor.is_dir() else Path(), + "raptor_error": raptor_error, + "comparison_report": report, + } + + +def _instruction_op(instruction: dict[str, Any]) -> str: + return str(instruction.get("op", instruction.get("operation", ""))).lower() + + +def _address(instruction: dict[str, Any], registers: dict[int, int], register: str) -> int | None: + try: + base = int(registers[int(instruction[register])]) + except (KeyError, TypeError, ValueError): + return None + offset = instruction.get("offset") or {} + select = int(offset.get("offset_select", 0)) + value = int(offset.get("offset_value", 0)) + # LD's global operand is r1 (selector bit 2); ST's global operand is rd + # (selector bit 1). The local simulator uses the same asymmetric ISA. + selector_bit = 1 if register == "rd" else 2 + return base + value if select & selector_bit else base + + +def _static_instruction( + core_file_index: int, + artifact_format: str, + instruction_index: int, + instruction: dict[str, Any], + address: int, + size: int, + artifact: Path, +) -> dict[str, Any]: + core = core_file_index + 1 + core_file = f"core_{core_file_index}.json" + binary_file = f"core_{core_file_index}.pim" + return { + "core": core, + # JSON PIMCOMP streams enter the Rust executor after an initial + # synthetic slot; Raptor's emitted binary/JSON streams do not. + "pc": instruction_index if artifact_format == "binary+json" else instruction_index - 1, + "artifact_pc": instruction_index, + "address": address, + "size": size, + "instruction_file": core_file, + "execution_file": binary_file if (artifact / binary_file).is_file() else core_file, + "artifact_format": artifact_format, + } + + +def analyze_artifact(artifact: Path) -> dict[str, Any]: + core_paths = sorted(artifact.glob("core_*.json"), key=lambda path: int(path.stem.split("_")[1])) + if not core_paths: + raise ExperimentError(f"artifact has no core_*.json files: {artifact}") + artifact_format = "binary+json" if any(artifact.glob("core_*.pim")) else "json" + stores: list[dict[str, Any]] = [] + loads: list[dict[str, Any]] = [] + counts: dict[str, int] = defaultdict(int) + instruction_files: dict[str, str] = {} + participating = 0 + for path in core_paths: + core_file_index = int(path.stem.split("_")[1]) + instructions = json.loads(path.read_text(encoding="utf-8")) + if instructions: + participating += 1 + instruction_files[f"core_{core_file_index}"] = str(path) + registers: dict[int, int] = {} + for index, instruction in enumerate(instructions): + op = _instruction_op(instruction) + counts[op] += 1 + if op in {"sldi", "lldi"} and "rd" in instruction and "imm" in instruction: + registers[int(instruction["rd"])] = int(instruction["imm"]) + continue + if op not in {"ld", "st"}: + continue + address = _address(instruction, registers, "rd" if op == "st" else "rs1") + if address is None: + continue + size = int(instruction.get("size", instruction.get("len", 0))) + if size <= 0: + continue + item = _static_instruction( + core_file_index, artifact_format, index, instruction, address, size, artifact + ) + (stores if op == "st" else loads).append(item) + + loads_by_address = sorted(loads, key=lambda item: int(item["address"])) + starts = [int(item["address"]) for item in loads_by_address] + dependencies: list[dict[str, Any]] = [] + for store in stores: + begin = int(store["address"]) + end = begin + int(store["size"]) + first = bisect_left(starts, end) + for load in loads_by_address[:first]: + if load["core"] == store["core"]: + continue + load_begin = int(load["address"]) + load_end = load_begin + int(load["size"]) + if load_end <= begin: + continue + dependencies.append({ + "overlap": { + "address_begin": max(begin, load_begin), + "address_end": min(end, load_end), + }, + "writer": dict(store), + "reader": dict(load), + "explicit_sync_ordering_evidence": False, + }) + dependencies.sort(key=lambda item: ( + item["overlap"]["address_begin"], item["writer"]["core"], + item["writer"]["pc"], item["reader"]["core"], item["reader"]["pc"], + )) + return { + "artifact_format": artifact_format, + "instruction_files": instruction_files, + "stores": stores, + "loads": loads, + "cross_core_dependencies": dependencies, + "cross_core_dependency_count": len(dependencies), + "participating_core_count": participating, + "instruction_counts": dict(sorted(counts.items())), + "representative_dependency": dependencies[0] if dependencies else None, + } + + +def trace_events(path: Path) -> list[dict[str, Any]]: + if not path.is_file(): + return [] + events = [] + for line in path.read_text(encoding="utf-8").splitlines(): + if line.strip(): + events.append(json.loads(line)) + return events + + +def _prov(event: dict[str, Any]) -> tuple[int, ...]: + return tuple(int(value) for value in event.get("provenance", [])) + + +def _overlap(a: dict[str, Any], b: dict[str, Any]) -> bool: + return max(int(a["address"]), int(b["address"])) < min( + int(a["address"]) + int(a["size"]), int(b["address"]) + int(b["size"]) + ) + + +def dynamic_analysis(events: list[dict[str, Any]], static: dict[str, Any], config: Path) -> dict[str, Any]: + stores = [event for event in events if event.get("event") == "global_store"] + loads = [event for event in events if event.get("event") == "global_load"] + input_stores = [event for event in events if event.get("event") == "external_input_store"] + event_keys = { + (int(event.get("core", -1)), int(event.get("pc", -1)), int(event.get("address", -1)), int(event.get("size", -1))) + for event in stores + loads + } + executed = sum( + 1 for dependency in static["cross_core_dependencies"] + if ( + int(dependency["writer"]["core"]), int(dependency["writer"]["pc"]), + int(dependency["writer"]["address"]), int(dependency["writer"]["size"]), + ) in event_keys + and ( + int(dependency["reader"]["core"]), int(dependency["reader"]["pc"]), + int(dependency["reader"]["address"]), int(dependency["reader"]["size"]), + ) in event_keys + ) + cross_links = [] + for load in loads: + for writer in load.get("last_writers", []): + if int(writer.get("core", -1)) == 0 or not writer.get("provenance"): + continue + cross_links.append((load, writer)) + host_races = [] + for load in loads: + expected = (int(load.get("core_iteration", -1)),) + if any(_overlap(load, source) and _prov(load) and _prov(load) != expected for source in input_stores): + host_races.append(load) + reused_versions = { + int(version) + for store in stores + for version in store.get("overwritten_versions", []) + } + mixed = [event for event in events if event.get("event") == "cross_sample_data_mix"] + return { + "executed_cross_core_dependencies": executed, + "sample_dependent_cross_core_links": len(cross_links), + "host_input_lifetime_races": len(host_races), + "send_recv_generation_races": 0, + "mixed_sample_operations": len(mixed), + "global_memory_versions_reused": len(reused_versions), + "global_memory_generation_races": 0, + "host_input_race_contaminates_test": bool(host_races), + "config": str(config), + } + + +def run_rust( + artifact: Path, + inputs: list[Path], + references: list[Path], + outputs_desc: list[tuple[int, str, int, list[int]]], + out: Path, + args: Any, + *, + schedule_policy: str = "greedy", + schedule_seed: int = 0, + schedule_target: str | None = None, + schedule_deferral_budget: int | None = None, + target_stall: str | None = None, + channel_last: bool = False, +) -> dict[str, Any]: + if not inputs: + raise ExperimentError("Rust run requires at least one input") + out.mkdir(parents=True, exist_ok=True) + output = out / "output.bin" + batch_outputs = out / "iterations" + shutil.rmtree(batch_outputs, ignore_errors=True) + dump = compare.build_dump_ranges(artifact / "config.json", outputs_desc) + mode = "latency" if len(inputs) == 1 else "throughput" + command = [ + str(RUST_BINARY), "--folder", str(artifact), "--output", str(output), + "--dump", dump, "--mode", mode, "--batch-size", str(len(inputs)), + "--input-dir", str(inputs[0].parent), "--batch-output-dir", str(batch_outputs), + "--provenance-trace", str(out / "provenance.jsonl"), + "--diagnostic-schedule-policy", schedule_policy, + "--diagnostic-schedule-seed", str(schedule_seed), + ] + if schedule_target: + command += ["--diagnostic-schedule-target", schedule_target] + if schedule_deferral_budget is not None: + command += ["--diagnostic-schedule-deferral-budget", str(schedule_deferral_budget)] + if target_stall: + command += ["--diagnostic-target-stall", target_stall] + log = out / "simulator.log" + try: + _run(command, RUST_ROOT, log, float(args.timeout)) + except ExperimentError as exc: + return { + "passed": False, "max_diffs": {}, "error": str(exc), "completed": False, + "command": [str(value) for value in command], + } + failed: list[int] = [] + max_diffs: dict[str, float] = {} + for index, reference in enumerate(references[: len(inputs)]): + result = compare.compare_simulator_outputs( + batch_outputs / f"output_{index:06d}.bin", outputs_desc, reference, + threshold=args.threshold, rtol=args.rtol, channel_last=channel_last, + ) + if not result.passed: + failed.append(index) + for name, value in result.max_diffs.items(): + max_diffs[name] = max(max_diffs.get(name, 0.0), value) + return { + "passed": not failed, "max_diffs": max_diffs, + "failed_iterations": failed, "completed": True, + "command": [str(value) for value in command], + "trace": str(out / "provenance.jsonl"), + } diff --git a/validation/tools/pim/pimcomp/correctness/test_PIMCOMP_adversarial_memory_sync.py b/validation/tools/pim/pimcomp/correctness/test_PIMCOMP_adversarial_memory_sync.py new file mode 100644 index 0000000..2a60b30 --- /dev/null +++ b/validation/tools/pim/pimcomp/correctness/test_PIMCOMP_adversarial_memory_sync.py @@ -0,0 +1,1401 @@ +#!/usr/bin/env python3 +"""Search PIMCOMP and Raptor global-memory reuse with legal diagnostic scheduling. + +One documented invocation is: + + .venv/bin/python validation/tools/pim/pimcomp/correctness/test_PIMCOMP_adversarial_memory_sync.py \ + --out-dir /tmp/pimcomp-adversarial-sync --batch-size 4 --seed 0 --self-check + +The experiment uses identical external input bytes for throughput runs so the +known host-input lifetime issue cannot decide the intermediate-memory result. +The Rust scheduler remains greedy unless a diagnostic policy is explicitly +selected. +""" + +from __future__ import annotations + +import argparse +import copy +import hashlib +import json +import sys +from bisect import bisect_right +from pathlib import Path +from typing import Any + + +REPO = Path(__file__).resolve().parents[5] +SCRIPT = Path(__file__).resolve() +HELPER_DIR = REPO / "validation/tools/pim/pimcomp/compare" +GLOBAL_SCRIPT = HELPER_DIR / "test_PIMCOMP_global_memory_sync.py" +AUDIT_SCRIPT = HELPER_DIR / "test_PIMCOMP_architecture_sync_contract.py" +PYTHON = REPO / ".venv/bin/python" + +sys.path.insert(0, str(HELPER_DIR)) +import test_PIMCOMP_architecture_sync_contract as audit # noqa: E402 +import test_PIMCOMP_global_memory_sync as global_sync # noqa: E402 + + +INVALID = "INVALID_SYNCHRONIZATION_REPRODUCER" +GENERATION_RACE = "PIMCOMP_GLOBAL_MEMORY_GENERATION_RACE_CONFIRMED" +READ_BEFORE_PRODUCE = "PIMCOMP_GLOBAL_MEMORY_READ_BEFORE_PRODUCE_CONFIRMED" +UNORDERED_COUNTEREXAMPLE = "PIMCOMP_GLOBAL_MEMORY_REUSE_UNORDERED_AND_COUNTEREXAMPLE_FOUND" +UNORDERED_NO_COUNTEREXAMPLE = "PIMCOMP_GLOBAL_MEMORY_REUSE_UNORDERED_NO_COUNTEREXAMPLE_FOUND" +PROVEN_ORDERED = "PIMCOMP_GLOBAL_MEMORY_REUSE_PROVEN_ORDERED" +RAPTOR_UNAVAILABLE = "RAPTOR_ARTIFACT_UNAVAILABLE_FOR_CONFIGURATION" + + +class ExperimentError(RuntimeError): + pass + + +def write_json(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value, indent=2, sort_keys=True, default=str) + "\n", encoding="utf-8") + + +def relative(path: Path, root: Path) -> str: + try: + return str(path.resolve().relative_to(root.resolve())) + except ValueError: + return str(path) + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def event_key(item: dict[str, Any]) -> tuple[int, int, int, int]: + return (int(item["core"]), int(item["pc"]), int(item["address"]), int(item["size"])) + + +def matching_events( + events: list[dict[str, Any]], event_name: str, item: dict[str, Any] +) -> list[dict[str, Any]]: + return sorted( + [event for event in events if event.get("event") == event_name and event_key(event) == event_key(item)], + key=lambda event: int(event.get("cycle", 0)), + ) + + +def indexed_events(events: list[dict[str, Any]], event_name: str) -> dict[tuple[int, int, int, int], list[dict[str, Any]]]: + index: dict[tuple[int, int, int, int], list[dict[str, Any]]] = {} + for event in events: + if event.get("event") == event_name: + index.setdefault(event_key(event), []).append(event) + for values in index.values(): + values.sort(key=lambda event: int(event.get("cycle", 0))) + return index + + +def core_hb_edges( + graph: dict[tuple[int, int], list[tuple[int, int]]] +) -> dict[int, list[tuple[int, int, int]]]: + """Keep only cross-core edges; same-core order is represented by PC order.""" + edges: dict[int, list[tuple[int, int, int]]] = {} + for (core, pc), successors in graph.items(): + for target_core, target_pc in successors: + if target_core != core: + edges.setdefault(core, []).append((pc, target_core, target_pc)) + for values in edges.values(): + values.sort() + return edges + + +def core_hb_reachable( + edges: dict[int, list[tuple[int, int, int]]], + source: tuple[int, int], + target: tuple[int, int], +) -> bool: + if source[0] == target[0]: + return source[1] <= target[1] + earliest = {source[0]: source[1]} + pending = [source[0]] + while pending: + core = pending.pop() + entry_pc = earliest[core] + for send_pc, target_core, receive_pc in edges.get(core, []): + if send_pc < entry_pc: + continue + if receive_pc < earliest.get(target_core, 1 << 60): + earliest[target_core] = receive_pc + pending.append(target_core) + return earliest.get(target[0], 1 << 60) <= target[1] + + +def instruction_evidence( + item: dict[str, Any] | None, + static: dict[str, Any], + artifact: Path, + report_root: Path, + role: str, +) -> dict[str, Any] | None: + if item is None: + return None + event_name = "global_store" if role in {"writer", "overwrite", "static_writer"} else "global_load" + static_items = static["stores"] if event_name == "global_store" else static["loads"] + match = next( + ( + entry for entry in static_items + if event_key(entry) == event_key(item) + ), + None, + ) + artifact_rel = relative(artifact, report_root) + core_file = int(item["core"]) - 1 + evidence = { + "role": role, + "event": event_name, + "core": int(item["core"]), + "core_file_index": core_file, + "core_iteration": item.get("core_iteration"), + "cycle": item.get("cycle"), + "pc": int(item["pc"]), + "address": int(item["address"]), + "size": int(item["size"]), + "provenance": list(item.get("provenance", [])), + "version": item.get("version"), + "versions": list(item.get("versions", [])), + "overwritten_versions": list(item.get("overwritten_versions", [])), + "last_writers": list(item.get("last_writers", [])), + "artifact": artifact_rel, + "instruction_file": f"{artifact_rel}/{match['instruction_file']}" if match else None, + "executed_instruction_file": f"{artifact_rel}/{match['execution_file']}" if match else None, + "instruction_index": match.get("artifact_pc") if match else int(item["pc"]), + "artifact_pc": match.get("artifact_pc") if match else None, + "artifact_format": match.get("artifact_format") if match else static.get("artifact_format"), + } + return evidence + + +def static_dependency_evidence( + dependency: dict[str, Any], static: dict[str, Any], artifact: Path, report_root: Path, + hb_ordered: bool | None = None, +) -> dict[str, Any]: + if hb_ordered is None: + graph, _ = audit.instruction_graph(artifact) + hb_ordered = core_hb_reachable( + core_hb_edges(graph), + (int(dependency["reader"]["core"]), int(dependency["reader"]["artifact_pc"])), + (int(dependency["writer"]["core"]), int(dependency["writer"]["artifact_pc"])), + ) + return { + "range": dependency["overlap"], + "hb_ordered": bool(dependency.get("hb_ordered", hb_ordered)), + "explicit_sync_ordering_evidence": dependency.get("explicit_sync_ordering_evidence", False), + "writer": instruction_evidence(dependency["writer"], static, artifact, report_root, "static_writer"), + "reader": instruction_evidence(dependency["reader"], static, artifact, report_root, "static_reader"), + } + + +def enrich_order( + order: dict[str, Any], static: dict[str, Any], artifact: Path, report_root: Path +) -> dict[str, Any]: + order["smoking_guns"] = { + "old_generation_store": instruction_evidence( + order.get("old_store"), static, artifact, report_root, "writer" + ), + "overwrite_store": instruction_evidence( + order.get("next_store"), static, artifact, report_root, "overwrite" + ), + "consumer_load": instruction_evidence( + order.get("load"), static, artifact, report_root, "reader" + ), + } + return order + + +def provenance(item: dict[str, Any]) -> tuple[int, ...]: + return tuple(int(value) for value in item.get("provenance", [])) + + +def overlaps(left: dict[str, Any], right: dict[str, Any]) -> bool: + return max(int(left["address"]), int(right["address"])) < min( + int(left["address"]) + int(left["size"]), + int(right["address"]) + int(right["size"]), + ) + + +def writer_seen_by_load(store: dict[str, Any], load: dict[str, Any]) -> bool: + old_provenance = provenance(store) + return any( + int(writer.get("core", -1)) == int(store["core"]) + and int(writer.get("pc", -1)) == int(store["pc"]) + and provenance(writer) == old_provenance + and ( + int(writer.get("version", -1)) in {int(version) for version in load.get("versions", [])} + or int(writer.get("cycle", -1)) == int(store.get("cycle", -2)) + ) + for writer in load.get("last_writers", []) + ) + + +def provenance_context_index( + events: list[dict[str, Any]], +) -> dict[tuple[int, int], tuple[list[int], list[tuple[int, ...]]]]: + result: dict[tuple[int, int], tuple[list[int], list[tuple[int, ...]]]] = {} + for event in events: + if event.get("event") != "local_compute" or not provenance(event): + continue + key = (int(event.get("core", -1)), int(event.get("core_iteration", -1))) + cycles, values = result.setdefault(key, ([], [])) + cycles.append(int(event.get("cycle", -1))) + values.append(provenance(event)) + return result + + +def consumer_context_provenance( + events: list[dict[str, Any]], load: dict[str, Any], + context_index: dict[tuple[int, int], tuple[list[int], list[tuple[int, ...]]]] | None = None, +) -> tuple[int, ...] | None: + if context_index is not None: + key = (int(load["core"]), int(load.get("core_iteration", -1))) + indexed = context_index.get(key) + if indexed: + cycles, values = indexed + position = bisect_right(cycles, int(load["cycle"])) - 1 + return values[position] if position >= 0 else None + context = [ + event + for event in events + if event.get("event") == "local_compute" + and int(event.get("core", -1)) == int(load["core"]) + and int(event.get("core_iteration", -1)) == int(load.get("core_iteration", -2)) + and int(event.get("cycle", -1)) < int(load["cycle"]) + and provenance(event) + ] + return provenance(context[-1]) if context else None + + +def dependency_candidates( + events: list[dict[str, Any]], static: dict[str, Any], artifact: Path +) -> tuple[list[dict[str, Any]], dict[str, int]]: + stores = [event for event in events if event.get("event") == "global_store"] + loads = [event for event in events if event.get("event") == "global_load"] + candidates: list[dict[str, Any]] = [] + graph, _ = audit.instruction_graph(artifact) + hb_edges = core_hb_edges(graph) + store_index = indexed_events(events, "global_store") + load_index = indexed_events(events, "global_load") + context_index = provenance_context_index(events) + ordered = 0 + unordered = 0 + for dependency in static["cross_core_dependencies"]: + writer = dependency["writer"] + reader = dependency["reader"] + source = (int(reader["core"]), int(reader["artifact_pc"])) + target = (int(writer["core"]), int(writer["artifact_pc"])) + hb_ordered = core_hb_reachable(hb_edges, source, target) + if hb_ordered: + ordered += 1 + else: + unordered += 1 + writer_events = store_index.get(event_key(writer), []) + reader_events = load_index.get(event_key(reader), []) + for old_store in writer_events: + old_provenance = provenance(old_store) + if not old_provenance: + continue + for load in reader_events: + if int(load["cycle"]) <= int(old_store["cycle"]): + continue + consumer_provenance = consumer_context_provenance(events, load, context_index) + if consumer_provenance != old_provenance: + continue + next_stores = [ + event + for event in writer_events + if int(event["cycle"]) > int(load["cycle"]) + and int(event.get("core_iteration", 0)) > int(old_store.get("core_iteration", 0)) + and provenance(event) + and provenance(event) != old_provenance + ] + if not next_stores: + continue + next_store = next_stores[0] + if provenance(load) == old_provenance: + if not writer_seen_by_load(old_store, load): + continue + elif provenance(load) != provenance(next_store): + continue + candidates.append( + { + "dependency": dependency, + "hb_ordered": hb_ordered, + "old_store": old_store, + "load": load, + "next_store": next_store, + "consumer_context_provenance": list(consumer_provenance), + "normal_slack": int(next_store["cycle"]) - int(load["cycle"]), + "target_spec": target_spec(dependency, old_store, load), + } + ) + break + candidates.sort(key=lambda item: (item["hb_ordered"], item["normal_slack"], event_key(item["load"]))) + unique: list[dict[str, Any]] = [] + seen: set[tuple[Any, ...]] = set() + for candidate in candidates: + old = candidate["old_store"] + load = candidate["load"] + key = ( + old["core"], old["pc"], load["core"], load["pc"], + load["address"], load["size"], + ) + if key not in seen: + seen.add(key) + unique.append(candidate) + return unique, {"hb_ordered": ordered, "hb_unordered": unordered} + + +def representative_dependency_candidate( + events: list[dict[str, Any]], static: dict[str, Any], artifact: Path +) -> dict[str, Any] | None: + graph, _ = audit.instruction_graph(artifact) + hb_edges = core_hb_edges(graph) + context_index = provenance_context_index(events) + store_index = indexed_events(events, "global_store") + load_index = indexed_events(events, "global_load") + for dependency in static["cross_core_dependencies"]: + writer = dependency["writer"] + reader = dependency["reader"] + hb_ordered = core_hb_reachable( + hb_edges, + (int(reader["core"]), int(reader["artifact_pc"])), + (int(writer["core"]), int(writer["artifact_pc"])), + ) + writer_events = store_index.get(event_key(writer), []) + for load in load_index.get(event_key(reader), []): + context = consumer_context_provenance(events, load, context_index) + old_store = next( + ( + store for store in reversed(writer_events) + if int(store["cycle"]) < int(load["cycle"]) + and context + and provenance(store) == context + and writer_seen_by_load(store, load) + ), + None, + ) + if old_store is None: + continue + return { + "dependency": dependency, + "hb_ordered": hb_ordered, + "old_store": old_store, + "load": load, + "next_store": None, + "consumer_context_provenance": list(context), + "normal_slack": None, + "target_spec": target_spec(dependency, old_store, load), + "reuse_target": False, + } + return None + + +def read_before_produce_events( + events: list[dict[str, Any]], static: dict[str, Any], artifact: Path | None = None, + report_root: Path | None = None, +) -> list[dict[str, Any]]: + result = [] + seen: set[tuple[Any, ...]] = set() + store_index = indexed_events(events, "global_store") + load_index = indexed_events(events, "global_load") + context_index = provenance_context_index(events) + for dependency in static["cross_core_dependencies"]: + writer = dependency["writer"] + reader = dependency["reader"] + writer_events = store_index.get(event_key(writer), []) + for load in load_index.get(event_key(reader), []): + if load.get("versions"): + continue + context = consumer_context_provenance(events, load, context_index) + if not context: + continue + prior_expected = any( + int(store["cycle"]) < int(load["cycle"]) + and provenance(store) == context + for store in writer_events + ) + later_writer = next( + ( + store for store in writer_events + if int(store["cycle"]) > int(load["cycle"]) + and provenance(store) == context + ), + None, + ) + if prior_expected or later_writer is None: + continue + key = (load.get("core"), load.get("pc"), load.get("cycle"), load.get("address")) + if key in seen: + continue + seen.add(key) + item = { + "load": load, + "dependency": dependency, + "later_writer": later_writer, + "expected_provenance": list(context), + } + if artifact is not None and report_root is not None: + item["smoking_gun"] = { + "reader": instruction_evidence(load, static, artifact, report_root, "reader"), + "required_writer": instruction_evidence( + later_writer, static, artifact, report_root, "writer" + ), + "relationship": "consumer loaded before the expected generation was stored", + } + result.append(item) + return result + + +def target_spec( + dependency: dict[str, Any], old_store: dict[str, Any], load: dict[str, Any] +) -> str: + overlap = dependency["overlap"] + writer_min = int(old_store.get("core_iteration", 0)) + 1 + return ":".join( + str(value) + for value in ( + old_store["core"], + old_store["pc"], + load["core"], + load["pc"], + overlap["address_begin"], + overlap["address_end"], + load.get("core_iteration", 0), + writer_min, + ) + ) + + +def target_order( + events: list[dict[str, Any]], candidate: dict[str, Any] +) -> dict[str, Any]: + old = candidate["old_store"] + baseline_load = candidate["load"] + writer_events = matching_events(events, "global_store", old) + reader_events = matching_events(events, "global_load", baseline_load) + old_prov = provenance(old) + loads = [ + event for event in reader_events + if int(event.get("core_iteration", -1)) == int(baseline_load.get("core_iteration", -2)) + ] + load = loads[0] if loads else None + expected = list(consumer_context_provenance(events, load)) if load and consumer_context_provenance(events, load) else [] + old_stores = [ + event for event in writer_events + if load + and int(event["cycle"]) < int(load["cycle"]) + and provenance(event) == tuple(expected) + ] + old_store = old_stores[-1] if old_stores else None + next_store = None + if old_store and load: + next_store = next( + ( + event for event in writer_events + if int(event["cycle"]) > int(old_store["cycle"]) + and int(event["core_iteration"]) > int(old_store["core_iteration"]) + and provenance(event) != old_prov + ), + None, + ) + observed = list(load.get("provenance", [])) if load else [] + inversion = bool(next_store and load and int(next_store["cycle"]) < int(load["cycle"])) + newer_observed = bool( + next_store + and load + and provenance(next_store) + and provenance(next_store) == tuple(observed) + and int(next_store["cycle"]) < int(load["cycle"]) + ) + future_expected = bool( + load + and expected + and any( + int(event["cycle"]) > int(load["cycle"]) + and provenance(event) == tuple(expected) + for event in writer_events + ) + ) + read_before_produce = bool(load and expected and not old_store and future_expected) + return { + "old_store": old_store, + "load": load, + "next_store": next_store, + "expected_provenance": expected, + "observed_provenance": observed, + "order_inverted": inversion, + "wrong_generation_observed": newer_observed and observed != expected, + "read_before_produce": read_before_produce, + } + + +def delay_values(slack: int) -> list[int]: + if slack <= 0: + return [1] + values = { + max(1, slack // 4), + max(1, slack // 2), + max(1, (3 * slack) // 4), + max(1, slack - 1), + slack, + slack + 1, + 2 * slack, + } + return sorted(values) + + +def sanitize_isolated_diagnostics(raw: dict[str, Any], identical: bool) -> dict[str, Any]: + result = copy.deepcopy(raw) + result["raw_host_input_provenance_mismatches"] = raw["host_input_lifetime_races"] + result["host_input_lifetime_races"] = 0 if identical else raw["host_input_lifetime_races"] + result["host_input_race_contaminates_test"] = not identical + return result + + +def run_fixed_search( + candidate: dict[str, Any], arch_out: Path, isolated_inputs: list[Path], references: list[Path], + outputs_desc: list[tuple[int, str, int, list[int]]], args: argparse.Namespace, + baseline_events: list[dict[str, Any]], target_index: int, static: dict[str, Any], + report_root: Path, channel_last: bool, +) -> dict[str, Any]: + if candidate["normal_slack"] is None: + return { + "normal_slack": None, + "tested_delays": [], + "all_delays": [], + "runs": [], + "minimum_stall_for_inversion": None, + "minimum_stall_for_provenance_failure": None, + "baseline_order": enrich_order( + target_order(baseline_events, candidate), static, + candidate["dependency"]["artifact"], report_root, + ), + "skipped": "no next-generation store observed in baseline", + } + values = delay_values(int(candidate["normal_slack"])) + runs_by_delay: dict[int, dict[str, Any]] = {} + + def run_delay(delay: int) -> dict[str, Any]: + run_out = arch_out / "fixed_stall" / f"target_{target_index:02d}" / f"delay_{delay}" + result = global_sync.run_rust( + candidate["dependency"]["artifact"], isolated_inputs, references, outputs_desc, + run_out, args, schedule_policy="greedy", schedule_seed=args.seed, + schedule_target=candidate["target_spec"], + target_stall=( + f"{candidate['load']['core']}:{candidate['load']['pc']}:{candidate['load'].get('core_iteration', 0)}:{delay}" + ), + channel_last=channel_last, + ) + trace_path = run_out / "provenance.jsonl" + events = global_sync.trace_events(trace_path) if trace_path.is_file() else [] + order = target_order(events, candidate) if events else {"order_inverted": False, "wrong_generation_observed": False, "read_before_produce": False} + order = enrich_order(order, static, candidate["dependency"]["artifact"], report_root) + item = { + "delay": delay, + "result": result, + "trace": relative(trace_path, arch_out), + "order": order, + } + runs_by_delay[delay] = item + return item + + baseline_order = enrich_order( + target_order(baseline_events, candidate), static, + candidate["dependency"]["artifact"], report_root, + ) + for delay in values: + run_delay(delay) + inverted_delays = [delay for delay, item in runs_by_delay.items() if item["order"].get("order_inverted")] + first_inversion = min(inverted_delays) if inverted_delays else None + if first_inversion is not None: + lower = max( + [delay for delay, item in runs_by_delay.items() if delay < first_inversion and not item["order"].get("order_inverted")] + or [0] + ) + upper = first_inversion + while upper - lower > 1: + middle = (lower + upper) // 2 + item = runs_by_delay.get(middle) or run_delay(middle) + if item["order"].get("order_inverted"): + upper = middle + else: + lower = middle + first_inversion = upper + runs = [runs_by_delay[delay] for delay in sorted(runs_by_delay)] + return { + "normal_slack": candidate["normal_slack"], + "tested_delays": values, + "all_delays": sorted(runs_by_delay), + "runs": runs, + "minimum_stall_for_inversion": first_inversion, + "minimum_stall_for_provenance_failure": next( + (item["delay"] for item in runs if item["order"].get("wrong_generation_observed")), None + ), + "baseline_order": baseline_order, + } + + +def run_adversarial( + candidate: dict[str, Any], arch_out: Path, isolated_inputs: list[Path], references: list[Path], + outputs_desc: list[tuple[int, str, int, list[int]]], args: argparse.Namespace, + target_index: int, static: dict[str, Any], report_root: Path, channel_last: bool, +) -> dict[str, Any]: + run_out = arch_out / "adversarial" / f"target_{target_index:02d}" + result = global_sync.run_rust( + candidate["dependency"]["artifact"], isolated_inputs, references, outputs_desc, + run_out, args, schedule_policy="adversarial", schedule_seed=args.seed, + schedule_target=candidate["target_spec"], schedule_deferral_budget=args.deferral_budget, + channel_last=channel_last, + ) + trace_path = run_out / "provenance.jsonl" + events = global_sync.trace_events(trace_path) if trace_path.is_file() else [] + order = target_order(events, candidate) if events else { + "order_inverted": False, "wrong_generation_observed": False, "read_before_produce": False, + } + order = enrich_order(order, static, candidate["dependency"]["artifact"], report_root) + scheduler_events = [ + event for event in events if event.get("event") in {"scheduler_defer", "scheduler_prefer", "scheduler_force"} + ] + return { + "result": result, + "trace": relative(trace_path, arch_out), + "order": order, + "scheduler_events": len(scheduler_events), + "deferrals": sum(event.get("event") == "scheduler_defer" for event in scheduler_events), + "force_reasons": sorted({event.get("reason") for event in scheduler_events if event.get("event") == "scheduler_force"}), + "completed": bool(events and result.get("error") is None), + } + + +def run_randomized( + candidate: dict[str, Any], arch_out: Path, isolated_inputs: list[Path], references: list[Path], + outputs_desc: list[tuple[int, str, int, list[int]]], args: argparse.Namespace, + static: dict[str, Any], report_root: Path, channel_last: bool, +) -> list[dict[str, Any]]: + results = [] + for seed in range(args.randomized_seeds): + run_out = arch_out / "randomized" / f"seed_{seed:04d}" + result = global_sync.run_rust( + candidate["dependency"]["artifact"], isolated_inputs, references, outputs_desc, + run_out, args, schedule_policy="randomized", schedule_seed=seed, + schedule_target=candidate["target_spec"], + channel_last=channel_last, + ) + trace_path = run_out / "provenance.jsonl" + events = global_sync.trace_events(trace_path) if trace_path.is_file() else [] + order = target_order(events, candidate) if events else { + "wrong_generation_observed": False, + "order_inverted": False, + "read_before_produce": False, + } + order = enrich_order(order, static, candidate["dependency"]["artifact"], report_root) + diagnostics = ( + sanitize_isolated_diagnostics( + global_sync.dynamic_analysis( + events, static, candidate["dependency"]["artifact"] / "config.json" + ), + True, + ) + if events + else {} + ) + results.append({ + "seed": seed, + "result": result, + "trace": relative(trace_path, arch_out), + "order": order, + "diagnostics": diagnostics, + "read_before_produce": read_before_produce_events( + events, static, candidate["dependency"]["artifact"], report_root + ) if events else [], + }) + return results + + +def architecture_classification( + validation: dict[str, Any], candidates: list[dict[str, Any]], + fixed: list[dict[str, Any]], adversarial: list[dict[str, Any]], + randomized: list[dict[str, Any]], hb_counts: dict[str, int], +) -> str: + if not validation["valid"]: + return INVALID + read_before = ( + any(item["order"].get("read_before_produce") for item in adversarial) + or any( + item["order"].get("read_before_produce") + for search in fixed for item in search["runs"] + ) + or any(item["read_before_produce"] for item in randomized) + ) + generation_race = ( + any(item["order"].get("wrong_generation_observed") for item in adversarial) + or any( + item["order"].get("wrong_generation_observed") + for search in fixed for item in search["runs"] + ) + or any(item["order"].get("wrong_generation_observed") for item in randomized) + ) + if generation_race: + return GENERATION_RACE + if read_before: + return READ_BEFORE_PRODUCE + unordered = [candidate for candidate in candidates if not candidate["hb_ordered"]] + if ( + any(item["order"].get("order_inverted") for item in adversarial) + or any(item["order"].get("order_inverted") for search in fixed for item in search["runs"]) + or any(item["order"].get("order_inverted") for item in randomized) + ): + return UNORDERED_COUNTEREXAMPLE + if unordered or hb_counts["hb_unordered"]: + return UNORDERED_NO_COUNTEREXAMPLE + return PROVEN_ORDERED + + +def aggregate_classifications(classifications: list[str]) -> str: + if INVALID in classifications: + return INVALID + if RAPTOR_UNAVAILABLE in classifications: + return RAPTOR_UNAVAILABLE + for classification in ( + GENERATION_RACE, + READ_BEFORE_PRODUCE, + UNORDERED_COUNTEREXAMPLE, + UNORDERED_NO_COUNTEREXAMPLE, + ): + if classification in classifications: + return classification + return PROVEN_ORDERED + + +def architecture_interpretation(architecture: str, classification: str) -> dict[str, str]: + if architecture == "arch-a": + return { + "pimcomp_model": "PIMCOMP_ARCH_A_MODEL_PERMITS_DELAY", + "original_hardware": "ADVERSARIAL_DELAY_NOT_PROVEN_LEGAL_FOR_ORIGINAL_ARCH_A", + "note": "ISAAC static timing/documentary mapping remains unproven for generic PIMCOMP global memory.", + } + if architecture == "arch-b": + return { + "pimcomp_model": "PIMCOMP_ARCH_B_ABSTRACTION_NOT_GENERATION_SAFE" if classification in {GENERATION_RACE, READ_BEFORE_PRODUCE, UNORDERED_COUNTEREXAMPLE} else "PIMCOMP_ARCH_B_ABSTRACTION_NO_COUNTEREXAMPLE", + "original_hardware": "PUMA_VALID_COUNT_NOT_AUTOMATICALLY_INHERITED", + "note": "A PUMA valid/count buffer is documentary hardware evidence; PIMCOMP ordinary LD/ST does not encode it.", + } + return { + "pimcomp_model": "PIMCOMP_ARCH_C_ABSTRACT_MODEL_NOT_GENERATION_SAFE" if classification in {GENERATION_RACE, READ_BEFORE_PRODUCE, UNORDERED_COUNTEREXAMPLE} else "PIMCOMP_ARCH_C_ABSTRACT_MODEL_NO_COUNTEREXAMPLE", + "original_hardware": "MAPPING_NOT_ESTABLISHED", + "note": "The Arch-C row scales the cited 4MB ReRAM processor to a larger simulated mesh; ordering equivalence is unresolved.", + } + + +def unavailable_raptor_artifact( + architecture: str, + manifest: dict[str, Any], + throughput_config: Path, + latency_config: Path, + report_root: Path, + comparison_report: Path, + error: str | None, +) -> dict[str, Any]: + return { + "source": "raptor", + "identity": manifest["architectures"][architecture]["pimcomp_identity"], + "classification": RAPTOR_UNAVAILABLE, + "architecture_interpretation": { + "pimcomp_model": "RAPTOR_ARTIFACT_NOT_AVAILABLE", + "original_hardware": "NOT_TESTED", + "note": "Raptor did not emit an executable artifact for this configuration; no synchronization claim is made.", + }, + "config": { + "throughput": relative(throughput_config, REPO), + "latency": relative(latency_config, REPO), + "throughput_sha256": sha256(throughput_config), + "latency_sha256": sha256(latency_config), + }, + "artifact": { + "source": "raptor", + "rust_export": None, + "pimsim_export": None, + "format": None, + "hashes": {}, + "instruction_files": {}, + }, + "validation": { + "valid": False, + "artifact_available": False, + "batch1": {"passed": False, "error": error}, + "isolated_throughput": {"passed": False, "error": error}, + "active_core_count": 0, + "cross_core_dependency_count": 0, + "executed_cross_core_dependencies": 0, + "memory_ranges_reused": 0, + "sample_dependent_cross_core_links": 0, + "host_input_race_contaminates_test": False, + }, + "happens_before": {"hb_ordered": 0, "hb_unordered": 0}, + "static_instruction_counts": {}, + "diagnostics": { + "host_input_lifetime_races": 0, + "raw_host_input_provenance_mismatches": 0, + "send_recv_generation_races": 0, + "mixed_sample_operations": 0, + "global_memory_versions_reused": 0, + "generation_races": 0, + "read_before_produce": 0, + }, + "smoking_guns": { + "availability": { + "comparison_report": relative(comparison_report, report_root), + "error": error, + }, + "static_dependency": None, + "first_generation_race": None, + "first_read_before_produce": None, + }, + "targets": [], + "source_identity": audit.architecture_source_evidence(architecture, manifest), + } + + +def run_artifact_experiment( + architecture: str, + source: str, + artifact: Path, + pimsim_artifact: Path | None, + source_out: Path, + inputs: list[Path], + isolated_inputs: list[Path], + references: list[Path], + outputs_desc: list[tuple[int, str, int, list[int]]], + args: argparse.Namespace, + manifest: dict[str, Any], + throughput_config: Path, + latency_config: Path, + report_root: Path, +) -> dict[str, Any]: + """Run the same evidence-producing experiment on one executable artifact.""" + source_out.mkdir(parents=True, exist_ok=True) + channel_last = source == "pimcomp" + static = global_sync.analyze_artifact(artifact) + static["artifact"] = artifact + static["artifact_hashes"] = { + "config": sha256(artifact / "config.json"), + "core_instruction_files": audit.tree_hash(artifact, "core_*.json"), + "core_binary_files": audit.tree_hash(artifact, "core_*.pim"), + } + static_output = copy.deepcopy(static) + static_output.pop("artifact", None) + write_json(source_out / "analysis/global_memory_dependencies.json", static_output) + + batch1 = global_sync.run_rust( + artifact, inputs[:1], references[:1], outputs_desc, + source_out / "rust/batch1", args, channel_last=channel_last, + ) + distinct = global_sync.run_rust( + artifact, inputs[:args.batch_size], references[:args.batch_size], outputs_desc, + source_out / "rust/distinct_throughput", args, channel_last=channel_last, + ) + isolated_references = [references[0]] * args.batch_size + isolated = global_sync.run_rust( + artifact, isolated_inputs, isolated_references, outputs_desc, + source_out / "rust/input_isolated", args, channel_last=channel_last, + ) + isolated_trace = source_out / "rust/input_isolated/provenance.jsonl" + isolated_events = global_sync.trace_events(isolated_trace) + raw_dynamic = global_sync.dynamic_analysis(isolated_events, static, artifact / "config.json") + dynamic = sanitize_isolated_diagnostics(raw_dynamic, True) + static_candidates, hb_counts = dependency_candidates(isolated_events, static, artifact) + candidates = [candidate for candidate in static_candidates if not candidate["hb_ordered"]][: args.max_targets] + if not candidates: + fallback = representative_dependency_candidate(isolated_events, static, artifact) + if fallback is not None and not fallback["hb_ordered"]: + candidates = [fallback] + for candidate in candidates: + candidate["dependency"]["artifact"] = artifact + candidate["static_smoking_gun"] = static_dependency_evidence( + candidate["dependency"], static, artifact, report_root, candidate["hb_ordered"] + ) + + validation = { + "batch1": batch1, + "distinct_throughput": distinct, + "isolated_throughput": isolated, + "input_layout": "flattened PIMCOMP" if source == "pimcomp" else "NCHW Raptor", + "active_core_count": static["participating_core_count"], + "cross_core_dependency_count": static["cross_core_dependency_count"], + "executed_cross_core_dependencies": dynamic["executed_cross_core_dependencies"], + "memory_ranges_reused": dynamic["global_memory_versions_reused"], + "sample_dependent_cross_core_links": dynamic["sample_dependent_cross_core_links"], + "host_input_race_contaminates_test": dynamic["host_input_race_contaminates_test"], + } + validation["valid"] = bool( + batch1.get("passed") + and isolated.get("passed") + and validation["active_core_count"] >= 2 + and validation["cross_core_dependency_count"] > 0 + and validation["executed_cross_core_dependencies"] > 0 + and validation["memory_ranges_reused"] > 0 + and validation["sample_dependent_cross_core_links"] > 0 + and not validation["host_input_race_contaminates_test"] + ) + + fixed_results = [] + adversarial_results = [] + randomized_results: list[dict[str, Any]] = [] + for index, candidate in enumerate(candidates): + fixed_results.append( + run_fixed_search( + candidate, source_out, isolated_inputs, isolated_references, + outputs_desc, args, isolated_events, index, static, report_root, channel_last, + ) + ) + adversarial_results.append( + run_adversarial( + candidate, source_out, isolated_inputs, isolated_references, + outputs_desc, args, index, static, report_root, channel_last, + ) + ) + if index == 0 and args.randomized_seeds: + randomized_results = run_randomized( + candidate, source_out, isolated_inputs, isolated_references, + outputs_desc, args, static, report_root, channel_last, + ) + classification = architecture_classification( + validation, candidates, fixed_results, adversarial_results, + randomized_results, hb_counts, + ) + first_proof = next( + ( + target["order"]["smoking_guns"] + for target in adversarial_results + if target["order"].get("wrong_generation_observed") + ), + None, + ) + first_read_before = next( + ( + event["smoking_gun"] + for result in randomized_results + for event in result["read_before_produce"] + if event.get("smoking_gun") + ), + None, + ) + artifact_info = { + "source": source, + "rust_export": relative(artifact, report_root), + "pimsim_export": relative(pimsim_artifact, report_root) if pimsim_artifact else None, + "format": static["artifact_format"], + "hashes": static["artifact_hashes"], + "instruction_files": static["instruction_files"], + } + return { + "source": source, + "identity": manifest["architectures"][architecture]["pimcomp_identity"], + "classification": classification, + "architecture_interpretation": architecture_interpretation(architecture, classification), + "config": { + "throughput": relative(throughput_config, REPO), + "latency": relative(latency_config, REPO), + "throughput_sha256": sha256(throughput_config), + "latency_sha256": sha256(latency_config), + }, + "artifact": artifact_info, + "validation": validation, + "happens_before": hb_counts, + "static_instruction_counts": static["instruction_counts"], + "diagnostics": { + "host_input_lifetime_races": dynamic["host_input_lifetime_races"], + "raw_host_input_provenance_mismatches": dynamic["raw_host_input_provenance_mismatches"], + "send_recv_generation_races": dynamic["send_recv_generation_races"], + "mixed_sample_operations": dynamic["mixed_sample_operations"], + "global_memory_versions_reused": dynamic["global_memory_versions_reused"], + "fixed_generation_races": sum( + item["order"].get("wrong_generation_observed", False) + for search in fixed_results for item in search["runs"] + ), + "adversarial_generation_races": sum( + item["order"].get("wrong_generation_observed", False) + for item in adversarial_results + ), + "randomized_generation_races": sum( + item["order"].get("wrong_generation_observed", False) + for item in randomized_results + ), + "randomized_read_before_produce": sum( + len(item["read_before_produce"]) for item in randomized_results + ), + "generation_races": sum( + item["order"].get("wrong_generation_observed", False) + for item in adversarial_results + ) + sum( + item["order"].get("wrong_generation_observed", False) + for search in fixed_results for item in search["runs"] + ) + sum( + item["order"].get("wrong_generation_observed", False) + for item in randomized_results + ), + "read_before_produce": sum( + item["order"].get("read_before_produce", False) + for item in adversarial_results + ) + sum( + item["order"].get("read_before_produce", False) + for search in fixed_results for item in search["runs"] + ) + sum( + len(item["read_before_produce"]) for item in randomized_results + ), + }, + "smoking_guns": { + "static_dependency": ( + static_dependency_evidence( + static["representative_dependency"], static, artifact, report_root + ) if static["representative_dependency"] else None + ), + "first_generation_race": first_proof, + "first_read_before_produce": first_read_before, + }, + "targets": [ + { + "candidate": { + key: value for key, value in candidate.items() if key != "dependency" + } | { + "dependency": { + key: value for key, value in candidate["dependency"].items() if key != "artifact" + } + }, + "fixed": fixed_results[index], + "adversarial": adversarial_results[index], + "randomized": randomized_results if index == 0 else [], + } + for index, candidate in enumerate(candidates) + ], + "source_identity": audit.architecture_source_evidence(architecture, manifest), + } + + +def markdown(report: dict[str, Any]) -> str: + def evidence_text(evidence: dict[str, Any] | None) -> str: + if evidence is None: + return "not observed" + provenance_text = evidence.get("provenance", []) + versions = evidence.get("versions", []) + version_text = evidence.get("version") + if versions: + version_text = versions if len(versions) <= 8 else versions[:3] + ["...", *versions[-3:]] + return ( + f"core {evidence['core']} (core_{evidence['core_file_index']}.json / " + f"{evidence.get('executed_instruction_file')}) " + f"PC {evidence['pc']} iter {evidence.get('core_iteration')} " + f"cycle {evidence.get('cycle')} range " + f"[{evidence['address']},{evidence['address'] + evidence['size']}) " + f"provenance {provenance_text} version {version_text}\n" + f" instruction: {evidence.get('instruction_file')}\n" + f" executed: {evidence.get('executed_instruction_file')}" + ) + + lines = [ + "PIMCOMP/Raptor Adversarial Global-Memory Synchronization Test", + "===============================================================", + "", + "Scheduler:", + " default: greedy / ASAP", + " diagnostics: bounded target delay, randomized, adversarial", + f" seed: {report['seed']}", + f" batch: {report['batch_size']}", + "", + ] + for architecture, architecture_item in report["architectures"].items(): + lines += [ + architecture, + "-" * len(architecture), + f"Contract classification: {architecture_item['contract_classification']}", + "", + ] + for source, item in architecture_item["artifacts"].items(): + validation = item["validation"] + lines += [ + f"{source.upper()} ARTIFACT", + f" artifact: {item['artifact']['rust_export']}", + f" format: {item['artifact']['format']}", + f" classification: {item['classification']}", + f" single inference: {'PASS' if validation['batch1']['passed'] else 'FAIL'}", + f" input-isolated greedy throughput: {'PASS' if validation['isolated_throughput']['passed'] else 'FAIL'}", + f" cores: {validation['active_core_count']}; cross-core ST→LD dependencies: {validation['cross_core_dependency_count']}", + f" executed dependencies: {validation['executed_cross_core_dependencies']}; reused ranges: {validation['memory_ranges_reused']}", + f" HB ordered/unordered: {item['happens_before']['hb_ordered']}/{item['happens_before']['hb_unordered']}", + f" host-input contamination: {validation['host_input_race_contaminates_test']}", + f" adversarial targets: {len(item['targets'])}; proven generation races: {item['diagnostics']['generation_races']}", + "", + ] + static = item["smoking_guns"].get("static_dependency") + if static: + lines += [ + " STATIC DEPENDENCY:", + f" range [{static['range']['address_begin']},{static['range']['address_end']})", + f" writer: {evidence_text(static['writer'])}", + f" reader: {evidence_text(static['reader'])}", + f" HB LD_N→ST_N+1: {'YES' if static['hb_ordered'] else 'NO'}", + "", + ] + for index, target in enumerate(item["targets"]): + candidate = target["candidate"] + order = target["adversarial"]["order"] + next_store = candidate.get("next_store") + next_store_text = next_store["cycle"] if next_store else "not observed" + lines += [ + f" TARGET {index}: range [{candidate['dependency']['overlap']['address_begin']},{candidate['dependency']['overlap']['address_end']})", + f" writer core {candidate['old_store']['core']} PC {candidate['old_store']['pc']} iter {candidate['old_store']['core_iteration']}", + f" reader core {candidate['load']['core']} PC {candidate['load']['pc']} iter {candidate['load']['core_iteration']}", + f" normal LD cycle {candidate['load']['cycle']}; next ST cycle {next_store_text}; slack {candidate['normal_slack']}", + f" HB LD_N→ST_N+1: {'YES' if candidate['hb_ordered'] else 'NO'}", + f" fixed minimum inversion delay: {target['fixed']['minimum_stall_for_inversion']}", + f" adversarial order inverted: {order.get('order_inverted', False)}", + f" expected provenance: {order.get('expected_provenance', [])}; observed: {order.get('observed_provenance', [])}", + "", + ] + guns = order.get("smoking_guns", {}) + if order.get("wrong_generation_observed"): + lines += [ + " CONFIRMED SMOKING GUN:", + f" old generation ST: {evidence_text(guns.get('old_generation_store'))}", + f" overwrite ST: {evidence_text(guns.get('overwrite_store'))}", + f" consumer LD: {evidence_text(guns.get('consumer_load'))}", + "", + ] + elif guns.get("consumer_load"): + lines += [ + " TRACE EVIDENCE (no wrong generation observed):", + f" old generation ST: {evidence_text(guns.get('old_generation_store'))}", + f" next generation ST: {evidence_text(guns.get('overwrite_store'))}", + f" consumer LD: {evidence_text(guns['consumer_load'])}", + "", + ] + read_before = item["smoking_guns"].get("first_read_before_produce") + if read_before: + lines += [ + " READ-BEFORE-PRODUCE SMOKING GUN:", + f" reader: {evidence_text(read_before.get('reader'))}", + f" required writer: {evidence_text(read_before.get('required_writer'))}", + "", + ] + lines += [" CONCLUSION:", f" {item['classification']}", ""] + lines += [ + "Interpretation:", + " A structural unordered relation without a dynamic counterexample is not called safe.", + " Each smoking gun names the generated JSON instruction and the executed .pim file when Raptor uses binary instructions.", + "", + f"PIMCOMP CLASSIFICATION: {report.get('pimcomp_classification', report['classification'])}", + f"RAPTOR CLASSIFICATION: {report.get('raptor_classification', 'not run')}", + f"FINAL CLASSIFICATION: {report['classification']}", + "", + f"PIMCOMP compiler semantics modified by this run: {report['pimcomp_semantics_modified']}", + ] + return "\n".join(lines) + "\n" + + +def self_check(report: dict[str, Any]) -> None: + if report["classification"] == INVALID: + raise ExperimentError("self-check rejected the reproducer") + for architecture, architecture_item in report["architectures"].items(): + for source, item in architecture_item["artifacts"].items(): + validation = item["validation"] + if item["classification"] == RAPTOR_UNAVAILABLE: + availability = item["smoking_guns"].get("availability", {}) + if not availability.get("comparison_report") or not availability.get("error"): + raise ExperimentError(f"{architecture}/{source}: missing-artifact evidence is incomplete") + continue + if not validation["valid"]: + raise ExperimentError(f"{architecture}/{source}: artifact validation failed") + for target in item["targets"]: + if not target["adversarial"]["completed"]: + raise ExperimentError(f"{architecture}/{source}: adversarial target did not complete") + order = target["adversarial"]["order"] + if order.get("wrong_generation_observed") and not order.get("order_inverted"): + raise ExperimentError(f"{architecture}/{source}: provenance mismatch lacks overwrite order proof") + if item["classification"] == READ_BEFORE_PRODUCE and not item["diagnostics"]["read_before_produce"]: + raise ExperimentError(f"{architecture}/{source}: read-before-produce lacks evidence") + if item["classification"] == GENERATION_RACE and not item["diagnostics"]["generation_races"]: + raise ExperimentError(f"{architecture}/{source}: generation-race lacks provenance proof") + if item["classification"] == PROVEN_ORDERED and item["happens_before"]["hb_unordered"]: + raise ExperimentError(f"{architecture}/{source}: ordered classification has unordered dependencies") + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--out-dir", type=Path, required=True) + parser.add_argument("--batch-size", type=int, default=4) + parser.add_argument("--seed", type=int, default=0) + parser.add_argument("--max-targets", type=int, default=2) + parser.add_argument("--randomized-seeds", type=int, default=2) + parser.add_argument("--deferral-budget", type=int, default=30000) + parser.add_argument("--threshold", type=float, default=1e-3) + parser.add_argument("--rtol", type=float, default=1e-4) + parser.add_argument("--timeout", type=float, default=1800.0) + parser.add_argument("--no-fast", action="store_true") + parser.add_argument("--self-check", action="store_true") + args = parser.parse_args() + if args.batch_size < 2: + parser.error("--batch-size must be at least 2") + if args.max_targets < 1 or args.randomized_seeds < 0 or args.deferral_budget < 1: + parser.error("target count, randomized seeds, and deferral budget are invalid") + args.out_dir = args.out_dir.resolve() + return args + + +def main() -> int: + args = parse_args() + out = args.out_dir + if out.exists() and any(out.iterdir()): + print(f"output directory must be new or empty: {out}", file=sys.stderr) + return 1 + out.mkdir(parents=True, exist_ok=True) + pimcomp_before = audit.git_identity(audit.PIMCOMP_ROOT) + manifest = audit.load_evidence() + source_contract = audit.source_contract() + report: dict[str, Any] = { + "schema": 1, + "experiment": "PIMCOMP adversarial global-memory synchronization", + "seed": args.seed, + "batch_size": args.batch_size, + "scheduler": { + "default_policy": "greedy", + "diagnostic_policies": ["bounded_target_stall", "randomized", "adversarial"], + "deferral_budget": args.deferral_budget, + }, + "source_identity": { + "repository": audit.git_identity(REPO), + "pimcomp": pimcomp_before, + "pimsim_nn": audit.git_identity(audit.PIMSIM_ROOT), + "rust_pim_simulator": audit.git_identity(audit.RUST_ROOT), + "script": {"path": relative(SCRIPT, REPO), "sha256": sha256(SCRIPT)}, + "architecture_audit_script": {"path": relative(AUDIT_SCRIPT, REPO), "sha256": sha256(AUDIT_SCRIPT)}, + }, + "architectures": {}, + "classification": INVALID, + "pimcomp_semantics_modified": False, + "errors": [], + } + try: + if not PYTHON.is_file(): + raise ExperimentError(f"Python virtual environment missing: {PYTHON}") + for architecture in ("arch-a", "arch-b", "arch-c"): + global_sync.check_prerequisites(*global_sync.architecture_configs(architecture)) + global_sync.build_simulator(args, out / "build") + model = out / "model.onnx" + global_sync.make_model(model) + report["model"] = { + "path": relative(model, out), + "sha256": sha256(model), + "description": "two connected identity padded 3x3 Conv stages", + } + run_count = max(8, args.batch_size) + input_batch, raptor_inputs, pimcomp_inputs, _ = global_sync.make_inputs(model, run_count, args.seed, out) + isolated_inputs = audit.make_identical_inputs(model, args.batch_size, out) + isolated_raptor_inputs = [ + Path(path) for path in global_sync.compare.write_input_batch_binaries( + [input_batch[0] for _ in range(args.batch_size)], out / "inputs/raptor_isolated" + ) + ] + isolated_hashes = {sha256(path) for path in isolated_inputs} + isolated_raptor_hashes = {sha256(path) for path in isolated_raptor_inputs} + if len(isolated_hashes) != 1: + raise ExperimentError("input-isolated files are not byte-identical") + if len(isolated_raptor_hashes) != 1: + raise ExperimentError("Raptor input-isolated files are not byte-identical") + _, outputs_desc = global_sync.compare.onnx_io(model) + references: list[Path] | None = None + for architecture in ("arch-a", "arch-b", "arch-c"): + architecture_out = out / architecture + throughput_config, latency_config = global_sync.architecture_configs(architecture) + compiled = global_sync.compile_artifact(args, model, architecture_out, throughput_config) + if references is None: + references = global_sync.make_references(model, input_batch, architecture_out, args) + contract = audit.classify_contract(architecture, source_contract, manifest) + raptor_available = ( + compiled["raptor_artifact"].is_dir() + and (compiled["raptor_artifact"] / "config.json").is_file() + and any(compiled["raptor_artifact"].glob("core_*.json")) + ) + raptor_result = ( + run_artifact_experiment( + architecture, "raptor", compiled["raptor_artifact"], None, + architecture_out / "raptor", raptor_inputs, isolated_raptor_inputs, references, + outputs_desc, args, manifest, throughput_config, + latency_config, out, + ) + if raptor_available + else unavailable_raptor_artifact( + architecture, manifest, throughput_config, latency_config, out, + architecture_out / "comparison/pimcomp/comparison_report.json", + compiled.get("raptor_error"), + ) + ) + artifacts = { + "pimcomp": run_artifact_experiment( + architecture, "pimcomp", compiled["artifact"], compiled["pimsim_artifact"], + architecture_out / "pimcomp", pimcomp_inputs, isolated_inputs, references, + outputs_desc, args, manifest, throughput_config, + latency_config, out, + ), + "raptor": raptor_result, + } + pimcomp_item = artifacts["pimcomp"] + item = { + "identity": manifest["architectures"][architecture]["pimcomp_identity"], + "contract_classification": contract, + # Keep the PIMCOMP fields at the architecture level for old consumers; + # the new nested records are the authoritative per-artifact evidence. + **pimcomp_item, + "contract_classification": contract, + "artifacts": artifacts, + "artifact_classifications": { + "pimcomp": artifacts["pimcomp"]["classification"], + "raptor": artifacts["raptor"]["classification"], + }, + "raptor": artifacts["raptor"], + } + report["architectures"][architecture] = item + write_json(architecture_out / "adversarial_memory_sync.json", item) + write_json(architecture_out / "pimcomp/adversarial_memory_sync.json", artifacts["pimcomp"]) + write_json(architecture_out / "raptor/adversarial_memory_sync.json", artifacts["raptor"]) + pimcomp_classes = [item["artifacts"]["pimcomp"]["classification"] for item in report["architectures"].values()] + raptor_classes = [item["artifacts"]["raptor"]["classification"] for item in report["architectures"].values()] + report["pimcomp_classification"] = aggregate_classifications(pimcomp_classes) + report["raptor_classification"] = aggregate_classifications(raptor_classes) + runnable_classes = [ + classification for classification in pimcomp_classes + raptor_classes + if classification != RAPTOR_UNAVAILABLE + ] + report["classification"] = aggregate_classifications(runnable_classes or raptor_classes) + pimcomp_after = audit.git_identity(audit.PIMCOMP_ROOT) + report["source_identity"]["pimcomp_after"] = pimcomp_after + report["pimcomp_semantics_modified"] = pimcomp_before["worktree_status"] != pimcomp_after["worktree_status"] + if args.self_check: + self_check(report) + except Exception as exc: + report["errors"].append(f"{type(exc).__name__}: {exc}") + report["classification"] = INVALID + write_json(out / "adversarial_memory_sync_report.json", report) + (out / "adversarial_memory_sync_report.md").write_text(markdown(report), encoding="utf-8") + print("=" * 60) + for architecture, item in report["architectures"].items(): + print(f"{architecture}/PIMCOMP: {item['artifacts']['pimcomp']['classification']}") + print(f"{architecture}/Raptor: {item['artifacts']['raptor']['classification']}") + print("=" * 60) + print(f"FINAL CLASSIFICATION: {report['classification']}") + print(f"JSON report: {out / 'adversarial_memory_sync_report.json'}") + print(f"Markdown report: {out / 'adversarial_memory_sync_report.md'}") + if report["errors"]: + print("Errors:", file=sys.stderr) + for error in report["errors"]: + print(f" {error}", file=sys.stderr) + return 0 if report["classification"] != INVALID else 1 + + +if __name__ == "__main__": + raise SystemExit(main())