pim simulator adversary mode

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<PathBuf>,
+
+    /// Optional JSONL shadow provenance trace
+    #[arg(long)]
+    provenance_trace: Option<PathBuf>,
+
+    /// 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<String>,
+
+    /// 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<String>,
+
+    /// 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<String>,
 }
 
 #[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<DiagnosticScheduleTarget> {
+    let values: Vec<usize> = spec
+        .split(':')
+        .map(|value| {
+            value
+                .parse()
+                .with_context(|| format!("invalid schedule target field: {value}"))
+        })
+        .collect::<Result<_>>()?;
+    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<u32>, 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<u32> {
     match (&args.mode, args.batch_size) {
         (ExecutionMode::Latency, None | Some(1)) => Ok(1),

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<BatchOutputs>,
+    provenance: Option<ProvenanceTracker>,
 }
 
 #[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::path::Path>,
+    ) -> 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<usize>,
+        target: Option<DiagnosticScheduleTarget>,
+        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::<u8>(core_address, size)?[0];

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::<M>(crossbar_stored_bytes)?[0];
+    let used_rows: Vec<bool> = 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::<F>(),
+        res_up.len() * size_of::<T>(),
+        &used_rows,
+        "mvmul",
+    );
 
     TRACER.lock().unwrap().post_mvm::<F, M, T>(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::<F, T>(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::<F>(),
+        stride,
+        element_count,
+        "vmv",
+    );
     Ok(InstructionStatus::Completed)
 }
 
@@ -799,16 +837,23 @@ pub fn vrsl(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus>
 #[inline(never)]
 pub fn ld(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus> {
     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::<u8>(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::<u8>(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<InstructionStatus> {
         (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<InstructionStatus>
 #[inline(never)]
 pub fn lmv(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus> {
     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<InstructionStatus>
     let local_memory = core.load::<u8>(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)
 }

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<Instructions> 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<u32>,
+    pub writer_min_iteration: Option<u32>,
+}
+
+#[derive(Clone, Copy, Debug)]
+struct DiagnosticScheduleConfig {
+    policy: DiagnosticSchedulePolicy,
+    seed: u64,
+    target: Option<DiagnosticScheduleTarget>,
+    deferral_budget: u64,
+    fixed_stall: Option<(usize, u64)>,
+    fixed_target_stall: Option<(usize, usize, Option<u32>, 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<u32>,
+    last_writer_state: Option<ScheduleCoreState>,
+    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<ScheduleCoreState> {
+        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<usize> {
+        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<ScheduleChoice> {
+        (!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<ScheduleChoice> {
+        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<CoreInstructions>,
     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<Path>) -> 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<u32>,
+        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<usize> = 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<ScheduleCoreState> {
+        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"
+            }
+        );
+    }
+}

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<Item = Self>) -> Self {
+        let mut values = values.into_iter();
+        values
+            .next()
+            .map_or(Self::Uninitialized, |first| values.fold(first, Self::merge))
+    }
+
+    fn samples(self) -> Vec<u32> {
+        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<Writer>,
+}
+
+#[derive(Debug)]
+struct TraceSink {
+    output: BufWriter<File>,
+}
+
+impl TraceSink {
+    fn new(path: impl AsRef<Path>) -> std::io::Result<Self> {
+        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<GlobalCell>,
+    local: Vec<Vec<Provenance>>,
+    next_version: u64,
+    context: ExecutionContext,
+    sink: Arc<Mutex<TraceSink>>,
+}
+
+impl ProvenanceTracker {
+    pub fn new(core_count: usize, path: impl AsRef<Path>) -> std::io::Result<Self> {
+        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<Value>,
+        deferral_budget: u64,
+        fixed_stall: Option<(usize, u64)>,
+        fixed_target_stall: Option<(usize, usize, Option<u32>, 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<Provenance> {
+        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<u64> {
+        cells
+            .iter()
+            .filter_map(|cell| cell.writer.map(|writer| writer.version))
+            .collect::<BTreeSet<_>>()
+            .into_iter()
+            .collect()
+    }
+
+    fn unique_writers(cells: &[GlobalCell]) -> Vec<Writer> {
+        cells
+            .iter()
+            .filter_map(|cell| cell.writer)
+            .collect::<BTreeSet<_>>()
+            .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::<Vec<_>>(),
+        });
+        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::<Vec<_>>(),
+        });
+        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<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::<Vec<_>>(),
+        });
+        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::<Vec<_>>(),
+                "provenance": provenance.samples(),
+            }));
+        }
+    }
+}

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::<u8>(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::<u8>(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;