use std::sync::Mutex;
use std::sync::atomic::AtomicBool;
use vj_abi::Region;
use super::{
CompiledProcess, JoinKind, PendingActivation, ProcessId, RegionQueuePushError, RegionQueues,
ResumeArg,
};
use crate::process::WaitSpec;
pub struct ProcessDescriptor {
pub compiled: Mutex<Box<dyn CompiledProcess>>,
pub queued_this_pass: AtomicBool,
pub started: AtomicBool,
pub wait_state: Option<WaitSpec>,
pub fork_parent: Option<ProcessId>,
pub disable_names: Vec<String>,
pub fork_children: Vec<ProcessId>,
pub fork_gen: u64,
pub fork_group: u64,
pub registration_order: u64,
pub fork_join_kind: Option<JoinKind>,
pub fork_remaining: u32,
pub fork_any_signaled: bool,
pub wait_fork_remaining: Option<u32>,
pub wait_order: Option<(Vec<ProcessId>, usize)>,
pub wait_order_violated: bool,
pub region_id: u32,
pub suspended: bool,
pub parked: Option<ResumeArg>,
pub awaiting: Option<ProcessId>,
pub sync_blocked: bool,
pub waiting: bool,
pub wait_set: Option<super::EventWaitSet>,
pub desensitized: bool,
pub reactive: bool,
pub program_thread: bool,
}
impl ProcessDescriptor {
fn new(compiled: Box<dyn CompiledProcess>, registration_order: u64) -> Self {
ProcessDescriptor {
compiled: Mutex::new(compiled),
queued_this_pass: AtomicBool::new(false),
started: AtomicBool::new(false),
wait_state: None,
fork_parent: None,
disable_names: Vec::new(),
fork_children: Vec::new(),
fork_gen: 0,
fork_group: 0,
registration_order,
fork_join_kind: None,
fork_remaining: 0,
fork_any_signaled: false,
wait_fork_remaining: None,
wait_order: None,
wait_order_violated: false,
region_id: 0,
suspended: false,
parked: None,
awaiting: None,
sync_blocked: false,
waiting: false,
wait_set: None,
desensitized: false,
reactive: false,
program_thread: false,
}
}
}
slotmap::new_key_type! {
pub struct ProcessKey;
}
#[derive(Default)]
pub struct ProcessTable {
slots: slotmap::SlotMap<ProcessKey, ProcessDescriptor>,
order_to_key: Vec<ProcessKey>,
}
impl ProcessTable {
pub fn new() -> Self {
ProcessTable {
slots: slotmap::SlotMap::with_key(),
order_to_key: Vec::new(),
}
}
pub fn register(&mut self, compiled: Box<dyn CompiledProcess>) -> ProcessId {
let id = self.order_to_key.len() as ProcessId;
let key = self
.slots
.insert(ProcessDescriptor::new(compiled, id as u64));
self.order_to_key.push(key);
id
}
pub fn count(&self) -> usize {
self.order_to_key.len()
}
pub fn get(&self, id: ProcessId) -> Option<&ProcessDescriptor> {
let key = *self.order_to_key.get(id as usize)?;
self.slots.get(key)
}
pub fn get_mut(&mut self, id: ProcessId) -> Option<&mut ProcessDescriptor> {
let key = *self.order_to_key.get(id as usize)?;
self.slots.get_mut(key)
}
pub fn set_region_id(&mut self, id: ProcessId, region_id: u32) {
if let Some(desc) = self.get_mut(id) {
desc.region_id = region_id;
}
}
pub fn set_reactive(&mut self, id: ProcessId) {
if let Some(desc) = self.get_mut(id) {
desc.reactive = true;
}
}
pub fn any_live_program_child(&self) -> bool {
self.slots
.values()
.any(|d| d.program_thread && d.fork_parent.is_some())
}
pub fn is_reactive(&self, id: ProcessId) -> bool {
self.get(id).is_some_and(|d| d.reactive)
}
pub fn region_id(&self, id: ProcessId) -> u32 {
self.get(id).map(|d| d.region_id).unwrap_or(0)
}
pub fn remove(&mut self, id: ProcessId) -> Option<ProcessDescriptor> {
let key = *self.order_to_key.get(id as usize)?;
self.slots.remove(key)
}
}
#[derive(Clone, Copy, Debug)]
pub struct WorkUnit {
pub proc: ProcessId,
pub arg: ResumeArg,
}
pub fn form_ready_set(
queues: &mut RegionQueues,
region: Region,
) -> Result<Vec<PendingActivation>, RegionQueuePushError> {
match region {
Region::Active => Ok(queues.drain_active()),
Region::Inactive => Ok(queues.drain_inactive()),
Region::Reactive => Ok(queues.drain_reactive_once()),
Region::Postponed => Ok(queues.drain_postponed_once()),
Region::Nba => Err(RegionQueuePushError::NbaHasNoQueue),
Region::Preponed => Err(RegionQueuePushError::PreponedHasNoQueue),
Region::Observed => Err(RegionQueuePushError::ObservedHasNoQueue),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::sched::{ResumeArg, SchedulerOps, Suspend};
use std::sync::atomic::Ordering;
use vj_abi::SimHandle;
struct NoopProcess;
impl CompiledProcess for NoopProcess {
fn resume(
&mut self,
_state: SimHandle,
_arg: ResumeArg,
_ops: &dyn SchedulerOps,
) -> Suspend {
Suspend::Done
}
}
#[test]
fn register_assigns_ids_in_call_order() {
let mut table = ProcessTable::new();
let a = table.register(Box::new(NoopProcess));
let b = table.register(Box::new(NoopProcess));
let c = table.register(Box::new(NoopProcess));
assert_eq!([a, b, c], [0, 1, 2]);
assert_eq!(table.get(a).unwrap().registration_order, 0);
assert_eq!(table.get(b).unwrap().registration_order, 1);
assert_eq!(table.get(c).unwrap().registration_order, 2);
}
#[test]
fn get_returns_none_for_an_unregistered_id() {
let mut table = ProcessTable::new();
let only = table.register(Box::new(NoopProcess));
assert!(table.get(only).is_some());
assert!(table.get(only + 1).is_none());
}
#[test]
fn new_descriptor_starts_unqueued_not_started_and_with_no_fork_state() {
let mut table = ProcessTable::new();
let id = table.register(Box::new(NoopProcess));
let desc = table.get(id).unwrap();
assert!(!desc.queued_this_pass.load(Ordering::Relaxed));
assert!(!desc.started.load(Ordering::Relaxed));
assert!(desc.wait_state.is_none());
assert!(desc.fork_parent.is_none());
assert!(desc.fork_children.is_empty());
assert!(desc.fork_join_kind.is_none());
assert_eq!(desc.fork_remaining, 0);
assert!(!desc.fork_any_signaled);
assert!(desc.wait_fork_remaining.is_none());
assert!(desc.wait_order.is_none());
assert!(!desc.wait_order_violated);
let _guard = desc.compiled.lock().unwrap();
}
#[test]
fn slotmap_migration_preserves_ids_and_registration_order() {
let mut table = ProcessTable::new();
let ids: Vec<ProcessId> = (0..5)
.map(|_| table.register(Box::new(NoopProcess)))
.collect();
assert_eq!(ids, vec![0, 1, 2, 3, 4]);
for id in ids {
assert_eq!(
table.get(id).unwrap().registration_order,
id as u64,
"ProcessId {id} must map to the descriptor registered at that order"
);
}
}
fn scan_is_genuinely_slotmap_backed(src: &str) -> bool {
let production_only = crate::determinism::scan_support::strip_cfg_test_blocks(src);
production_only.contains("slots: slotmap::SlotMap<ProcessKey")
}
#[test]
fn process_table_is_genuinely_slotmap_backed() {
let src = std::fs::read_to_string(concat!(
env!("CARGO_MANIFEST_DIR"),
"/src/sched/process_table.rs"
))
.expect("must be able to read process_table.rs's own source");
assert!(
scan_is_genuinely_slotmap_backed(&src),
"ProcessTable's `slots` field must be genuinely SlotMap-backed (SC-063), \
not just behaviorally equivalent to one -- scanning production code \
only (test bodies stripped) for the field's actual declared type"
);
}
#[test]
fn slotmap_scan_rejects_a_reverted_vec_even_with_stale_doc_comment_mentions() {
let reverted_src = concat!(
"//! Process storage, genuinely `slotmap::SlotMap`-backed (SC-063).\n",
"pub struct ProcessTable {\n",
" slots: Vec<ProcessDescriptor>,\n",
"}\n",
"#[cfg(test)]\n",
"mod tests {\n",
" #[test]\n",
" fn t() {\n",
" assert!(\"x\".contains(\"slotmap::SlotMap\"));\n",
" }\n",
"}\n",
);
assert!(reverted_src.contains("slotmap::SlotMap"));
assert!(
!scan_is_genuinely_slotmap_backed(reverted_src),
"a source with `slots: Vec<...>` must be rejected even though the \
generic phrase \"slotmap::SlotMap\" still appears in a doc comment \
and in a #[cfg(test)] body"
);
}
#[test]
fn slotmap_scan_accepts_the_actual_current_field_declaration() {
assert!(scan_is_genuinely_slotmap_backed(
"slots: slotmap::SlotMap<ProcessKey, ProcessDescriptor>,"
));
}
#[test]
fn form_ready_set_collects_all_activations_at_one_time_region_as_one_object() {
use crate::sched::{PendingActivation, RegionQueues, ResumeArg, SchedTime};
use vj_abi::Region;
let mut queues = RegionQueues::new();
for i in 0..4u64 {
queues
.push(
Region::Active,
PendingActivation {
id: i,
proc: i as u32,
arg: ResumeArg::Start,
reconv_iter: 0,
logical_order: (SchedTime { ticks: 5 }, Region::Active, i),
},
)
.unwrap();
}
let ready = form_ready_set(&mut queues, Region::Active).unwrap();
assert_eq!(
ready.len(),
4,
"all processes ready at the same (time, region) must be collected \
into one ready-set object before dispatch"
);
}
#[test]
fn form_ready_set_rejects_nba_with_a_typed_error_not_a_silent_empty_vec() {
use crate::sched::{RegionQueuePushError, RegionQueues};
use vj_abi::Region;
let mut queues = RegionQueues::new();
let result = form_ready_set(&mut queues, Region::Nba);
assert_eq!(result.unwrap_err(), RegionQueuePushError::NbaHasNoQueue);
}
#[test]
fn form_ready_set_preponed_returns_named_error() {
use crate::sched::{RegionQueuePushError, RegionQueues};
use vj_abi::Region;
let mut queues = RegionQueues::new();
let result = form_ready_set(&mut queues, Region::Preponed);
assert_eq!(
result.unwrap_err(),
RegionQueuePushError::PreponedHasNoQueue
);
}
#[test]
fn form_ready_set_observed_returns_named_error() {
use crate::sched::{RegionQueuePushError, RegionQueues};
use vj_abi::Region;
let mut queues = RegionQueues::new();
let result = form_ready_set(&mut queues, Region::Observed);
assert_eq!(
result.unwrap_err(),
RegionQueuePushError::ObservedHasNoQueue
);
}
#[test]
fn work_unit_wraps_exactly_one_resume_invocation_no_mid_statement_split() {
use crate::sched::ResumeArg;
let unit = WorkUnit {
proc: 7,
arg: ResumeArg::Start,
};
let WorkUnit { proc, arg } = unit;
assert_eq!(proc, 7);
assert!(matches!(arg, ResumeArg::Start));
}
#[test]
fn region_id_defaults_to_zero_and_set_region_id_is_read_back_by_region_id() {
let mut table = ProcessTable::new();
let id = table.register(Box::new(NoopProcess));
assert_eq!(table.region_id(id), 0);
table.set_region_id(id, 9);
assert_eq!(
table.region_id(id),
9,
"set_region_id/region_id must round-trip a real value, not stay pinned at 0"
);
}
#[test]
fn region_id_of_an_unregistered_id_is_zero_not_a_panic() {
let table = ProcessTable::new();
assert_eq!(table.region_id(42), 0);
}
#[test]
fn remove_drops_the_descriptor_and_a_later_get_returns_none() {
let mut table = ProcessTable::new();
let a = table.register(Box::new(NoopProcess));
let b = table.register(Box::new(NoopProcess));
assert!(table.get(a).is_some());
let removed = table.remove(a);
assert!(
removed.is_some(),
"remove must return the removed descriptor, not None, for a live id"
);
assert!(
table.get(a).is_none(),
"a later get() for a removed ProcessId must return None -- the slot \
itself must be gone from the SlotMap, not just logically ignored"
);
assert!(
table.get(b).is_some(),
"removing one process must not affect an unrelated, still-live one"
);
}
#[test]
fn removing_an_already_removed_id_returns_none_not_a_panic() {
let mut table = ProcessTable::new();
let a = table.register(Box::new(NoopProcess));
assert!(table.remove(a).is_some());
assert!(table.remove(a).is_none());
}
#[test]
fn removing_an_unregistered_id_returns_none_not_a_panic() {
let mut table = ProcessTable::new();
assert!(table.remove(42).is_none());
}
}