use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use crate::sched::{ProcessId, ProcessRequest};
#[derive(Default)]
struct Event {
waiters: Vec<ProcessId>,
last_trigger: Option<u64>,
}
#[derive(Default)]
struct Table {
events: Vec<Event>,
waiting_on: HashMap<ProcessId, Vec<u64>>,
dual: HashSet<ProcessId>,
}
thread_local! {
static TABLE: RefCell<Table> = RefCell::new(Table::default());
}
fn with_table<R>(f: impl FnOnce(&mut Table) -> R) -> R {
TABLE.with(|t| f(&mut t.borrow_mut()))
}
pub fn alloc() -> u64 {
with_table(|t| {
t.events.push(Event::default());
t.events.len() as u64
})
}
thread_local! {
static DEFERRED: RefCell<Vec<ProcessRequest>> = const { RefCell::new(Vec::new()) };
}
pub fn defer_wake(req: ProcessRequest) {
DEFERRED.with(|d| d.borrow_mut().push(req));
}
pub fn take_deferred_wakes() -> Vec<ProcessRequest> {
DEFERRED.with(|d| std::mem::take(&mut *d.borrow_mut()))
}
pub fn trigger(id: u64, now: u64, emit: &mut dyn FnMut(ProcessRequest)) {
let woken = with_table(|t| {
let ev = usize::try_from(id)
.ok()
.and_then(|i| i.checked_sub(1))
.and_then(|i| t.events.get_mut(i))?;
ev.last_trigger = Some(now);
let woken = std::mem::take(&mut ev.waiters);
for p in &woken {
for other in t.waiting_on.remove(p).unwrap_or_default() {
if other != id
&& let Some(oe) = t.events.get_mut((other - 1) as usize)
{
oe.waiters.retain(|w| w != p);
}
}
}
Some(woken)
});
for p in woken.unwrap_or_default() {
emit(ProcessRequest::SyncWake(p));
}
}
pub fn enqueue(id: u64, proc: ProcessId) {
with_table(|t| {
let known = usize::try_from(id)
.ok()
.and_then(|i| i.checked_sub(1))
.is_some_and(|i| i < t.events.len());
if known && !t.waiting_on.get(&proc).is_some_and(|v| v.contains(&id)) {
t.events[(id - 1) as usize].waiters.push(proc);
t.waiting_on.entry(proc).or_default().push(id);
}
});
}
pub fn enqueue_dual(id: u64, proc: ProcessId) {
with_table(|t| {
t.dual.insert(proc);
});
enqueue(id, proc);
}
pub fn is_dual(proc: ProcessId) -> bool {
with_table(|t| t.dual.contains(&proc))
}
pub fn leave(proc: ProcessId) {
cancel(proc);
}
pub fn wait(id: u64, proc: ProcessId, emit: &mut dyn FnMut(ProcessRequest)) {
enqueue(id, proc);
emit(ProcessRequest::SyncBlock);
}
pub fn triggered(id: u64, now: u64) -> bool {
with_table(|t| {
usize::try_from(id)
.ok()
.and_then(|i| i.checked_sub(1))
.and_then(|i| t.events.get(i))
.is_some_and(|ev| ev.last_trigger == Some(now))
})
}
pub fn waiters(id: u64) -> usize {
with_table(|t| {
usize::try_from(id)
.ok()
.and_then(|i| i.checked_sub(1))
.and_then(|i| t.events.get(i))
.map_or(0, |ev| ev.waiters.len())
})
}
pub fn cancel(proc: ProcessId) {
with_table(|t| {
t.dual.remove(&proc);
for id in t.waiting_on.remove(&proc).unwrap_or_default() {
if let Some(ev) = t.events.get_mut((id - 1) as usize) {
ev.waiters.retain(|&p| p != proc);
}
}
});
}
pub fn reset_for_new_run() {
DEFERRED.with(|d| d.borrow_mut().clear());
with_table(|t| *t = Table::default());
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Default)]
struct Log(Vec<ProcessRequest>);
#[test]
fn trigger_wakes_every_waiter_in_order_and_stamps_the_step() {
reset_for_new_run();
let e = alloc();
let mut log = Log::default();
for p in [3, 1, 2] {
wait(e, p, &mut |r| log.0.push(r));
}
assert_eq!(waiters(e), 3);
log.0.clear();
assert!(!triggered(e, 5));
trigger(e, 5, &mut |r| log.0.push(r));
assert_eq!(
log.0,
[
ProcessRequest::SyncWake(3),
ProcessRequest::SyncWake(1),
ProcessRequest::SyncWake(2)
]
);
assert_eq!(waiters(e), 0);
assert!(triggered(e, 5));
assert!(!triggered(e, 6), "triggered lasts one time step");
}
#[test]
fn an_or_list_wakes_once_on_the_first_trigger_and_leaves_the_other_queues() {
reset_for_new_run();
let (a, b) = (alloc(), alloc());
let mut log = Log::default();
enqueue(a, 4);
enqueue(a, 4);
wait(b, 4, &mut |r| log.0.push(r));
assert_eq!((waiters(a), waiters(b)), (1, 1));
log.0.clear();
trigger(b, 1, &mut |r| log.0.push(r));
assert_eq!(log.0, [ProcessRequest::SyncWake(4)]);
assert_eq!((waiters(a), waiters(b)), (0, 0));
trigger(a, 1, &mut |r| log.0.push(r));
assert_eq!(log.0.len(), 1, "the process was already woken once");
}
#[test]
fn a_dual_waiter_is_marked_until_it_leaves() {
reset_for_new_run();
let (a, b) = (alloc(), alloc());
enqueue_dual(a, 6);
enqueue_dual(b, 6);
assert!(is_dual(6) && !is_dual(7));
assert_eq!((waiters(a), waiters(b)), (1, 1));
let mut log = Vec::new();
trigger(a, 1, &mut |r| log.push(r));
assert_eq!(log, [ProcessRequest::SyncWake(6)]);
assert!(is_dual(6), "still marked until the process runs again");
assert_eq!(waiters(b), 0, "the first trigger drops the other queues");
leave(6);
assert!(!is_dual(6));
enqueue_dual(a, 6);
leave(6);
assert_eq!(waiters(a), 0, "a signal-woken process leaves its queues");
}
#[test]
fn null_and_unknown_ids_trigger_nothing_and_never_wake() {
reset_for_new_run();
let mut log = Log::default();
trigger(0, 1, &mut |r| log.0.push(r));
trigger(99, 1, &mut |r| log.0.push(r));
assert!(log.0.is_empty());
assert!(!triggered(0, 1));
wait(0, 7, &mut |r| log.0.push(r));
assert_eq!(log.0, [ProcessRequest::SyncBlock]);
trigger(0, 2, &mut |r| log.0.push(r));
assert_eq!(log.0.len(), 1, "a null wait is never woken");
}
#[test]
fn a_killed_waiter_leaves_the_queue() {
reset_for_new_run();
let e = alloc();
let mut log = Log::default();
wait(e, 4, &mut |r| log.0.push(r));
wait(e, 5, &mut |r| log.0.push(r));
log.0.clear();
cancel(4);
trigger(e, 1, &mut |r| log.0.push(r));
assert_eq!(log.0, [ProcessRequest::SyncWake(5)]);
}
}