Add deterministic emulator timing and stage profiling

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<dyn ErasedMemoryModel>,
     total_len: usize,
     reads: Vec<ChunkRead>,
 ) -> Vec<u8> {
     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<Vec<u64>>,
+    }
+
+    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<dyn ErasedMemoryModel> = 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| {

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;
     }
 }

transactional_emulator/lib/runtime/src/executor.rs

@@ -20,24 +20,25 @@ impl Future for ResolveAt {
 struct Timer {
     executor: Executor,
     resolve_at: Instant,
+    sequence_id: u64,
     waker: Option<Waker>,
     _phantom: PhantomPinned,
 }
 
 impl PartialEq for Timer {
     fn eq(&self, other: &Self) -> bool {
-        core::ptr::eq(self, other)
+        self.sequence_id == other.sequence_id
     }
 }
 
 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<Arc<Task>>,
     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::<Vec<_>>());
+        assert_eq!(ex.now(), at(7));
+    }
+
     #[tokio::test]
     async fn test_timeout_excludes_event_at_deadline() {
         let ex = Executor::new();

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<Duration> 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]

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,
+    }
+}