From 4e7fe721f8b0b2a151d0907aecc384c79a5eddf0 Mon Sep 17 00:00:00 2001 From: ilgeco Date: Fri, 21 Aug 2026 15:22:16 +0200 Subject: [PATCH] pim simulator adversary mode --- .../src/bin/pim-simulator/main.rs | 136 ++++ .../pim/pim-simulator/src/lib/cpu/mod.rs | 261 +++++++ .../src/lib/instruction_set/isa.rs | 70 +- .../pim/pim-simulator/src/lib/pimcore.rs | 729 +++++++++++++++++- .../pim/pim-simulator/src/lib/provenance.rs | 639 +++++++++++++++ .../pim/pim-simulator/src/lib/send_recv.rs | 33 +- 6 files changed, 1812 insertions(+), 56 deletions(-) create mode 100644 backend-simulators/pim/pim-simulator/src/lib/provenance.rs 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 6e4b2c5..5358af1 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 5bcae69..e95a7a0 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 aa62c20..17da145 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 { @@ -134,9 +441,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, @@ -169,28 +519,113 @@ 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 = vec![[0; 32]; max_core]; 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() @@ -198,48 +633,129 @@ 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) { - bail!( - "Deadlock cycle 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) { + bail!( + "Deadlock cycle 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; @@ -271,6 +787,7 @@ impl<'a> Executable<'a> { #[cfg(feature = "profile_time")] TRACER.lock().unwrap().report(); + cpu.finish_provenance(); Ok(cpu.finish_host_store_recording()) } @@ -306,11 +823,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(()) @@ -465,3 +988,143 @@ fn handle_wait_sync( _ => false, } } + +#[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" + } + ); + } +} 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;