diff --git a/src/nodes/lib/rushB/src/lib.rs b/src/nodes/lib/rushB/src/lib.rs index c72cdf4..1eafe54 100644 --- a/src/nodes/lib/rushB/src/lib.rs +++ b/src/nodes/lib/rushB/src/lib.rs @@ -1,4 +1,10 @@ -//! Shared Rust contract for the backend-neutral rushB host ABI. +//! Backend-neutral rushB protocol and runtime. + +mod protocol; +mod runtime; + +pub use protocol::{CommandId, DmaChunk, DmaOperation, RushCommand, RushEvent, RushEventKind, RushResponse, WaitMode}; +pub use runtime::{RushBackend, RushMessage, RushRequest, RushRuntime}; pub const FUNCT7_FENCE: u32 = 0; pub const FUNCT7_MVOUT: u32 = 16; diff --git a/src/nodes/lib/rushB/src/protocol.rs b/src/nodes/lib/rushB/src/protocol.rs new file mode 100644 index 0000000..74e794a --- /dev/null +++ b/src/nodes/lib/rushB/src/protocol.rs @@ -0,0 +1,52 @@ +pub type CommandId = u64; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct RushCommand { + pub id: CommandId, + pub core_id: u32, + pub xs1: u64, + pub xs2: u64, + pub funct7: u32, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum WaitMode { + Accepted, + Completed, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RushEventKind { + Accepted, + Completed, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct RushEvent { + pub command_id: CommandId, + pub core_id: u32, + pub kind: RushEventKind, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct DmaChunk { + pub offset: usize, + pub data: Vec, +} + +#[derive(Debug)] +pub enum DmaOperation { + None, + Mvin { + spans: Vec<(usize, usize)>, + chunks: Vec, + }, + Mvout { + spans: Vec<(usize, usize)>, + }, +} + +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct RushResponse { + pub output: Vec, +} diff --git a/src/nodes/lib/rushB/src/runtime.rs b/src/nodes/lib/rushB/src/runtime.rs new file mode 100644 index 0000000..c7bf6f3 --- /dev/null +++ b/src/nodes/lib/rushB/src/runtime.rs @@ -0,0 +1,442 @@ +use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::mpsc; + +use crate::{CommandId, DmaOperation, RushCommand, RushEvent, RushEventKind, RushResponse, WaitMode}; + +pub struct RushRequest { + pub command: RushCommand, + pub wait: WaitMode, + pub dma: DmaOperation, + pub response: mpsc::Sender>, +} + +pub enum RushMessage { + Command(RushRequest), + Shutdown(mpsc::Sender>), +} + +/// Backend-specific mechanics used by the backend-neutral scheduler. +pub trait RushBackend { + type Prepared; + + fn prepare(&mut self, command: &mut RushCommand, dma: DmaOperation) -> Result; + fn submit(&mut self, command: &RushCommand) -> Result<(), String>; + fn poll_event(&mut self) -> Result, String>; + fn finish(&mut self, command: &RushCommand, prepared: Self::Prepared) -> Result; + fn tick(&mut self) -> Result<(), String>; + fn cycles(&self) -> u64; + fn is_idle(&self) -> bool; + fn reset_idle_resources(&mut self); + fn diagnostics(&self, core_id: u32) -> String; + fn shutdown(&mut self) -> Result<(), String>; +} + +struct ActiveCommand

{ + request: RushRequest, + prepared: P, + started_cycle: u64, +} + +struct CompletionWait

{ + request: RushRequest, + prepared: P, + started_cycle: u64, +} + +struct CoreQueue

{ + queued: VecDeque, + active: Option>, + completion_wait: Option>, +} + +impl

Default for CoreQueue

{ + fn default() -> Self { + Self { + queued: VecDeque::new(), + active: None, + completion_wait: None, + } + } +} + +pub struct RushRuntime { + backend: B, + queues: HashMap>, + accepted_only: HashMap, + outstanding: HashSet, + max_wait_cycles: u64, + resources_idle: bool, +} + +impl RushRuntime { + pub fn new(backend: B, max_wait_cycles: u64) -> Self { + Self { + backend, + queues: HashMap::new(), + accepted_only: HashMap::new(), + outstanding: HashSet::new(), + max_wait_cycles, + resources_idle: false, + } + } + + pub fn run(mut self, receiver: mpsc::Receiver) -> Result<(), String> { + let mut shutdown = None; + loop { + self.drain_messages(&receiver, &mut shutdown)?; + self.process_events()?; + self.submit_available()?; + + let idle = self.is_drained(); + if idle && !self.resources_idle { + self.backend.reset_idle_resources(); + } + self.resources_idle = idle; + + if let Some(reply) = shutdown.take() { + if idle { + let result = self.backend.shutdown(); + let _ = reply.send(result.clone()); + return result; + } + shutdown = Some(reply); + } + + if !self.has_runtime_work() { + let message = receiver + .recv() + .map_err(|_| "rushB host disconnected without shutting down the runtime".to_string())?; + self.handle_message(message, &mut shutdown)?; + continue; + } + + self.backend.tick()?; + self.process_events()?; + self.check_timeouts()?; + } + } + + fn drain_messages( + &mut self, + receiver: &mpsc::Receiver, + shutdown: &mut Option>>, + ) -> Result<(), String> { + while let Ok(message) = receiver.try_recv() { + self.handle_message(message, shutdown)?; + } + Ok(()) + } + + fn handle_message( + &mut self, + message: RushMessage, + shutdown: &mut Option>>, + ) -> Result<(), String> { + match message { + RushMessage::Command(request) => { + if shutdown.is_some() { + let _ = request.response.send(Err("rushB runtime is shutting down".to_string())); + } else if !self.outstanding.insert(request.command.id) { + let _ = request + .response + .send(Err(format!("duplicate rushB command id {}", request.command.id))); + } else { + self.queues + .entry(request.command.core_id) + .or_default() + .queued + .push_back(request); + self.resources_idle = false; + } + } + RushMessage::Shutdown(reply) => { + if shutdown.replace(reply).is_some() { + return Err("duplicate rushB runtime shutdown request".to_string()); + } + } + } + Ok(()) + } + + fn submit_available(&mut self) -> Result<(), String> { + let core_ids = self.queues.keys().copied().collect::>(); + for core_id in core_ids { + let queue = self.queues.get_mut(&core_id).expect("Core queue exists"); + if queue.active.is_some() || queue.completion_wait.is_some() { + continue; + } + let Some(mut request) = queue.queued.pop_front() else { + continue; + }; + let dma = std::mem::replace(&mut request.dma, DmaOperation::None); + let prepared = self.backend.prepare(&mut request.command, dma)?; + self.backend.submit(&request.command)?; + queue.active = Some(ActiveCommand { + request, + prepared, + started_cycle: self.backend.cycles(), + }); + } + Ok(()) + } + + fn process_events(&mut self) -> Result<(), String> { + while let Some(event) = self.backend.poll_event()? { + match event.kind { + RushEventKind::Accepted => self.process_accepted(event)?, + RushEventKind::Completed => self.process_completed(event)?, + } + } + Ok(()) + } + + fn process_accepted(&mut self, event: RushEvent) -> Result<(), String> { + let queue = self + .queues + .get_mut(&event.core_id) + .ok_or_else(|| format!("rushB accepted command {} for an unknown Core", event.command_id))?; + let active = queue + .active + .take() + .ok_or_else(|| format!("rushB accepted command {} with no pending command", event.command_id))?; + if active.request.command.id != event.command_id { + return Err(format!( + "rushB acceptance id mismatch: expected={} actual={}", + active.request.command.id, event.command_id + )); + } + + match active.request.wait { + WaitMode::Accepted => { + let response = self.backend.finish(&active.request.command, active.prepared)?; + self.accepted_only.insert(active.request.command.id, event.core_id); + let _ = active.request.response.send(Ok(response)); + } + WaitMode::Completed => { + queue.completion_wait = Some(CompletionWait { + request: active.request, + prepared: active.prepared, + started_cycle: self.backend.cycles(), + }); + } + } + Ok(()) + } + + fn process_completed(&mut self, event: RushEvent) -> Result<(), String> { + let queue = self + .queues + .get_mut(&event.core_id) + .ok_or_else(|| format!("rushB completed command {} for an unknown Core", event.command_id))?; + let is_waiting = queue + .completion_wait + .as_ref() + .is_some_and(|waiting| waiting.request.command.id == event.command_id); + if is_waiting { + let waiting = queue.completion_wait.take().expect("matching completion waiter exists"); + let response = self.backend.finish(&waiting.request.command, waiting.prepared)?; + self.outstanding.remove(&event.command_id); + let _ = waiting.request.response.send(Ok(response)); + return Ok(()); + } + + // Accepted-only calls still produce completion events. Consume those + // events so the backend can track all in-flight commands precisely. + let core_id = self + .accepted_only + .remove(&event.command_id) + .ok_or_else(|| format!("rushB completed unknown or duplicate command {}", event.command_id))?; + if core_id != event.core_id { + return Err(format!( + "rushB completion Core mismatch for command {}", + event.command_id + )); + } + self.outstanding.remove(&event.command_id); + Ok(()) + } + + fn check_timeouts(&self) -> Result<(), String> { + let cycle = self.backend.cycles(); + for (&core_id, queue) in &self.queues { + if let Some(active) = &queue.active { + if cycle.saturating_sub(active.started_cycle) >= self.max_wait_cycles { + return Err(self.timeout_message(core_id, &active.request.command, "acceptance")); + } + } + if let Some(waiting) = &queue.completion_wait { + if cycle.saturating_sub(waiting.started_cycle) >= self.max_wait_cycles { + return Err(self.timeout_message(core_id, &waiting.request.command, "completion")); + } + } + } + Ok(()) + } + + fn timeout_message(&self, core_id: u32, command: &RushCommand, phase: &str) -> String { + format!( + "rushB runtime timed out waiting for {phase}: command={} core={} funct7={} xs1=0x{:016x} xs2=0x{:016x} {}", + command.id, + core_id, + command.funct7, + command.xs1, + command.xs2, + self.backend.diagnostics(core_id), + ) + } + + fn is_drained(&self) -> bool { + self.queues + .values() + .all(|queue| queue.queued.is_empty() && queue.active.is_none() && queue.completion_wait.is_none()) + && self.accepted_only.is_empty() + && self.backend.is_idle() + } + + fn has_runtime_work(&self) -> bool { + self.queues + .values() + .any(|queue| !queue.queued.is_empty() || queue.active.is_some() || queue.completion_wait.is_some()) + || !self.accepted_only.is_empty() + || !self.backend.is_idle() + } +} + +#[cfg(test)] +mod tests { + use std::collections::VecDeque; + use std::sync::{Arc, Mutex}; + use std::thread; + + use super::*; + + #[derive(Default)] + struct FakeState { + submitted: Vec<(CommandId, u32)>, + cycles: u64, + shutdown: bool, + } + + struct FakeBackend { + state: Arc>, + events: VecDeque, + inflight: usize, + } + + impl RushBackend for FakeBackend { + type Prepared = (); + + fn prepare(&mut self, _command: &mut RushCommand, _dma: DmaOperation) -> Result { + Ok(()) + } + + fn submit(&mut self, command: &RushCommand) -> Result<(), String> { + self.state.lock().unwrap().submitted.push((command.id, command.core_id)); + self.inflight += 1; + self.events.push_back(RushEvent { + command_id: command.id, + core_id: command.core_id, + kind: RushEventKind::Accepted, + }); + self.events.push_back(RushEvent { + command_id: command.id, + core_id: command.core_id, + kind: RushEventKind::Completed, + }); + Ok(()) + } + + fn poll_event(&mut self) -> Result, String> { + let event = self.events.pop_front(); + if event.is_some_and(|event| event.kind == RushEventKind::Completed) { + self.inflight -= 1; + } + Ok(event) + } + + fn finish(&mut self, _command: &RushCommand, _prepared: Self::Prepared) -> Result { + Ok(RushResponse::default()) + } + + fn tick(&mut self) -> Result<(), String> { + self.state.lock().unwrap().cycles += 1; + Ok(()) + } + + fn cycles(&self) -> u64 { + self.state.lock().unwrap().cycles + } + + fn is_idle(&self) -> bool { + self.inflight == 0 && self.events.is_empty() + } + + fn reset_idle_resources(&mut self) {} + + fn diagnostics(&self, _core_id: u32) -> String { + "fake-backend".to_string() + } + + fn shutdown(&mut self) -> Result<(), String> { + self.state.lock().unwrap().shutdown = true; + Ok(()) + } + } + + #[test] + fn schedules_commands_by_core() { + let state = Arc::new(Mutex::new(FakeState::default())); + let backend = FakeBackend { + state: Arc::clone(&state), + events: VecDeque::new(), + inflight: 0, + }; + let (sender, receiver) = mpsc::channel(); + let worker = thread::spawn(move || RushRuntime::new(backend, 100).run(receiver)); + + let (response, result) = mpsc::channel(); + sender + .send(RushMessage::Command(RushRequest { + command: RushCommand { + id: 7, + core_id: 65_536, + xs1: 1, + xs2: 2, + funct7: 3, + }, + wait: WaitMode::Completed, + dma: DmaOperation::None, + response, + })) + .unwrap(); + result.recv().unwrap().unwrap(); + + let (shutdown, done) = mpsc::channel(); + sender.send(RushMessage::Shutdown(shutdown)).unwrap(); + done.recv().unwrap().unwrap(); + worker.join().unwrap().unwrap(); + + let state = state.lock().unwrap(); + assert_eq!(state.submitted, vec![(7, 65_536)]); + assert!(state.shutdown); + } + + #[test] + fn rejects_unknown_completion_ids() { + let state = Arc::new(Mutex::new(FakeState::default())); + let mut events = VecDeque::new(); + events.push_back(RushEvent { + command_id: 99, + core_id: 0, + kind: RushEventKind::Completed, + }); + let backend = FakeBackend { + state, + events, + inflight: 1, + }; + let (_sender, receiver) = mpsc::channel(); + let error = RushRuntime::new(backend, 100).run(receiver).unwrap_err(); + assert!(error.contains("unknown Core") || error.contains("unknown or duplicate command 99")); + } +} diff --git a/src/nodes/verilator/src/rushb/command.rs b/src/nodes/verilator/src/rushb/command.rs index 1e66f99..0ccee98 100644 --- a/src/nodes/verilator/src/rushb/command.rs +++ b/src/nodes/verilator/src/rushb/command.rs @@ -1,33 +1,7 @@ -use super::dma::{DmaChunk, DmaOperation}; use super::state; +use bebop_rushb::{DmaOperation, RushCommand, RushMessage, RushRequest, RushResponse, WaitMode}; use std::sync::mpsc; -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub(crate) enum WaitMode { - Accepted, - Completed, -} - -pub(crate) struct CommandResponse { - pub(crate) output: Vec, -} - -pub(crate) struct CommandRequest { - pub(crate) command_id: u64, - pub(crate) core_id: u32, - pub(crate) xs1: u64, - pub(crate) xs2: u64, - pub(crate) funct7: u32, - pub(crate) wait: WaitMode, - pub(crate) dma: DmaOperation, - pub(crate) response: mpsc::Sender>, -} - -pub(crate) enum SchedulerMessage { - Command(CommandRequest), - Shutdown(mpsc::Sender>), -} - pub(crate) fn execute( core_id: u32, xs1: u64, @@ -35,20 +9,22 @@ pub(crate) fn execute( funct7: u32, wait: WaitMode, dma: DmaOperation, -) -> Result { +) -> Result { let command_id = state::next_command_id(); let (response, receiver) = mpsc::channel(); - let request = CommandRequest { - command_id, - core_id, - xs1, - xs2, - funct7, + let request = RushRequest { + command: RushCommand { + id: command_id, + core_id, + xs1, + xs2, + funct7, + }, wait, dma, response, }; - state::send(SchedulerMessage::Command(request))?; + state::send(RushMessage::Command(request))?; receiver .recv() .map_err(|_| format!("rushB NPU scheduler stopped while waiting for host command #{command_id}"))? diff --git a/src/nodes/verilator/src/rushb/dma.rs b/src/nodes/verilator/src/rushb/dma.rs index 8744147..02f66d4 100644 --- a/src/nodes/verilator/src/rushb/dma.rs +++ b/src/nodes/verilator/src/rushb/dma.rs @@ -1,25 +1,9 @@ use super::state::BankConfig; use crate::ffi::{bbsim_host_memory_range, bbsim_host_memory_read, bbsim_host_memory_write}; +use bebop_rushb::DmaChunk; const DMA_ADDR_MASK: u64 = (1_u64 << 39) - 1; const CHIP_ID: i32 = 0; -#[derive(Debug)] -pub(crate) struct DmaChunk { - pub(crate) offset: usize, - pub(crate) data: Vec, -} - -pub(crate) enum DmaOperation { - None, - Mvin { - spans: Vec<(usize, usize)>, - chunks: Vec, - }, - Mvout { - spans: Vec<(usize, usize)>, - }, -} - pub(crate) struct PreparedDma { pub(crate) address: u64, pub(crate) spans: Vec<(usize, usize)>, diff --git a/src/nodes/verilator/src/rushb/mod.rs b/src/nodes/verilator/src/rushb/mod.rs index 444b8c1..22c971d 100644 --- a/src/nodes/verilator/src/rushb/mod.rs +++ b/src/nodes/verilator/src/rushb/mod.rs @@ -3,9 +3,7 @@ mod dma; mod scheduler; mod state; -use bebop_rushb::{FUNCT7_MSET, FUNCT7_MVIN, FUNCT7_MVIN_MMIO, FUNCT7_MVOUT}; -use command::WaitMode; -use dma::DmaOperation; +use bebop_rushb::{DmaOperation, WaitMode, FUNCT7_MSET, FUNCT7_MVIN, FUNCT7_MVIN_MMIO, FUNCT7_MVOUT}; use std::ffi::c_void; fn mvin_mmio_spans(rows: u64, columns: u64) -> (Vec<(usize, usize)>, Vec<(usize, usize)>) { diff --git a/src/nodes/verilator/src/rushb/scheduler.rs b/src/nodes/verilator/src/rushb/scheduler.rs index 5088537..89db0e5 100644 --- a/src/nodes/verilator/src/rushb/scheduler.rs +++ b/src/nodes/verilator/src/rushb/scheduler.rs @@ -1,12 +1,14 @@ -use super::command::{CommandRequest, CommandResponse, SchedulerMessage, WaitMode}; -use super::dma::{self, DmaOperation, PreparedDma, StagingAllocator}; +use super::dma::{self, PreparedDma, StagingAllocator}; use crate::ffi::{ verilator_context_time, verilator_rushb_accepted, verilator_rushb_clear, verilator_rushb_complete_on_accept, verilator_rushb_completed, verilator_rushb_inflight, verilator_rushb_last_ready, verilator_rushb_last_retired, verilator_rushb_probes, verilator_rushb_submit, }; use crate::Simulator; -use bebop_rushb::FUNCT7_FENCE; +use bebop_rushb::{ + CommandId, DmaOperation, RushBackend, RushCommand, RushEvent, RushEventKind, RushMessage, RushResponse, + RushRuntime, FUNCT7_FENCE, +}; use std::collections::{HashMap, VecDeque}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{mpsc, Arc}; @@ -14,303 +16,252 @@ use std::sync::{mpsc, Arc}; const POST_RESET_SETTLE_CYCLES: u64 = 4_096; const MAX_WAIT_CYCLES: u64 = 100_000_000; -#[derive(Default)] -struct CoreQueue { - queued: VecDeque, - active: Option, - completion_wait: Option, -} - -struct ActiveCommand { - request: CommandRequest, +struct PendingCommand { + id: CommandId, accepted_before: u64, - prepared_dma: Option, - started_cycle: u64, + fence: bool, } -struct CompletionWait { - request: CommandRequest, - target_completed: u64, - prepared_dma: Option, - started_cycle: u64, +struct ChannelState { + pending: Option, + inflight: VecDeque, + accepted_seen: u64, + completed_seen: u64, } -pub(crate) fn run( - receiver: mpsc::Receiver, +struct VerilatorBackend { + simulator: Simulator, + staging: StagingAllocator, cycles: Arc, - ready: mpsc::Sender>, -) -> Result<(), String> { - unsafe { verilator_rushb_clear() }; - let mut simulator = match Simulator::new(None, &[]) { - Ok(simulator) => simulator, - Err(error) => { - let message = format!("failed to create rushB Verilator simulator: {error}"); - let _ = ready.send(Err(message.clone())); - return Err(message); - } - }; - for _ in 0..POST_RESET_SETTLE_CYCLES { - simulator.exec_once(); - } - update_cycles(&simulator, &cycles); - ready - .send(Ok(())) - .map_err(|_| "rushB host disappeared during scheduler initialization".to_string())?; - - let mut queues = HashMap::::new(); - let mut staging = StagingAllocator::default(); - let mut shutdown = None; - - loop { - drain_messages(&receiver, &mut queues, &mut shutdown)?; - process_completions(&mut queues, current_cycle(&simulator))?; - - if all_inflight_zero(&queues) && queues.values().all(|queue| queue.active.is_none()) { - staging.reset(); - } - submit_available(&mut queues, &mut staging, current_cycle(&simulator))?; - - if let Some(reply) = shutdown.take() { - if is_drained(&queues) { - simulator.finalize(); - unsafe { verilator_rushb_clear() }; - let _ = reply.send(Ok(())); - return Ok(()); - } - shutdown = Some(reply); - } + channels: HashMap, + events: VecDeque, +} - if !has_runtime_work(&queues) { - let message = receiver - .recv() - .map_err(|_| "rushB host disconnected without shutting down the NPU scheduler".to_string())?; - handle_message(message, &mut queues, &mut shutdown)?; - continue; +impl VerilatorBackend { + fn new(cycles: Arc) -> Result { + unsafe { verilator_rushb_clear() }; + let mut simulator = Simulator::new(None, &[]) + .map_err(|error| format!("failed to create rushB Verilator simulator: {error}"))?; + // Reset must reach BBSimDRAM before it can allocate its DPI backing store. + for _ in 0..POST_RESET_SETTLE_CYCLES { + simulator.exec_once(); } + let backend = Self { + simulator, + staging: StagingAllocator::default(), + cycles, + channels: HashMap::new(), + events: VecDeque::new(), + }; + backend.update_cycles(); + Ok(backend) + } - simulator.exec_once(); - update_cycles(&simulator, &cycles); - let cycle = current_cycle(&simulator); - process_accepts(&mut queues, cycle)?; - process_completions(&mut queues, cycle)?; - check_timeouts(&queues, cycle)?; + fn update_cycles(&self) { + self.cycles.store(self.current_cycle(), Ordering::Relaxed); } -} -fn drain_messages( - receiver: &mpsc::Receiver, - queues: &mut HashMap, - shutdown: &mut Option>>, -) -> Result<(), String> { - while let Ok(message) = receiver.try_recv() { - handle_message(message, queues, shutdown)?; + fn current_cycle(&self) -> u64 { + unsafe { verilator_context_time(self.simulator.context_for_rushb()) / 2 } } - Ok(()) -} -fn handle_message( - message: SchedulerMessage, - queues: &mut HashMap, - shutdown: &mut Option>>, -) -> Result<(), String> { - match message { - SchedulerMessage::Command(request) => { - if shutdown.is_some() { - let _ = request - .response - .send(Err("rushB NPU scheduler is shutting down".to_string())); - } else { - queues.entry(request.core_id).or_default().queued.push_back(request); + fn poll_hardware(&mut self) -> Result<(), String> { + let core_ids = self.channels.keys().copied().collect::>(); + for core_id in core_ids { + let channel = self.channels.get_mut(&core_id).expect("rushB channel exists"); + let accepted = unsafe { verilator_rushb_accepted(core_id) }; + if accepted < channel.accepted_seen { + return Err(format!( + "rushB accepted counter moved backwards for Core {core_id}: before={} after={accepted}", + channel.accepted_seen + )); } - } - SchedulerMessage::Shutdown(reply) => { - if shutdown.replace(reply).is_some() { - return Err("duplicate rushB NPU scheduler shutdown request".to_string()); + if accepted > channel.accepted_seen { + if accepted != channel.accepted_seen + 1 { + return Err(format!( + "rushB accepted counter skipped for Core {core_id}: before={} after={accepted}", + channel.accepted_seen + )); + } + let pending = channel + .pending + .take() + .ok_or_else(|| format!("rushB Core {core_id} accepted a command that was not submitted"))?; + if pending.accepted_before != channel.accepted_seen { + return Err(format!( + "rushB acceptance baseline changed for command {} on Core {core_id}", + pending.id + )); + } + channel.accepted_seen = accepted; + channel.inflight.push_back(pending.id); + self.events.push_back(RushEvent { + command_id: pending.id, + core_id, + kind: RushEventKind::Accepted, + }); + if pending.fence { + unsafe { verilator_rushb_complete_on_accept(core_id) }; + } + } + + let completed = unsafe { verilator_rushb_completed(core_id) }; + if completed < channel.completed_seen { + return Err(format!( + "rushB completed counter moved backwards for Core {core_id}: before={} after={completed}", + channel.completed_seen + )); + } + while channel.completed_seen < completed { + let command_id = channel + .inflight + .pop_front() + .ok_or_else(|| format!("rushB Core {core_id} completed a command that was not in flight"))?; + channel.completed_seen += 1; + self.events.push_back(RushEvent { + command_id, + core_id, + kind: RushEventKind::Completed, + }); } } + Ok(()) } - Ok(()) } -fn submit_available( - queues: &mut HashMap, - staging: &mut StagingAllocator, - cycle: u64, -) -> Result<(), String> { - let core_ids = queues.keys().copied().collect::>(); - for core_id in core_ids { - let queue = queues.get_mut(&core_id).expect("Core queue exists"); - if queue.active.is_some() || queue.completion_wait.is_some() { - continue; - } - let Some(mut request) = queue.queued.pop_front() else { - continue; - }; +impl RushBackend for VerilatorBackend { + type Prepared = Option; - let dma_operation = std::mem::replace(&mut request.dma, DmaOperation::None); - let prepared_dma_result = match dma_operation { + fn prepare(&mut self, command: &mut RushCommand, dma_operation: DmaOperation) -> Result { + match dma_operation { DmaOperation::None => Ok(None), - DmaOperation::Mvin { spans, chunks } => staging.allocate(&spans).and_then(|address| { + DmaOperation::Mvin { spans, chunks } => { + let address = self.staging.allocate(&spans)?; dma::write_staging(address, &chunks)?; - request.xs2 = dma::staged_xs2(request.xs2, address); + command.xs2 = dma::staged_xs2(command.xs2, address); Ok(Some(PreparedDma { address, spans, output: false, })) - }), - DmaOperation::Mvout { spans } => staging.allocate(&spans).map(|address| { - request.xs2 = dma::staged_xs2(request.xs2, address); - Some(PreparedDma { + } + DmaOperation::Mvout { spans } => { + let address = self.staging.allocate(&spans)?; + command.xs2 = dma::staged_xs2(command.xs2, address); + Ok(Some(PreparedDma { address, spans, output: true, - }) - }), - }; - let prepared_dma = match prepared_dma_result { - Ok(prepared) => prepared, - Err(error) => { - let message = format!( - "failed to prepare rushB DMA for command {}: {error}", - request.command_id - ); - let _ = request.response.send(Err(message.clone())); - return Err(message); + })) } - }; - - let accepted_before = unsafe { verilator_rushb_accepted(core_id) }; - unsafe { verilator_rushb_submit(core_id, request.xs1, request.xs2, request.funct7) }; - queue.active = Some(ActiveCommand { - request, - accepted_before, - prepared_dma, - started_cycle: cycle, - }); + } } - Ok(()) -} -fn process_accepts(queues: &mut HashMap, cycle: u64) -> Result<(), String> { - let core_ids = queues.keys().copied().collect::>(); - for core_id in core_ids { - let queue = queues.get_mut(&core_id).expect("Core queue exists"); - let Some(active) = queue.active.as_ref() else { - continue; - }; + fn submit(&mut self, command: &RushCommand) -> Result<(), String> { + let core_id = command.core_id; let accepted = unsafe { verilator_rushb_accepted(core_id) }; - if accepted == active.accepted_before { - continue; + let completed = unsafe { verilator_rushb_completed(core_id) }; + let channel = self.channels.entry(core_id).or_insert_with(|| ChannelState { + pending: None, + inflight: VecDeque::new(), + accepted_seen: accepted, + completed_seen: completed, + }); + if channel.pending.is_some() { + return Err(format!( + "rushB submitted command {} while Core {core_id} still has a pending command", + command.id + )); } - if accepted != active.accepted_before + 1 { + if accepted != channel.accepted_seen || completed != channel.completed_seen { return Err(format!( - "rushB accepted counter skipped for Core {core_id}: before={} after={accepted}", - active.accepted_before + "rushB counters changed before command {} was submitted to Core {core_id}", + command.id )); } + channel.pending = Some(PendingCommand { + id: command.id, + accepted_before: accepted, + fence: command.funct7 == FUNCT7_FENCE, + }); + unsafe { verilator_rushb_submit(core_id, command.xs1, command.xs2, command.funct7) }; + Ok(()) + } - let active = queue.active.take().expect("active command exists"); - if active.request.funct7 == FUNCT7_FENCE { - unsafe { verilator_rushb_complete_on_accept(core_id) }; - } - match active.request.wait { - WaitMode::Accepted => { - let _ = active.request.response.send(Ok(CommandResponse { output: Vec::new() })); - } - WaitMode::Completed => { - queue.completion_wait = Some(CompletionWait { - request: active.request, - target_completed: accepted, - prepared_dma: active.prepared_dma, - started_cycle: cycle, - }); - } + fn poll_event(&mut self) -> Result, String> { + if let Some(event) = self.events.pop_front() { + return Ok(Some(event)); } + self.poll_hardware()?; + Ok(self.events.pop_front()) } - Ok(()) -} -fn process_completions(queues: &mut HashMap, _cycle: u64) -> Result<(), String> { - let core_ids = queues.keys().copied().collect::>(); - for core_id in core_ids { - let queue = queues.get_mut(&core_id).expect("Core queue exists"); - let Some(waiting) = queue.completion_wait.as_ref() else { - continue; - }; - let completed = unsafe { verilator_rushb_completed(core_id) }; - if completed < waiting.target_completed { - continue; - } - let waiting = queue.completion_wait.take().expect("completion waiter exists"); - let output = match waiting.prepared_dma { + fn finish(&mut self, _command: &RushCommand, prepared: Self::Prepared) -> Result { + let output = match prepared { Some(prepared) if prepared.output => dma::read_staging(prepared.address, &prepared.spans)?, _ => Vec::new(), }; - let _ = waiting.request.response.send(Ok(CommandResponse { output })); + Ok(RushResponse { output }) } - Ok(()) -} -fn check_timeouts(queues: &HashMap, cycle: u64) -> Result<(), String> { - for (&core_id, queue) in queues { - if let Some(active) = &queue.active { - if cycle.saturating_sub(active.started_cycle) >= MAX_WAIT_CYCLES { - return Err(timeout_message(core_id, &active.request, "acceptance")); - } - } - if let Some(waiting) = &queue.completion_wait { - if cycle.saturating_sub(waiting.started_cycle) >= MAX_WAIT_CYCLES { - return Err(timeout_message(core_id, &waiting.request, "completion")); - } - } + fn tick(&mut self) -> Result<(), String> { + self.simulator.exec_once(); + self.update_cycles(); + Ok(()) } - Ok(()) -} -fn timeout_message(core_id: u32, request: &CommandRequest, phase: &str) -> String { - unsafe { - format!( - "rushB NPU scheduler timed out waiting for {phase}: command={} core={} funct7={} xs1=0x{:016x} xs2=0x{:016x} probes={} accepted={} completed={} inflight={} ready={} retired={}", - request.command_id, - core_id, - request.funct7, - request.xs1, - request.xs2, - verilator_rushb_probes(core_id), - verilator_rushb_accepted(core_id), - verilator_rushb_completed(core_id), - verilator_rushb_inflight(core_id), - verilator_rushb_last_ready(core_id), - verilator_rushb_last_retired(core_id), - ) + fn cycles(&self) -> u64 { + self.current_cycle() } -} -fn all_inflight_zero(queues: &HashMap) -> bool { - queues - .keys() - .all(|&core_id| unsafe { verilator_rushb_inflight(core_id) == 0 }) -} + fn is_idle(&self) -> bool { + self.events.is_empty() + && self.channels.iter().all(|(&core_id, channel)| { + channel.pending.is_none() + && channel.inflight.is_empty() + && unsafe { verilator_rushb_inflight(core_id) == 0 } + }) + } -fn is_drained(queues: &HashMap) -> bool { - queues - .values() - .all(|queue| queue.queued.is_empty() && queue.active.is_none() && queue.completion_wait.is_none()) - && all_inflight_zero(queues) -} + fn reset_idle_resources(&mut self) { + self.staging.reset(); + } -fn has_runtime_work(queues: &HashMap) -> bool { - queues - .values() - .any(|queue| !queue.queued.is_empty() || queue.active.is_some() || queue.completion_wait.is_some()) - || !all_inflight_zero(queues) -} + fn diagnostics(&self, core_id: u32) -> String { + unsafe { + format!( + "probes={} accepted={} completed={} inflight={} ready={} retired={}", + verilator_rushb_probes(core_id), + verilator_rushb_accepted(core_id), + verilator_rushb_completed(core_id), + verilator_rushb_inflight(core_id), + verilator_rushb_last_ready(core_id), + verilator_rushb_last_retired(core_id), + ) + } + } -fn current_cycle(simulator: &Simulator) -> u64 { - unsafe { verilator_context_time(simulator.context_for_rushb()) / 2 } + fn shutdown(&mut self) -> Result<(), String> { + self.simulator.finalize(); + unsafe { verilator_rushb_clear() }; + Ok(()) + } } -fn update_cycles(simulator: &Simulator, cycles: &AtomicU64) { - cycles.store(current_cycle(simulator), Ordering::Relaxed); +pub(crate) fn run( + receiver: mpsc::Receiver, + cycles: Arc, + ready: mpsc::Sender>, +) -> Result<(), String> { + let backend = match VerilatorBackend::new(cycles) { + Ok(backend) => backend, + Err(error) => { + let _ = ready.send(Err(error.clone())); + return Err(error); + } + }; + ready + .send(Ok(())) + .map_err(|_| "rushB host disappeared during runtime initialization".to_string())?; + RushRuntime::new(backend, MAX_WAIT_CYCLES).run(receiver) } diff --git a/src/nodes/verilator/src/rushb/state.rs b/src/nodes/verilator/src/rushb/state.rs index eb99272..166ea8d 100644 --- a/src/nodes/verilator/src/rushb/state.rs +++ b/src/nodes/verilator/src/rushb/state.rs @@ -1,5 +1,5 @@ -use super::command::SchedulerMessage; use super::scheduler; +use bebop_rushb::RushMessage; use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{mpsc, Arc, Mutex}; @@ -12,7 +12,7 @@ pub(crate) struct BankConfig { } struct SchedulerHandle { - sender: mpsc::Sender, + sender: mpsc::Sender, cycles: Arc, worker: JoinHandle>, } @@ -68,7 +68,7 @@ pub(crate) fn destroy() { let (reply, receiver) = mpsc::channel(); handle .sender - .send(SchedulerMessage::Shutdown(reply)) + .send(RushMessage::Shutdown(reply)) .expect("rushB NPU scheduler stopped before shutdown"); receiver .recv() @@ -82,7 +82,7 @@ pub(crate) fn destroy() { *BANK_CONFIGS.lock().expect("rushB bank metadata poisoned") = None; } -pub(crate) fn send(message: SchedulerMessage) -> Result<(), String> { +pub(crate) fn send(message: RushMessage) -> Result<(), String> { let sender = SCHEDULER .lock() .map_err(|_| "rushB scheduler state poisoned".to_string())?