use vj_abi::Region;
use crate::sched::{PendingActivation, ProcessId, SchedTime};
use crate::state::{FourStateValue, StateCellId};
mod kind_guard;
mod policy;
#[cfg(test)]
pub(crate) mod scan_support;
mod sequential;
mod tiebreak;
pub use kind_guard::*;
pub use policy::*;
pub use sequential::SequentialController;
pub use tiebreak::committed_effect_tiebreak;
pub use crate::sched::ActivationId;
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)]
pub struct CanonicalEventKey {
pub vtime: SchedTime,
pub region_ordinal: u32,
pub reconv_iter: u32,
pub canonical_process_id: ProcessId,
pub local_send_seq: u64,
}
#[derive(Clone, Copy, Debug)]
pub struct WorkItem {
pub activation: PendingActivation,
pub key: CanonicalEventKey,
}
#[derive(Clone, Debug, Default)]
pub struct ResolvedBatch {
pub completed: Vec<ActivationId>,
}
pub trait DeterminismController: Send + Sync {
fn partition(
&self,
ready: &[PendingActivation],
region: Region,
now: SchedTime,
) -> Vec<WorkItem>;
fn on_read(&self, act: ActivationId, cell: StateCellId, version: u64);
fn on_write(&self, act: ActivationId, cell: StateCellId, prior: FourStateValue, version: u64);
fn gvt(&self) -> SchedTime;
fn drain_and_resolve(&self, work: Vec<WorkItem>) -> ResolvedBatch;
fn kind(&self) -> &'static str;
}
#[cfg(test)]
mod tests {
use super::*;
use crate::sched::ResumeArg;
use vj_abi::Region;
#[test]
fn work_item_activation_id_correlates_back_to_the_activation() {
let now = SchedTime { ticks: 5 };
let activation = PendingActivation {
id: 99,
proc: 1,
arg: ResumeArg::Start,
reconv_iter: 0,
logical_order: (now, Region::Active, 0),
};
let key = CanonicalEventKey {
vtime: now,
region_ordinal: Region::Active as u32,
reconv_iter: activation.reconv_iter,
canonical_process_id: activation.proc,
local_send_seq: 0,
};
let item = WorkItem { activation, key };
let resolved = ResolvedBatch {
completed: vec![item.activation.id],
};
assert_eq!(resolved.completed[0], item.activation.id);
assert_eq!(item.key.reconv_iter, item.activation.reconv_iter);
}
#[test]
fn canonical_event_key_is_a_plain_totally_ordered_tuple() {
let k1 = CanonicalEventKey {
vtime: SchedTime { ticks: 0 },
region_ordinal: Region::Active as u32,
reconv_iter: 0,
canonical_process_id: 0,
local_send_seq: 0,
};
let k2 = k1;
assert_eq!(k1, k2);
assert_eq!(k1.region_ordinal, 0);
}
#[test]
fn determinism_controller_trait_uses_sched_time_for_now_and_gvt() {
struct NoopController {
gvt: SchedTime,
}
impl DeterminismController for NoopController {
fn partition(
&self,
_ready: &[PendingActivation],
_region: Region,
_now: SchedTime,
) -> Vec<WorkItem> {
vec![]
}
fn on_read(&self, _act: ActivationId, _cell: StateCellId, _version: u64) {}
fn on_write(
&self,
_act: ActivationId,
_cell: StateCellId,
_prior: FourStateValue,
_version: u64,
) {
}
fn gvt(&self) -> SchedTime {
self.gvt
}
fn drain_and_resolve(&self, _work: Vec<WorkItem>) -> ResolvedBatch {
ResolvedBatch::default()
}
fn kind(&self) -> &'static str {
"noop"
}
}
let controller = NoopController {
gvt: SchedTime { ticks: 7 },
};
let out = controller.partition(&[], Region::Active, SchedTime { ticks: 7 });
assert!(out.is_empty());
assert_eq!(controller.gvt(), SchedTime { ticks: 7 });
}
#[test]
fn hypothetical_levelized_controller_compiles_against_the_unmodified_trait() {
struct FakeLevelizedController;
impl DeterminismController for FakeLevelizedController {
fn partition(
&self,
ready: &[PendingActivation],
region: Region,
_now: SchedTime,
) -> Vec<WorkItem> {
let mut sorted: Vec<PendingActivation> = ready.to_vec();
sorted.sort_by_key(|a| a.proc % 3);
sorted
.into_iter()
.enumerate()
.map(|(i, activation)| {
let key = CanonicalEventKey {
vtime: activation.logical_order.0,
region_ordinal: crate::sched::region_ordinal(region),
reconv_iter: activation.reconv_iter,
canonical_process_id: activation.proc,
local_send_seq: i as u64,
};
WorkItem { activation, key }
})
.collect()
}
fn on_read(&self, _act: ActivationId, _cell: StateCellId, _version: u64) {}
fn on_write(
&self,
_act: ActivationId,
_cell: StateCellId,
_prior: FourStateValue,
_version: u64,
) {
}
fn gvt(&self) -> SchedTime {
SchedTime { ticks: 0 }
}
fn drain_and_resolve(&self, work: Vec<WorkItem>) -> ResolvedBatch {
ResolvedBatch {
completed: work.iter().map(|w| w.activation.id).collect(),
}
}
fn kind(&self) -> &'static str {
"fake-levelized"
}
}
struct NoopArena;
impl crate::state::SignalAccess for NoopArena {
fn read(&self, _cell: StateCellId) -> FourStateValue {
FourStateValue::Narrow(0)
}
fn write(&mut self, _cell: StateCellId, _value: FourStateValue) -> bool {
true
}
fn subscribe(
&mut self,
_cell: StateCellId,
_role: crate::state::ListenerRole,
_sink: Box<dyn crate::state::ValueChangeSink + Send>,
) {
}
}
struct NoopIoSink;
impl crate::sched::IoSink for NoopIoSink {
fn enqueue(
&self,
_at: SchedTime,
_region: Region,
_seq: u64,
cb: crate::sched::DeferredIoCallback,
) {
drop(cb);
}
}
use crate::sched::SchedulerControl;
let mut engine = crate::sched::Engine::new(Box::new(NoopArena), Box::new(NoopIoSink));
assert_eq!(
engine.controller_kind(),
"sequential",
"engine must still default to SequentialController before this test installs \
the hypothetical levelized one"
);
engine.set_determinism_controller(Box::new(FakeLevelizedController));
assert_eq!(
engine.controller_kind(),
"fake-levelized",
"set_determinism_controller must have genuinely swapped the installed \
controller -- this is the round-trip proof, not just a compile check"
);
}
#[test]
fn notify_change_has_exactly_three_call_sites_in_the_crate() {
let sched_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src/sched");
let mut hits: Vec<(String, usize)> = Vec::new();
scan_dir_for_notify_change_calls(&sched_dir, &mut hits);
assert!(
!hits.is_empty(),
"scan matched no files under {sched_dir:?} -- the scan itself is broken"
);
let total: usize = hits.iter().map(|(_, n)| n).sum();
assert_eq!(
total, 3,
"expected exactly three production `.notify_change(` call sites crate-wide \
(SC-042 + B-sched Task 6's SC-069 addition); found: {hits:?}"
);
let count_for = |suffix: &str| -> usize {
hits.iter()
.filter(|(f, _)| f.ends_with(suffix))
.map(|(_, n)| *n)
.sum()
};
assert_eq!(
count_for("notify.rs"),
1,
"notify.rs (Task 7's commit_nba) must be exactly one of the three sanctioned \
call sites: {hits:?}"
);
assert_eq!(
count_for("ops.rs"),
1,
"ops.rs (Task 8's commit_blocking_write) must be exactly one of the three \
sanctioned call sites: {hits:?}"
);
assert_eq!(
count_for("control.rs"),
1,
"control.rs (B-sched Task 6's advance_time_to, SC-069's TriggerEvent fan-out) \
must be exactly one of the three sanctioned call sites: {hits:?}"
);
for (file, n) in &hits {
if !file.ends_with("notify.rs")
&& !file.ends_with("ops.rs")
&& !file.ends_with("control.rs")
{
assert_eq!(
*n, 0,
"unexpected production `.notify_change(` call site outside \
notify.rs/ops.rs/control.rs: {file} (SC-042: a fourth ad-hoc caller is a \
design defect) -- full scan: {hits:?}"
);
}
}
}
fn strip_line_comments(src: &str) -> String {
src.lines()
.filter(|line| !line.trim_start().starts_with("//"))
.collect::<Vec<_>>()
.join("\n")
}
fn scan_dir_for_notify_change_calls(dir: &std::path::Path, hits: &mut Vec<(String, usize)>) {
for entry in std::fs::read_dir(dir).expect("read sched/ dir for notify_change scan") {
let entry = entry.expect("dir entry for notify_change scan");
let path = entry.path();
if path.is_dir() {
scan_dir_for_notify_change_calls(&path, hits);
continue;
}
if path.extension().is_none_or(|e| e != "rs") {
continue;
}
let src = std::fs::read_to_string(&path)
.unwrap_or_else(|e| panic!("read {path:?} for notify_change scan: {e}"));
let production_only = strip_line_comments(&scan_support::strip_cfg_test_blocks(&src));
let count = production_only.matches(".notify_change(").count();
hits.push((path.display().to_string(), count));
}
}
}