use crate::sched::ProcessId;
use crate::state::{FourStateValue, StateCellId};
#[derive(Clone, Debug)]
pub struct NbaUpdate {
pub cell: StateCellId,
pub value: FourStateValue,
pub source: ProcessId,
pub order: u32,
}
#[derive(Debug, Default)]
pub struct NbaLog {
entries: Vec<NbaUpdate>,
masked: Vec<MaskedNbaUpdate>,
next_order: u32,
}
#[derive(Clone, Debug)]
pub struct MaskedNbaUpdate {
pub cell: StateCellId,
pub row: Option<u64>,
pub value: FourStateValue,
pub mask: Vec<u64>,
pub window: Option<u32>,
pub source: ProcessId,
pub order: u32,
pub pulse: bool,
pub coll: Option<CollWrite>,
pub obj: Option<ObjWrite>,
}
#[derive(Clone, Debug)]
pub struct ObjWrite {
pub h: u64,
pub offset: usize,
pub val: Vec<u64>,
pub mask: Vec<u64>,
pub class_event: bool,
}
impl ObjWrite {
pub fn new(h: u64, offset: usize, val: Vec<u64>, mask: Vec<u64>) -> Self {
crate::objects::retain(h);
Self {
h,
offset,
val,
mask,
class_event: false,
}
}
pub fn class_event(id: u64) -> Self {
Self {
h: id,
offset: 0,
val: Vec::new(),
mask: Vec::new(),
class_event: true,
}
}
pub fn commit(self) {
debug_assert!(
!self.class_event,
"a class-event trigger commits via commit_nba"
);
crate::objects::field_merge(self.h, self.offset, &self.val, &self.mask);
crate::objects::release(self.h);
}
}
#[derive(Clone, Debug)]
pub struct CollWrite {
pub shape: crate::collections::Shape,
pub index: i64,
pub key: Option<crate::collections::Key>,
pub queue: bool,
}
impl NbaLog {
pub fn new() -> Self {
Self::default()
}
fn bump_order(&mut self) -> u32 {
let order = self.next_order;
self.next_order += 1;
order
}
pub fn stage(&mut self, cell: StateCellId, value: FourStateValue, source: ProcessId) {
let order = self.bump_order();
self.entries.push(NbaUpdate {
cell,
value,
source,
order,
});
}
pub fn stage_masked(
&mut self,
cell: StateCellId,
row: Option<u64>,
value: FourStateValue,
mask: Vec<u64>,
window: Option<u32>,
source: ProcessId,
) {
let order = self.bump_order();
self.masked.push(MaskedNbaUpdate {
cell,
row,
value,
mask,
window,
source,
order,
pulse: false,
coll: None,
obj: None,
});
}
pub fn stage_coll(
&mut self,
cell: StateCellId,
row: Option<u64>,
elem: Vec<u64>,
coll: CollWrite,
source: ProcessId,
) {
let order = self.bump_order();
self.masked.push(MaskedNbaUpdate {
cell,
row,
value: FourStateValue::Narrow(0),
mask: elem,
window: None,
source,
order,
pulse: false,
coll: Some(coll),
obj: None,
});
}
pub fn stage_obj(&mut self, w: ObjWrite, source: ProcessId) {
let order = self.bump_order();
self.masked.push(MaskedNbaUpdate {
cell: StateCellId(0),
row: None,
value: FourStateValue::Narrow(0),
mask: Vec::new(),
window: None,
source,
order,
pulse: false,
coll: None,
obj: Some(w),
});
}
pub fn stage_pulse(&mut self, cell: StateCellId, row: Option<u64>, source: ProcessId) {
let order = self.bump_order();
self.masked.push(MaskedNbaUpdate {
cell,
row,
value: FourStateValue::Narrow(0),
mask: Vec::new(),
window: None,
source,
order,
pulse: true,
coll: None,
obj: None,
});
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty() && self.masked.is_empty()
}
pub fn len(&self) -> usize {
self.entries.len() + self.masked.len()
}
pub fn clear(&mut self) {
self.entries.clear();
self.masked.clear();
self.next_order = 0;
}
pub fn take(&mut self) -> Vec<NbaUpdate> {
if self.masked.is_empty() {
self.next_order = 0;
}
std::mem::take(&mut self.entries)
}
pub fn take_masked(&mut self) -> Vec<MaskedNbaUpdate> {
if self.entries.is_empty() {
self.next_order = 0;
}
std::mem::take(&mut self.masked)
}
}
#[cfg(test)]
mod tests {
use crate::state::{FourStateValue, StateCellId};
use super::{NbaLog, NbaUpdate};
#[test]
fn stage_appends_without_applying_to_any_state() {
let mut log = NbaLog::new();
assert!(log.is_empty());
log.stage(StateCellId(1), FourStateValue::Narrow(5), 100);
log.stage(StateCellId(1), FourStateValue::Narrow(9), 200);
assert_eq!(log.len(), 2);
assert!(!log.is_empty());
let entries = log.take();
assert_eq!(entries.len(), 2);
assert_eq!(entries[0].source, 100);
assert_eq!(entries[1].source, 200);
assert!(matches!(entries[0].value, FourStateValue::Narrow(5)));
assert!(matches!(entries[1].value, FourStateValue::Narrow(9)));
}
#[test]
fn order_field_is_populated_in_strict_push_order() {
let mut log = NbaLog::new();
log.stage(StateCellId(7), FourStateValue::Narrow(1), 1);
log.stage(StateCellId(3), FourStateValue::Narrow(2), 2);
log.stage(StateCellId(7), FourStateValue::Narrow(3), 1);
let entries = log.take();
assert_eq!(entries[0].order, 0);
assert_eq!(entries[1].order, 1);
assert_eq!(entries[2].order, 2);
}
#[test]
fn clear_empties_the_log_and_is_idempotent() {
let mut log = NbaLog::new();
log.stage(StateCellId(1), FourStateValue::Narrow(1), 1);
log.stage(StateCellId(2), FourStateValue::Narrow(2), 1);
assert_eq!(log.len(), 2);
log.clear();
assert!(log.is_empty());
assert_eq!(log.len(), 0);
log.clear();
assert!(log.is_empty());
}
#[test]
fn take_drains_exactly_once() {
let mut log = NbaLog::new();
log.stage(StateCellId(1), FourStateValue::Narrow(1), 1);
log.stage(StateCellId(2), FourStateValue::Narrow(2), 2);
let first = log.take();
assert_eq!(first.len(), 2);
assert!(log.is_empty());
let second = log.take();
assert!(second.is_empty());
}
#[test]
fn no_direct_apply_path_exists_on_nba_log() {
#[allow(dead_code)]
struct ExpectedShape {
entries: Vec<NbaUpdate>,
masked: Vec<super::MaskedNbaUpdate>,
next_order: u32,
}
assert_eq!(
std::mem::size_of::<NbaLog>(),
std::mem::size_of::<ExpectedShape>()
);
}
#[test]
fn several_processes_staging_in_one_timestep_stay_unapplied_until_one_take() {
let mut log = NbaLog::new();
log.stage(StateCellId(1), FourStateValue::Narrow(10), 1);
log.stage(StateCellId(2), FourStateValue::Narrow(20), 2);
log.stage(StateCellId(1), FourStateValue::Narrow(30), 3);
log.stage(StateCellId(3), FourStateValue::Narrow(40), 4);
assert_eq!(log.len(), 4);
let batch = log.take();
assert_eq!(
batch.len(),
4,
"the whole timestep's writes drain as one batch"
);
assert!(log.is_empty());
}
}