diff --git a/transactional_emulator/lib/memory/src/chunked.rs b/transactional_emulator/lib/memory/src/chunked.rs index 65959552..0d14a6c1 100644 --- a/transactional_emulator/lib/memory/src/chunked.rs +++ b/transactional_emulator/lib/memory/src/chunked.rs @@ -11,8 +11,7 @@ use std::sync::Arc; -use futures_util::StreamExt; -use futures_util::stream::FuturesUnordered; +use futures_util::future::join_all; use crate::{ErasedMemoryModel, MemoryModel}; @@ -31,21 +30,21 @@ pub struct ChunkRead { /// Issue every [`ChunkRead`] concurrently against `hbm` and assemble the /// results into a single `total_len`-byte buffer. /// -/// All reads race in one pool, preserving the simulator's concurrent-access -/// timing; the buffer is filled as each read completes (completion order does -/// not matter — each result carries its own destination offset). +/// Reads are first issued in the input order, then allowed to complete +/// concurrently. Completion order does not matter because each result carries +/// its own destination offset, but deterministic issue order is part of the +/// simulator timing contract. pub async fn gather( hbm: &Arc, total_len: usize, reads: Vec, ) -> Vec { let mut out = vec![0u8; total_len]; - let mut futures = FuturesUnordered::new(); - for r in reads { + let futures = reads.into_iter().map(|r| { // A single read cannot span more than the 64-byte block it lands in. debug_assert!(r.len <= 64, "ChunkRead::len {} exceeds 64", r.len); let hbm = hbm.clone(); - futures.push(async move { + async move { let aligned = (r.addr / 64) * 64; let within = (r.addr % 64) as usize; let block = hbm.read(aligned).await; @@ -54,9 +53,9 @@ pub async fn gather( let mut buf = [0u8; 64]; buf[..n].copy_from_slice(&block[within..end]); (r.dst_offset, buf, n) - }); - } - while let Some((offset, data, n)) = futures.next().await { + } + }); + for (offset, data, n) in join_all(futures).await { out[offset..offset + n].copy_from_slice(&data[..n]); } out @@ -127,7 +126,7 @@ mod tests { use super::*; use crate::MemoryBacked; use proptest::prelude::*; - use std::sync::Arc; + use std::sync::{Arc, Mutex}; /// A `MemoryBacked` HBM seeded by `init`, returned both as a typed handle /// (for inspection) and as the erased `Arc` the primitives consume. @@ -209,6 +208,50 @@ mod tests { assert_eq!(out, vec![64, 65, 66, 67, 0, 1, 2, 3]); } + struct RecordingMemory { + read_order: Mutex>, + } + + impl MemoryModel for RecordingMemory { + async fn read(&self, addr: u64) -> [u8; 64] { + self.read_order.lock().unwrap().push(addr); + [0; 64] + } + + async fn write(&self, _addr: u64, _bytes: [u8; 64]) {} + } + + #[tokio::test] + async fn test_gather_issues_reads_in_input_order() { + let memory = Arc::new(RecordingMemory { + read_order: Mutex::new(Vec::new()), + }); + let hbm: Arc = memory.clone(); + let _ = gather( + &hbm, + 12, + vec![ + ChunkRead { + addr: 128, + dst_offset: 0, + len: 4, + }, + ChunkRead { + addr: 0, + dst_offset: 4, + len: 4, + }, + ChunkRead { + addr: 64, + dst_offset: 8, + len: 4, + }, + ], + ) + .await; + assert_eq!(*memory.read_order.lock().unwrap(), vec![128, 0, 64]); + } + #[tokio::test] async fn test_gather_clamps_read_to_block_end() { let (_mb, hbm) = seeded(128, |b| { diff --git a/transactional_emulator/lib/ramulator/src/model.rs b/transactional_emulator/lib/ramulator/src/model.rs index 635b3dea..e9655b82 100644 --- a/transactional_emulator/lib/ramulator/src/model.rs +++ b/transactional_emulator/lib/ramulator/src/model.rs @@ -162,20 +162,18 @@ impl Ramulator { impl memory::MemoryTimingModel for Ramulator { async fn read(&self, addr: u64) { - futures::future::join_all( - (0..64) - .step_by(self.0.transfer_size as _) - .map(|offset| self.read_transfer(addr + offset)), - ) - .await; + let transfers: Vec<_> = (0..64u64) + .step_by(self.0.transfer_size as usize) + .map(|offset| self.read_transfer(addr + offset)) + .collect(); + futures::future::join_all(transfers).await; } async fn write(&self, addr: u64) { - futures::future::join_all( - (0..64) - .step_by(self.0.transfer_size as _) - .map(|offset| self.write_transfer(addr + offset)), - ) - .await; + let transfers: Vec<_> = (0..64u64) + .step_by(self.0.transfer_size as usize) + .map(|offset| self.write_transfer(addr + offset)) + .collect(); + futures::future::join_all(transfers).await; } } diff --git a/transactional_emulator/lib/runtime/src/executor.rs b/transactional_emulator/lib/runtime/src/executor.rs index baa2c843..a7c2bd98 100644 --- a/transactional_emulator/lib/runtime/src/executor.rs +++ b/transactional_emulator/lib/runtime/src/executor.rs @@ -20,13 +20,14 @@ impl Future for ResolveAt { struct Timer { executor: Executor, resolve_at: Instant, + sequence_id: u64, waker: Option, _phantom: PhantomPinned, } impl PartialEq for Timer { fn eq(&self, other: &Self) -> bool { - core::ptr::eq(self, other) + self.sequence_id == other.sequence_id } } @@ -34,10 +35,10 @@ impl Eq for Timer {} impl Ord for Timer { fn cmp(&self, other: &Self) -> core::cmp::Ordering { - // Only compare equal when they're the same task, so we can keep multiple copies in the B-Tree. + // Break same-instant ties by allocation-independent creation order. self.resolve_at .cmp(&other.resolve_at) - .then((self as *const Self).cmp(&(other as _))) + .then(self.sequence_id.cmp(&other.sequence_id)) } } @@ -108,6 +109,7 @@ impl Wake for Task { struct ExecutorInner { now: Instant, + next_timer_sequence_id: u64, ready_tasks: VecDeque>, timers: BTreeSet<&'static mut Timer>, } @@ -125,6 +127,7 @@ impl Executor { pub fn new() -> Self { Self(Arc::new(Mutex::new(ExecutorInner { now: Instant::INIT, + next_timer_sequence_id: 0, ready_tasks: VecDeque::new(), timers: BTreeSet::new(), }))) @@ -147,10 +150,20 @@ impl Executor { /// Create a timer that is resolved at later time. pub fn resolve_at(&self, fire_at: impl Deadline) -> ResolveAt { - let instant = fire_at.to_instant(self.now()); + let (instant, sequence_id) = { + let mut inner = self.0.lock().unwrap(); + let instant = fire_at.to_instant(inner.now); + let sequence_id = inner.next_timer_sequence_id; + inner.next_timer_sequence_id = inner + .next_timer_sequence_id + .checked_add(1) + .expect("executor timer sequence id overflowed"); + (instant, sequence_id) + }; ResolveAt(Timer { executor: self.clone(), resolve_at: instant, + sequence_id, waker: None, _phantom: PhantomPinned, }) @@ -328,6 +341,21 @@ mod tests { assert_eq!(ex.now(), at(7)); } + #[tokio::test] + async fn test_same_instant_events_fire_in_schedule_order() { + let ex = Executor::new(); + let log = Arc::new(Mutex::new(Vec::new())); + for id in 0..8 { + let l = log.clone(); + ex.schedule(ns(7), async move { + l.lock().unwrap().push(id); + }); + } + ex.enter(Instant::ETERNITY).await; + assert_eq!(*log.lock().unwrap(), (0..8).collect::>()); + assert_eq!(ex.now(), at(7)); + } + #[tokio::test] async fn test_timeout_excludes_event_at_deadline() { let ex = Executor::new(); diff --git a/transactional_emulator/lib/runtime/src/time.rs b/transactional_emulator/lib/runtime/src/time.rs index 1c52e14b..beff8280 100644 --- a/transactional_emulator/lib/runtime/src/time.rs +++ b/transactional_emulator/lib/runtime/src/time.rs @@ -33,6 +33,14 @@ impl Duration { pub const fn from_secs(v: u64) -> Self { Self(v * 1_000_000_000_000) } + + pub const fn as_picos(&self) -> u64 { + self.0 + } + + pub const fn as_nanos_floor(&self) -> u64 { + self.0 / 1000 + } } impl Add for Duration { @@ -110,6 +118,10 @@ impl Instant { pub const fn to_secs(&self) -> f64 { self.0 as f64 / (1_000_000_000_000_u64 as f64) } + + pub const fn as_picos(&self) -> u64 { + self.0 + } } pub trait Deadline { @@ -144,6 +156,8 @@ mod tests { Duration::from_secs(1), Duration::from_picos(1_000_000_000_000) ); + assert_eq!(Duration::from_picos(1500).as_picos(), 1500); + assert_eq!(Duration::from_picos(1500).as_nanos_floor(), 1); } #[test] diff --git a/transactional_emulator/src/accelerator/dispatch.rs b/transactional_emulator/src/accelerator/dispatch.rs index b5401947..12f3b0d3 100644 --- a/transactional_emulator/src/accelerator/dispatch.rs +++ b/transactional_emulator/src/accelerator/dispatch.rs @@ -11,7 +11,7 @@ use crate::runtime_config::{ SCALAR_FP_BASIC_CYCLES, SCALAR_FP_EXP_CYCLES, SCALAR_FP_RECI_CYCLES, SCALAR_FP_SQRT_CYCLES, SCALAR_INT_BASIC_CYCLES, STORE_V_AMOUNT, VECTOR_ACTIVATION_TYPE, VECTOR_KV_TYPE, VLEN, }; -use crate::stage_profile::StageProfiler; +use crate::stage_profile::{ResourceKind, StageProfiler}; use crate::{cycle, dma, op}; use runtime::Executor; @@ -60,8 +60,8 @@ impl Accelerator { while pc < ops.len() { let executed_pc = pc; let op = &ops[pc]; - let profile_start_secs = if stage_profiler.is_some() { - Some(Executor::current().now().to_secs()) + let profile_start_instant = if stage_profiler.is_some() { + Some(Executor::current().now()) } else { None }; @@ -625,10 +625,12 @@ impl Accelerator { pc += 1; } - if let (Some(start_secs), Some(profiler)) = - (profile_start_secs, stage_profiler.as_deref_mut()) + if let (Some(start_instant), Some(profiler)) = + (profile_start_instant, stage_profiler.as_deref_mut()) { - let elapsed_secs = Executor::current().now().to_secs() - start_secs; + let elapsed_duration = Executor::current().now() - start_instant; + let elapsed_secs = elapsed_duration.as_picos() as f64 / 1_000_000_000_000.0; + let elapsed_cycles = StageProfiler::duration_to_cycles(elapsed_duration); let (hbm_bytes_read, hbm_bytes_written) = if let (Some(before), Some(after)) = (profile_start_hbm, self.hbm.statistics()) { @@ -643,8 +645,78 @@ impl Accelerator { } else { (0, 0) }; - profiler.record(executed_pc, elapsed_secs, hbm_bytes_read, hbm_bytes_written); + profiler.record( + executed_pc, + elapsed_secs, + elapsed_cycles, + resource_kind_for_opcode(op), + hbm_bytes_read, + hbm_bytes_written, + ); } } } } + +fn resource_kind_for_opcode(op: &op::Opcode) -> ResourceKind { + match op { + op::Opcode::M_MM { .. } + | op::Opcode::M_TMM { .. } + | op::Opcode::M_BMM { .. } + | op::Opcode::M_BTMM { .. } + | op::Opcode::M_BMM_WO { .. } + | op::Opcode::M_MM_WO { .. } + | op::Opcode::M_MV { .. } + | op::Opcode::M_TMV { .. } + | op::Opcode::M_BMV { .. } + | op::Opcode::M_BTMV { .. } + | op::Opcode::M_MV_WO { .. } + | op::Opcode::M_BMV_WO { .. } => ResourceKind::Matrix, + + op::Opcode::V_ADD_VV { .. } + | op::Opcode::V_ADD_VF { .. } + | op::Opcode::V_SUB_VV { .. } + | op::Opcode::V_SUB_VF { .. } + | op::Opcode::V_MUL_VV { .. } + | op::Opcode::V_MUL_VF { .. } + | op::Opcode::V_MAX_VF { .. } + | op::Opcode::V_MIN_VF { .. } + | op::Opcode::V_TOPK { .. } + | op::Opcode::V_EXP_V { .. } + | op::Opcode::V_RECI_V { .. } + | op::Opcode::V_RED_SUM { .. } + | op::Opcode::V_RED_MAX { .. } + | op::Opcode::V_SHFT_V { .. } => ResourceKind::Vector, + + op::Opcode::S_ADD_FP { .. } + | op::Opcode::S_SUB_FP { .. } + | op::Opcode::S_MAX_FP { .. } + | op::Opcode::S_MUL_FP { .. } + | op::Opcode::S_EXP_FP { .. } + | op::Opcode::S_RECI_FP { .. } + | op::Opcode::S_SQRT_FP { .. } + | op::Opcode::S_LD_FP { .. } + | op::Opcode::S_ST_FP { .. } + | op::Opcode::S_MAP_V_FP { .. } + | op::Opcode::S_ADD_INT { .. } + | op::Opcode::S_ADDI_INT { .. } + | op::Opcode::S_SUB_INT { .. } + | op::Opcode::S_MUL_INT { .. } + | op::Opcode::S_LUI_INT { .. } + | op::Opcode::S_LD_INT { .. } + | op::Opcode::S_ST_INT { .. } + | op::Opcode::C_SET_ADDR_REG { .. } + | op::Opcode::C_SET_SCALE_REG { .. } + | op::Opcode::C_SET_STRIDE_REG { .. } + | op::Opcode::C_SET_V_MASK_REG { .. } + | op::Opcode::C_LOOP_START { .. } + | op::Opcode::C_LOOP_END { .. } + | op::Opcode::C_BREAK => ResourceKind::Scalar, + + op::Opcode::H_PREFETCH_M { .. } + | op::Opcode::H_PREFETCH_V { .. } + | op::Opcode::H_STORE_V { .. } => ResourceKind::Dma, + + op::Opcode::Invalid => ResourceKind::Other, + } +} diff --git a/transactional_emulator/src/main.rs b/transactional_emulator/src/main.rs index 8d732704..5b74b768 100644 --- a/transactional_emulator/src/main.rs +++ b/transactional_emulator/src/main.rs @@ -26,5 +26,34 @@ async fn main() { let executor = Executor::new(); executor.spawn(runner::run_from_cli()); executor.enter(Instant::ETERNITY).await; - tracing::info!("Simulation completed. Latency {:?}", executor.now()); + let latency = executor.now() - Instant::INIT; + let cycles = latency + .as_picos() + .div_ceil(runtime_config::PERIOD.as_picos().max(1)); + tracing::info!( + "Simulation completed. Latency {:?} cycles {}", + executor.now(), + cycles + ); +} + +#[cfg(test)] +mod timing_golden_tests { + const FIXTURE: &str = include_str!("../testbench/timing_goldens/golden_workloads.json"); + + #[test] + fn timing_golden_fixture_pins_required_workloads() { + for needle in [ + "\"gpt_synthetic_small\"", + "\"qwen_synthetic_small\"", + "\"gpt_real_layer0_tok1\"", + "\"qwen_real_decoder_block\"", + "\"sim_latency_cycles\": 147758", + "\"sim_latency_cycles\": 256559", + "\"sim_latency_cycles\": 34354344", + "\"sim_latency_cycles\": 65073075", + ] { + assert!(FIXTURE.contains(needle), "missing fixture entry: {needle}"); + } + } } diff --git a/transactional_emulator/src/runner.rs b/transactional_emulator/src/runner.rs index b3233ae9..4f2a2f0c 100644 --- a/transactional_emulator/src/runner.rs +++ b/transactional_emulator/src/runner.rs @@ -2,7 +2,7 @@ use std::io::Write; use std::mem::ManuallyDrop; use std::sync::Arc; -use runtime::Executor; +use runtime::{Executor, Instant}; use sram::{MatrixSram, VectorSram}; use tracing_subscriber::prelude::*; @@ -219,6 +219,10 @@ pub(crate) async fn run_from_cli() { .do_ops(&decoded_ops, stage_profiler.as_mut()) .await; + if let Some(profile) = stage_profiler.as_mut() { + profile.set_total_simulation_duration(Executor::current().now() - Instant::INIT); + } + if let Some(profile) = stage_profiler.as_ref() { let out_path = opts .stage_profile_out diff --git a/transactional_emulator/src/stage_profile.rs b/transactional_emulator/src/stage_profile.rs index 8c44e7f7..cc0baddd 100644 --- a/transactional_emulator/src/stage_profile.rs +++ b/transactional_emulator/src/stage_profile.rs @@ -2,6 +2,19 @@ use std::collections::BTreeMap; use std::fs; use std::path::Path; +use runtime::Duration; + +use crate::runtime_config::PERIOD; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ResourceKind { + Matrix, + Vector, + Scalar, + Dma, + Other, +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum StageKind { RouterTopk, @@ -68,9 +81,47 @@ impl StageKind { #[derive(Clone, Copy, Debug, Default)] struct StageRuntime { instructions: u64, + wall_cycles: u64, seconds: f64, hbm_bytes_read: u64, hbm_bytes_written: u64, + resource_proxy: ResourceRuntime, +} + +#[derive(Clone, Copy, Debug, Default)] +struct ResourceRuntime { + matrix_cycles: u64, + vector_cycles: u64, + scalar_cycles: u64, + dma_cycles: u64, + ramulator_proxy_cycles: u64, + other_cycles: u64, +} + +impl ResourceRuntime { + fn add(&mut self, resource: ResourceKind, cycles: u64) { + match resource { + ResourceKind::Matrix => self.matrix_cycles += cycles, + ResourceKind::Vector => self.vector_cycles += cycles, + ResourceKind::Scalar => self.scalar_cycles += cycles, + ResourceKind::Dma => { + self.dma_cycles += cycles; + // Sub-view only: ramulator_proxy_cycles is included in dma_cycles. + // Totals should use matrix+vector+scalar+dma+other, not both. + self.ramulator_proxy_cycles += cycles; + } + ResourceKind::Other => self.other_cycles += cycles, + } + } + + fn add_runtime(&mut self, other: Self) { + self.matrix_cycles += other.matrix_cycles; + self.vector_cycles += other.vector_cycles; + self.scalar_cycles += other.scalar_cycles; + self.dma_cycles += other.dma_cycles; + self.ramulator_proxy_cycles += other.ramulator_proxy_cycles; + self.other_cycles += other.other_cycles; + } } pub(crate) struct StageProfiler { @@ -79,9 +130,12 @@ pub(crate) struct StageProfiler { stages: [StageRuntime; 11], pair_stages: BTreeMap, total_instructions: u64, + total_profiled_cycles: u64, + total_simulation_cycles: Option, total_seconds: f64, total_hbm_bytes_read: u64, total_hbm_bytes_written: u64, + total_resource_proxy: ResourceRuntime, } impl StageProfiler { @@ -127,29 +181,50 @@ impl StageProfiler { stages: [StageRuntime::default(); 11], pair_stages: BTreeMap::new(), total_instructions: 0, + total_profiled_cycles: 0, + total_simulation_cycles: None, total_seconds: 0.0, total_hbm_bytes_read: 0, total_hbm_bytes_written: 0, + total_resource_proxy: ResourceRuntime::default(), }) } + // First-pass proxy: per-op div_ceil can systematically overcount when op + // durations are not exact cycle multiples. Calibrate this with RTL primitive + // measurements before treating stage-profile cycle sums as final timing. + pub(crate) fn duration_to_cycles(duration: Duration) -> u64 { + let period_picos = PERIOD.as_picos().max(1); + duration.as_picos().div_ceil(period_picos) + } + + pub(crate) fn set_total_simulation_duration(&mut self, duration: Duration) { + self.total_simulation_cycles = Some(Self::duration_to_cycles(duration)); + } + pub(crate) fn record( &mut self, pc: usize, seconds: f64, + wall_cycles: u64, + resource: ResourceKind, hbm_bytes_read: u64, hbm_bytes_written: u64, ) { let stage = self.labels.get(pc).copied().unwrap_or(StageKind::Other); let bucket = &mut self.stages[stage.index()]; bucket.instructions += 1; + bucket.wall_cycles += wall_cycles; bucket.seconds += seconds; bucket.hbm_bytes_read += hbm_bytes_read; bucket.hbm_bytes_written += hbm_bytes_written; + bucket.resource_proxy.add(resource, wall_cycles); self.total_instructions += 1; + self.total_profiled_cycles += wall_cycles; self.total_seconds += seconds; self.total_hbm_bytes_read += hbm_bytes_read; self.total_hbm_bytes_written += hbm_bytes_written; + self.total_resource_proxy.add(resource, wall_cycles); if let Some(pair_id) = self.pair_labels.get(pc).copied().flatten() { let pair_buckets = self @@ -158,21 +233,54 @@ impl StageProfiler { .or_insert([StageRuntime::default(); 11]); let pair_bucket = &mut pair_buckets[stage.index()]; pair_bucket.instructions += 1; + pair_bucket.wall_cycles += wall_cycles; pair_bucket.seconds += seconds; pair_bucket.hbm_bytes_read += hbm_bytes_read; pair_bucket.hbm_bytes_written += hbm_bytes_written; + pair_bucket.resource_proxy.add(resource, wall_cycles); } } pub(crate) fn write_json(&self, path: &Path) -> std::io::Result<()> { let mut out = String::new(); + let total_stage_wall_cycles = sum_stage_runtimes(&self.stages).wall_cycles; + let total_simulation_cycles = self.total_simulation_cycles; + let total_unprofiled_cycles = total_simulation_cycles + .map(|cycles| cycles.saturating_sub(self.total_profiled_cycles)) + .unwrap_or(0); + let cycle_accounting_status = match total_simulation_cycles { + Some(cycles) if cycles == self.total_profiled_cycles => "profiled_cycles_match_total", + Some(_) => "profiled_cycles_do_not_match_total", + None => "total_simulation_cycles_unset", + }; + out.push_str("{\n"); - out.push_str(" \"schema_version\": 1,\n"); + out.push_str(" \"schema_version\": 2,\n"); out.push_str(&format!(" \"label_count\": {},\n", self.labels.len())); out.push_str(&format!( " \"total_instructions_executed\": {},\n", self.total_instructions )); + match total_simulation_cycles { + Some(cycles) => out.push_str(&format!(" \"total_simulation_cycles\": {},\n", cycles)), + None => out.push_str(" \"total_simulation_cycles\": null,\n"), + } + out.push_str(&format!( + " \"total_profiled_cycles\": {},\n", + self.total_profiled_cycles + )); + out.push_str(&format!( + " \"total_stage_wall_cycles\": {},\n", + total_stage_wall_cycles + )); + out.push_str(&format!( + " \"total_unprofiled_cycles\": {},\n", + total_unprofiled_cycles + )); + out.push_str(&format!( + " \"cycle_accounting_status\": \"{}\",\n", + cycle_accounting_status + )); out.push_str(&format!( " \"total_profiled_seconds\": {:.12},\n", self.total_seconds @@ -185,6 +293,13 @@ impl StageProfiler { " \"total_hbm_bytes_written\": {},\n", self.total_hbm_bytes_written )); + out.push_str(&format!( + " \"total_resource_proxy_cycles\": {},\n", + resource_json(self.total_resource_proxy) + )); + out.push_str(" \"logical_byte_status\": \"not_declared_by_opcode_profile; join benchmark route/shape formulas for logical bytes\",\n"); + out.push_str(" \"physical_byte_status\": \"HBM bytes are emulator WithStats 64B physical transfer deltas\",\n"); + out.push_str(" \"resource_cycle_status\": \"first-pass opcode-class wall-cycle proxy, not calibrated per-component busy counters\",\n"); out.push_str(" \"stages\": {\n"); for (idx, stage) in StageKind::ALL.iter().enumerate() { @@ -199,15 +314,25 @@ impl StageProfiler { } else { stats.seconds / self.total_seconds }; + let cycle_fraction = if self.total_profiled_cycles == 0 { + 0.0 + } else { + stats.wall_cycles as f64 / self.total_profiled_cycles as f64 + }; out.push_str(&format!( - " \"{}\": {{\"instructions\": {}, \"seconds\": {:.12}, \"instruction_fraction\": {:.12}, \"time_fraction\": {:.12}, \"hbm_bytes_read\": {}, \"hbm_bytes_written\": {}}}", + " \"{}\": {{\"instructions\": {}, \"wall_cycles\": {}, \"seconds\": {:.12}, \"instruction_fraction\": {:.12}, \"time_fraction\": {:.12}, \"cycle_fraction\": {:.12}, \"logical_bytes_read\": null, \"logical_bytes_written\": null, \"physical_hbm_bytes_read\": {}, \"physical_hbm_bytes_written\": {}, \"hbm_bytes_read\": {}, \"hbm_bytes_written\": {}, \"resource_proxy_cycles\": {}}}", stage.name(), stats.instructions, + stats.wall_cycles, stats.seconds, instr_fraction, time_fraction, + cycle_fraction, + stats.hbm_bytes_read, + stats.hbm_bytes_written, stats.hbm_bytes_read, - stats.hbm_bytes_written + stats.hbm_bytes_written, + resource_json(stats.resource_proxy) )); if idx + 1 != StageKind::ALL.len() { out.push(','); @@ -221,22 +346,30 @@ impl StageProfiler { for (pair_idx, (pair_id, stages)) in self.pair_stages.iter().enumerate() { let totals = sum_stage_runtimes(stages); out.push_str(&format!( - " \"{}\": {{\"instructions\": {}, \"seconds\": {:.12}, \"hbm_bytes_read\": {}, \"hbm_bytes_written\": {}, \"stages\": {{\n", + " \"{}\": {{\"instructions\": {}, \"wall_cycles\": {}, \"seconds\": {:.12}, \"logical_bytes_read\": null, \"logical_bytes_written\": null, \"physical_hbm_bytes_read\": {}, \"physical_hbm_bytes_written\": {}, \"hbm_bytes_read\": {}, \"hbm_bytes_written\": {}, \"resource_proxy_cycles\": {}, \"stages\": {{\n", pair_id, totals.instructions, + totals.wall_cycles, totals.seconds, totals.hbm_bytes_read, - totals.hbm_bytes_written + totals.hbm_bytes_written, + totals.hbm_bytes_read, + totals.hbm_bytes_written, + resource_json(totals.resource_proxy) )); for (stage_idx, stage) in StageKind::ALL.iter().enumerate() { let stats = stages[stage.index()]; out.push_str(&format!( - " \"{}\": {{\"instructions\": {}, \"seconds\": {:.12}, \"hbm_bytes_read\": {}, \"hbm_bytes_written\": {}}}", + " \"{}\": {{\"instructions\": {}, \"wall_cycles\": {}, \"seconds\": {:.12}, \"logical_bytes_read\": null, \"logical_bytes_written\": null, \"physical_hbm_bytes_read\": {}, \"physical_hbm_bytes_written\": {}, \"hbm_bytes_read\": {}, \"hbm_bytes_written\": {}, \"resource_proxy_cycles\": {}}}", stage.name(), stats.instructions, + stats.wall_cycles, stats.seconds, stats.hbm_bytes_read, - stats.hbm_bytes_written + stats.hbm_bytes_written, + stats.hbm_bytes_read, + stats.hbm_bytes_written, + resource_json(stats.resource_proxy) )); if stage_idx + 1 != StageKind::ALL.len() { out.push(','); @@ -253,7 +386,7 @@ impl StageProfiler { out.push_str(" },\n"); out.push_str( - " \"caveat\": \"Stage and routed-pair labels are derived from generated ASM comments. Time and HBM bytes are attributed by dynamically executed opcode. Bytes are measured from the global HBM stats delta before/after each opcode. Pair labels identify static routed pair slots, not necessarily unique expert IDs without joining the routing dump.\"\n", + " \"caveat\": \"Stage and routed-pair labels are derived from generated ASM comments. Cycles use simulator time only. physical_hbm_bytes_* are measured from the global WithStats 64B HBM deltas before/after each opcode. logical_bytes_* are intentionally null here and must be joined from workload shape/route formulas. resource_proxy_cycles are first-pass opcode-class wall-cycle attribution, not calibrated component busy counters. ramulator_proxy is a sub-view of dma, not an additive peer; totals should use matrix+vector+scalar+dma+other. Current do_ops still awaits each opcode, so this profile does not by itself prove cross-op overlap. Pair labels identify static routed pair slots, not necessarily unique expert IDs without joining the routing dump.\"\n", ); out.push_str("}\n"); fs::write(path, out) @@ -264,13 +397,29 @@ fn sum_stage_runtimes(stages: &[StageRuntime; 11]) -> StageRuntime { let mut total = StageRuntime::default(); for stats in stages { total.instructions += stats.instructions; + total.wall_cycles += stats.wall_cycles; total.seconds += stats.seconds; total.hbm_bytes_read += stats.hbm_bytes_read; total.hbm_bytes_written += stats.hbm_bytes_written; + total.resource_proxy.add_runtime(stats.resource_proxy); } total } +fn resource_json(stats: ResourceRuntime) -> String { + // ramulator_proxy is a DMA sub-view for readers that want a memory proxy; + // do not add it to dma when computing total resource proxy cycles. + format!( + "{{\"matrix\": {}, \"vector\": {}, \"scalar\": {}, \"dma\": {}, \"ramulator_proxy\": {}, \"other\": {}}}", + stats.matrix_cycles, + stats.vector_cycles, + stats.scalar_cycles, + stats.dma_cycles, + stats.ramulator_proxy_cycles, + stats.other_cycles + ) +} + fn is_opcode_line(line: &str) -> bool { line.as_bytes() .first() @@ -362,3 +511,28 @@ fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option { .windows(needle.len()) .position(|window| window == needle) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn duration_to_cycles_rounds_up_to_period() { + assert_eq!( + StageProfiler::duration_to_cycles(Duration::from_picos(0)), + 0 + ); + assert_eq!( + StageProfiler::duration_to_cycles(Duration::from_picos(999)), + 1 + ); + assert_eq!( + StageProfiler::duration_to_cycles(Duration::from_picos(1000)), + 1 + ); + assert_eq!( + StageProfiler::duration_to_cycles(Duration::from_picos(1001)), + 2 + ); + } +} diff --git a/transactional_emulator/testbench/direct_emit/v_shift_v_test.py b/transactional_emulator/testbench/direct_emit/v_shift_v_test.py index 9f210c73..98d80146 100644 --- a/transactional_emulator/testbench/direct_emit/v_shift_v_test.py +++ b/transactional_emulator/testbench/direct_emit/v_shift_v_test.py @@ -1,3 +1,5 @@ +# Historical filename retained because direct_emit/README.md still references it; +# the emitted ISA mnemonic is the RTL-aligned V_SHFT_V. from pathlib import Path diff --git a/transactional_emulator/testbench/emulator_runner.py b/transactional_emulator/testbench/emulator_runner.py index 1fe74752..d4759f24 100644 --- a/transactional_emulator/testbench/emulator_runner.py +++ b/transactional_emulator/testbench/emulator_runner.py @@ -30,13 +30,20 @@ def _build_emulator_binary(emulator_dir: Path, binary: Path) -> None: the binary is already up to date, so this stays out of the way on warm runs. """ print( - f"Emulator binary not found at {binary}\n" - "Building it now (one-time release compile; subsequent runs reuse it)...", + f"Ensuring emulator binary is current at {binary}\n" + "Running cargo build --release (incremental/no-op when already current)...", file=sys.stderr, flush=True, ) + build_cmd = _release_build_command(emulator_dir) + if build_cmd[0] == "nix": + print( + "Direct cargo is unavailable; using nix develop -c cargo build --release.", + file=sys.stderr, + flush=True, + ) result = subprocess.run( - ["cargo", "build", "--release"], + build_cmd, cwd=str(emulator_dir), env={**os.environ, "RUST_BACKTRACE": "1"}, ) @@ -47,7 +54,41 @@ def _build_emulator_binary(emulator_dir: Path, binary: Path) -> None: ) -def run_emulator(build_dir: Path, hbm_size: int | None = None, threads: int | None = None) -> dict: +def _release_build_command(emulator_dir: Path) -> list[str]: + cargo_probe = subprocess.run( + ["cargo", "--version"], + cwd=str(emulator_dir), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + if cargo_probe.returncode == 0: + return ["cargo", "build", "--release"] + if (emulator_dir.parent / "flake.nix").exists(): + return ["nix", "develop", "-c", "cargo", "build", "--release"] + return ["cargo", "build", "--release"] + + +def _env_flag(name: str) -> bool: + return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"} + + +def _run_file_suffix(run_label: str | None) -> str: + if not run_label: + return "" + safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", run_label.strip()) + if not safe: + raise ValueError("run_label must contain at least one filename-safe character") + return f".{safe}" + + +def run_emulator( + build_dir: Path, + hbm_size: int | None = None, + threads: int | None = None, + stage_profile: bool | None = None, + stage_profile_out: Path | None = None, + run_label: str | None = None, +) -> dict: """Run the Rust transactional emulator with build artifacts from build_dir. Args: @@ -64,12 +105,22 @@ def run_emulator(build_dir: Path, hbm_size: int | None = None, threads: int | No on-disk size, rounded up to the next 64-byte multiple. This matches the actual preload — anything beyond is unused virtual space that the emulator would otherwise lazy-commit pages into. + stage_profile: when true, pass generated_asm_code.asm to the Rust + stage profiler and write stage_profile.json in build_dir. + When None, PLENA_EMULATOR_STAGE_PROFILE controls it. + run_label: optional filename suffix for repeat/determinism runs. When + omitted, output filenames keep their historical names. """ emulator_dir = Path(__file__).parent.parent # transactional_emulator/ binary = emulator_dir / "target" / "release" / "transactional_emulator" - if not binary.exists(): - _build_emulator_binary(emulator_dir, binary) + if stage_profile is None: + stage_profile = _env_flag("PLENA_EMULATOR_STAGE_PROFILE") + + # Always rebuild before running. `cargo build --release` is a fast no-op when + # current, and this prevents false failures from new ASM hitting a stale + # release binary with old opcode decode logic. + _build_emulator_binary(emulator_dir, binary) asm_path = build_dir / "generated_machine_code.mem" hbm_path = build_dir / "hbm_for_behave_sim.bin" @@ -121,6 +172,26 @@ def run_emulator(build_dir: Path, hbm_size: int | None = None, threads: int | No if vram_preload_path.exists(): cmd += ["--vram", str(vram_preload_path)] + run_suffix = _run_file_suffix(run_label) + if stage_profile_out is not None: + profile_out_path = stage_profile_out + elif run_suffix: + profile_out_path = build_dir / f"stage_profile{run_suffix}.json" + else: + profile_out_path = build_dir / "stage_profile.json" + if stage_profile: + profile_asm_path = build_dir / "generated_asm_code.asm" + if not profile_asm_path.exists(): + raise FileNotFoundError( + f"stage_profile=True requires ASM comments at {profile_asm_path}" + ) + cmd += [ + "--stage-profile-asm", + str(profile_asm_path), + "--stage-profile-out", + str(profile_out_path), + ] + # tch's download-libtorch stores libtorch in the Cargo build cache. # The binary needs LD_LIBRARY_PATH to find it at runtime. libtorch_pattern = str( @@ -147,7 +218,7 @@ def run_emulator(build_dir: Path, hbm_size: int | None = None, threads: int | No new_ldpath = libtorch_dirs[0] env["LD_LIBRARY_PATH"] = f"{new_ldpath}:{existing_ldpath}" if existing_ldpath else new_ldpath - log_path = build_dir / "rust_emulator_stdout.log" + log_path = build_dir / f"rust_emulator_stdout{run_suffix}.log" started_at = datetime.now(UTC) start = time.perf_counter() metrics: dict[str, object] = { @@ -161,9 +232,16 @@ def run_emulator(build_dir: Path, hbm_size: int | None = None, threads: int | No "hbm_size_bytes": hbm_size, "artifacts": _artifact_summary(build_dir, asm_path, hbm_path), "log_path": str(log_path), + "stage_profile_requested": bool(stage_profile), } + if run_label: + metrics["run_label"] = run_label + if stage_profile: + metrics["stage_profile_path"] = str(profile_out_path) - sim_latency_re = re.compile(r"Simulation completed\. Latency\s+([0-9.eE+-]+)ns") + sim_latency_re = re.compile( + r"Simulation completed\. Latency\s+([0-9.eE+-]+)ns(?:\s+cycles\s+([0-9]+))?" + ) topology_re = re.compile(r"mlen=(\d+)\s+vlen=(\d+)\s+.*blen=(\d+)") hbm_stats_re = re.compile( r"HBM Statistics - Bytes read:\s*([0-9]+)\s*\|\s*" @@ -197,6 +275,8 @@ def run_emulator(build_dir: Path, hbm_size: int | None = None, threads: int | No sim_latency_ns = float(sim_match.group(1)) metrics["sim_latency_ns"] = sim_latency_ns metrics["sim_latency_ms"] = sim_latency_ns / 1_000_000.0 + if sim_match.group(2) is not None: + metrics["sim_latency_cycles"] = int(sim_match.group(2)) topo_match = topology_re.search(line) if topo_match: @@ -216,8 +296,13 @@ def run_emulator(build_dir: Path, hbm_size: int | None = None, threads: int | No metrics["ended_at_utc"] = ended_at.isoformat() metrics["host_wall_time_seconds"] = time.perf_counter() - start metrics["return_code"] = return_code + if stage_profile: + metrics["stage_profile_exists"] = profile_out_path.exists() + if profile_out_path.exists(): + metrics["stage_profile_size_bytes"] = profile_out_path.stat().st_size - stats_path = build_dir / "rust_emulator_run_stats.json" + stats_path = build_dir / f"rust_emulator_run_stats{run_suffix}.json" + metrics["stats_path"] = str(stats_path) stats_path.write_text(json.dumps(metrics, indent=2) + "\n", encoding="utf-8") print(f"Rust emulator host wall time: {metrics['host_wall_time_seconds']:.3f}s (stats: {stats_path})") @@ -253,6 +338,77 @@ def run_emulator(build_dir: Path, hbm_size: int | None = None, threads: int | No return metrics +def run_emulator_repeat_gate( + build_dir: Path, + repeats: int = 3, + hbm_size: int | None = None, + threads: int | None = None, + stage_profile: bool | None = None, +) -> dict: + """Run the same emulator artifact repeatedly and require identical cycles. + + This is intentionally opt-in: ordinary functional tests still call + `run_emulator()` once, while timing baselines can call this helper to prove + that a measurement is deterministic before accepting it. + """ + if repeats < 2: + raise ValueError("repeat gate requires at least two runs") + + build_dir = Path(build_dir) + runs = [] + for index in range(repeats): + label = f"repeat{index + 1:02d}" + print(f"\n--- Repeat gate run {index + 1}/{repeats} ({label}) ---") + runs.append( + run_emulator( + build_dir, + hbm_size=hbm_size, + threads=threads, + stage_profile=stage_profile, + run_label=label, + ) + ) + + required_keys = ("sim_latency_cycles",) + optional_stable_keys = ("hbm_bytes_read", "hbm_bytes_written") + series: dict[str, list[object]] = { + key: [run.get(key) for run in runs] for key in required_keys + optional_stable_keys + } + + missing_required = [key for key in required_keys if any(value is None for value in series[key])] + if missing_required: + raise RuntimeError( + "Repeat gate could not find required metrics: " + ", ".join(missing_required) + ) + + stable_checks = { + key: len(set(values)) == 1 + for key, values in series.items() + if key in required_keys or all(value is not None for value in values) + } + passed = all(stable_checks.values()) + summary = { + "schema_version": 1, + "build_dir": str(build_dir), + "repeats": repeats, + "passed": passed, + "stable_checks": stable_checks, + "series": series, + "run_stats_paths": [run.get("stats_path") for run in runs], + } + summary_path = build_dir / "rust_emulator_repeat_gate.json" + summary["summary_path"] = str(summary_path) + summary_path.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8") + + if not passed: + raise RuntimeError( + f"Repeat gate failed; metric series written to {summary_path}: {series}" + ) + + print(f"Repeat gate PASSED: {series} (summary: {summary_path})") + return summary + + def _current_plena_settings_path() -> Path: return Path(os.environ.get("PLENA_SETTINGS_TOML", Path(__file__).parents[2] / "plena_settings.toml")) @@ -358,7 +514,13 @@ def _current_vector_sram_fp_format() -> tuple[int, int, int]: def run_and_assert( - build_dir: Path, op_name: str, mlen: int = 64, blen: int = 4, vlen: int | None = None, threads: int | None = None + build_dir: Path, + op_name: str, + mlen: int = 64, + blen: int = 4, + vlen: int | None = None, + threads: int | None = None, + stage_profile: bool | None = None, ) -> dict: """ Sync HW config, run the Rust emulator, compare output, exit(1) on failure. @@ -376,7 +538,7 @@ def run_and_assert( update_plena_config(vlen=vlen, mlen=mlen, blen=blen, verbose=False) print("\n--- Running Rust transactional emulator ---") - run_metrics = run_emulator(build_dir, threads=threads) + run_metrics = run_emulator(build_dir, threads=threads, stage_profile=stage_profile) emu_mlen = run_metrics.get("emu_mlen") emu_blen = run_metrics.get("emu_blen") @@ -412,6 +574,7 @@ def emulate_from_result( blen: int = 4, vlen: int | None = None, threads: int | None = None, + stage_profile: bool | None = None, ) -> dict: """Write sim artifacts from a compile result dict and run the Rust emulator. @@ -452,4 +615,12 @@ def emulate_from_result( with open(build_dir / "generated_asm_code.asm", "w") as f: f.write(result["isa"]) - return run_and_assert(build_dir, asm_name, mlen=mlen, blen=blen, vlen=vlen, threads=threads) + return run_and_assert( + build_dir, + asm_name, + mlen=mlen, + blen=blen, + vlen=vlen, + threads=threads, + stage_profile=stage_profile, + ) diff --git a/transactional_emulator/testbench/timing_goldens/check_timing_goldens.py b/transactional_emulator/testbench/timing_goldens/check_timing_goldens.py new file mode 100755 index 00000000..2cb74a59 --- /dev/null +++ b/transactional_emulator/testbench/timing_goldens/check_timing_goldens.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +"""Fail-on-drift checker for deterministic emulator timing goldens.""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path +from typing import Any + +from transactional_emulator.testbench.emulator_runner import run_emulator + + +REPO_ROOT = Path(__file__).resolve().parents[3] +DEFAULT_PLENA_ROOT = REPO_ROOT.parents[1] +DEFAULT_FIXTURE = Path(__file__).with_name("golden_workloads.json") +CHECK_KEYS = ("sim_latency_cycles", "hbm_bytes_read", "hbm_bytes_written") + + +def _load_fixture(path: Path) -> dict[str, Any]: + with path.open(encoding="utf-8") as f: + fixture = json.load(f) + if fixture.get("schema_version") != 1: + raise ValueError(f"Unsupported fixture schema_version in {path}") + workloads = fixture.get("workloads") + if not isinstance(workloads, list) or not workloads: + raise ValueError(f"Fixture {path} must contain a non-empty workloads list") + return fixture + + +def _selected_workloads(fixture: dict[str, Any], selected_ids: set[str] | None) -> list[dict[str, Any]]: + workloads = fixture["workloads"] + if selected_ids is None: + return workloads + selected = [workload for workload in workloads if workload.get("id") in selected_ids] + missing = selected_ids - {workload.get("id") for workload in selected} + if missing: + raise ValueError("Unknown workload id(s): " + ", ".join(sorted(missing))) + return selected + + +def _compare_metrics(workload_id: str, expected: dict[str, int], actual: dict[str, Any]) -> list[str]: + failures = [] + for key in CHECK_KEYS: + expected_value = expected.get(key) + actual_value = actual.get(key) + status = "PASS" if actual_value == expected_value else "FAIL" + print(f"{workload_id:28s} {key:20s} expected={expected_value} actual={actual_value} {status}") + if actual_value != expected_value: + failures.append( + f"{workload_id}.{key}: expected {expected_value}, got {actual_value}" + ) + return failures + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE) + parser.add_argument("--plena-root", type=Path, default=DEFAULT_PLENA_ROOT) + parser.add_argument("--id", dest="ids", action="append", help="Run only this workload id; repeatable.") + parser.add_argument("--threads", type=int, default=None, help="Override fixture thread count.") + args = parser.parse_args() + + os.environ.setdefault("CUDA_VISIBLE_DEVICES", "") + fixture = _load_fixture(args.fixture) + defaults = fixture.get("defaults", {}) + workloads = _selected_workloads(fixture, set(args.ids) if args.ids else None) + + failures: list[str] = [] + for workload in workloads: + workload_id = workload["id"] + build_dir = args.plena_root / workload["build_dir"] + settings_path = build_dir / "plena_settings.toml" + if not build_dir.exists(): + raise FileNotFoundError(f"{workload_id}: build_dir does not exist: {build_dir}") + if not settings_path.exists(): + raise FileNotFoundError(f"{workload_id}: missing settings file: {settings_path}") + + os.environ["PLENA_SETTINGS_TOML"] = str(settings_path) + threads = args.threads + if threads is None: + threads = int(workload.get("threads", defaults.get("threads", 1))) + stage_profile = bool(workload.get("stage_profile", defaults.get("stage_profile", False))) + + print(f"\n=== timing golden: {workload_id} ===") + metrics = run_emulator( + build_dir, + threads=threads, + stage_profile=stage_profile, + run_label=f"golden.{workload_id}", + ) + failures.extend(_compare_metrics(workload_id, workload["expected"], metrics)) + + if failures: + print("\nTiming golden drift detected:", file=sys.stderr) + for failure in failures: + print(f" - {failure}", file=sys.stderr) + return 1 + + print("\nAll timing goldens matched.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/transactional_emulator/testbench/timing_goldens/golden_workloads.json b/transactional_emulator/testbench/timing_goldens/golden_workloads.json new file mode 100644 index 00000000..e36614d1 --- /dev/null +++ b/transactional_emulator/testbench/timing_goldens/golden_workloads.json @@ -0,0 +1,46 @@ +{ + "schema_version": 1, + "description": "Deterministic timing goldens for PLENA transactional emulator. Expected values are simulation cycles and emulator HBM byte counters from final-source repeat gates.", + "defaults": { + "threads": 1, + "stage_profile": false + }, + "workloads": [ + { + "id": "gpt_synthetic_small", + "build_dir": "outputs/timing_validation/g3_synthetic_gpt_small/true_decoder_block", + "expected": { + "sim_latency_cycles": 147758, + "hbm_bytes_read": 475136, + "hbm_bytes_written": 16384 + } + }, + { + "id": "qwen_synthetic_small", + "build_dir": "outputs/timing_validation/g3_synthetic_qwen_small/true_decoder_block", + "expected": { + "sim_latency_cycles": 256559, + "hbm_bytes_read": 868352, + "hbm_bytes_written": 16384 + } + }, + { + "id": "gpt_real_layer0_tok1", + "build_dir": "outputs/timing_validation/g3_real_gpt_default", + "expected": { + "sim_latency_cycles": 34354344, + "hbm_bytes_read": 199096320, + "hbm_bytes_written": 0 + } + }, + { + "id": "qwen_real_decoder_block", + "build_dir": "outputs/timing_validation/g3_real_qwen_default/true_decoder_block", + "expected": { + "sim_latency_cycles": 65073075, + "hbm_bytes_read": 165019648, + "hbm_bytes_written": 262144 + } + } + ] +}