use std::cell::RefCell;
use std::collections::{HashMap, VecDeque};
use crate::objects;
use crate::sched::{ProcessId, ProcessRequest};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Message {
pub words: Box<[u64]>,
pub fingerprint: u64,
pub obj: bool,
pub coll: bool,
}
impl Message {
fn peeked(&self) -> Message {
if self.obj {
objects::retain(self.words[0]);
}
let mut m = self.clone();
if self.coll {
m.words[0] = crate::collections::clone(self.words[0]);
}
m
}
pub fn discard(self) {
if self.obj {
objects::release(self.words[0]);
}
if self.coll {
crate::collections::release(self.words[0]);
}
}
}
#[derive(Debug, PartialEq, Eq)]
enum Delivery {
Message(Message),
Mismatch,
}
#[derive(Debug)]
enum Pending {
Completed,
Waiting(u64),
Delivered(Delivery),
}
struct Getter {
proc: ProcessId,
peek: bool,
fingerprint: u64,
}
#[derive(Default)]
struct Mailbox {
bound: usize,
messages: VecDeque<Message>,
getters: VecDeque<Getter>,
putters: VecDeque<(ProcessId, Message)>,
}
#[derive(Default)]
struct Table {
mailboxes: HashMap<u64, Mailbox>,
pending: HashMap<ProcessId, Pending>,
stash: HashMap<(ProcessId, u32), u64>,
}
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()))
}
#[derive(Default)]
struct Releases {
handles: Vec<u64>,
colls: Vec<u64>,
}
impl Releases {
fn owned(&mut self, m: &Message) {
if m.obj {
self.handles.push(m.words[0]);
}
if m.coll {
self.colls.push(m.words[0]);
}
}
fn run(self) {
for h in self.handles {
objects::release(h);
}
for h in self.colls {
crate::collections::release(h);
}
super::notify_object_state_change();
}
}
pub fn init(h: u64, bound: i64) -> bool {
if !objects::is_live(h) {
return false;
}
let bound = usize::try_from(bound).unwrap_or(0);
with_table(|t| {
t.mailboxes.insert(
h,
Mailbox {
bound,
..Mailbox::default()
},
);
});
true
}
pub fn num(h: u64) -> usize {
with_table(|t| t.mailboxes.get(&h).map_or(0, |m| m.messages.len()))
}
pub fn waiters(h: u64) -> usize {
with_table(|t| {
t.mailboxes
.get(&h)
.map_or(0, |m| m.getters.len() + m.putters.len())
})
}
fn is_full(m: &Mailbox) -> bool {
m.bound != 0 && m.messages.len() >= m.bound
}
fn wake(
t: &mut Table,
proc: ProcessId,
h: u64,
delivery: Option<Delivery>,
emit: &mut dyn FnMut(ProcessRequest),
rel: &mut Releases,
) {
match delivery {
Some(d) => {
t.pending.insert(proc, Pending::Delivered(d));
}
None => {
t.pending.remove(&proc);
}
}
emit(ProcessRequest::SyncWake(proc));
rel.handles.push(h);
}
fn admit_putters(t: &mut Table, h: u64, emit: &mut dyn FnMut(ProcessRequest), rel: &mut Releases) {
loop {
let Some(m) = t.mailboxes.get_mut(&h) else {
return;
};
if is_full(m) {
return;
}
let Some((proc, msg)) = m.putters.pop_front() else {
return;
};
m.messages.push_back(msg);
wake(t, proc, h, None, emit, rel);
}
}
fn serve_getters(t: &mut Table, h: u64, emit: &mut dyn FnMut(ProcessRequest), rel: &mut Releases) {
loop {
let Some(m) = t.mailboxes.get_mut(&h) else {
return;
};
if m.messages.is_empty() {
return;
}
let Some(g) = m.getters.pop_front() else {
return;
};
let head_fp = m.messages[0].fingerprint;
let delivery = if head_fp != g.fingerprint {
Delivery::Mismatch
} else if g.peek {
Delivery::Message(m.messages[0].peeked())
} else {
let msg = m.messages.pop_front().expect("checked non-empty");
Delivery::Message(msg)
};
let removed = matches!(&delivery, Delivery::Message(_)) && !g.peek;
wake(t, g.proc, h, Some(delivery), emit, rel);
if removed {
admit_putters(t, h, emit, rel);
}
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum PutResult {
Stored,
Blocked,
}
pub fn put(
h: u64,
proc: ProcessId,
msg: Message,
emit: &mut dyn FnMut(ProcessRequest),
) -> Result<PutResult, Message> {
let mut rel = Releases::default();
let result = with_table(|t| {
let Some(m) = t.mailboxes.get_mut(&h) else {
return Err(msg);
};
if !is_full(m) {
m.messages.push_back(msg);
serve_getters(t, h, emit, &mut rel);
return Ok(PutResult::Stored);
}
assert!(
!t.pending.contains_key(&proc),
"mailbox: process {proc} called put() while already blocked in a mailbox call"
);
objects::retain(h);
t.mailboxes
.get_mut(&h)
.expect("looked up above")
.putters
.push_back((proc, msg));
t.pending.insert(proc, Pending::Waiting(h));
emit(ProcessRequest::SyncBlock);
Ok(PutResult::Blocked)
});
rel.run();
result
}
pub fn try_put(h: u64, msg: Message, emit: &mut dyn FnMut(ProcessRequest)) -> Result<(), Message> {
let mut rel = Releases::default();
let result = with_table(|t| {
let Some(m) = t.mailboxes.get_mut(&h) else {
return Err(msg);
};
if is_full(m) {
return Err(msg);
}
m.messages.push_back(msg);
serve_getters(t, h, emit, &mut rel);
Ok(())
});
rel.run();
result
}
#[derive(Debug, PartialEq, Eq)]
pub enum Taken {
Empty,
Mismatch,
Got(Message),
}
fn take_front(
t: &mut Table,
h: u64,
fingerprint: u64,
peek: bool,
emit: &mut dyn FnMut(ProcessRequest),
rel: &mut Releases,
) -> Taken {
let Some(m) = t.mailboxes.get_mut(&h) else {
return Taken::Empty;
};
let Some(head) = m.messages.front() else {
return Taken::Empty;
};
if head.fingerprint != fingerprint {
return Taken::Mismatch;
}
if peek {
return Taken::Got(head.peeked());
}
let msg = m.messages.pop_front().expect("checked non-empty");
admit_putters(t, h, emit, rel);
Taken::Got(msg)
}
pub fn try_take(
h: u64,
fingerprint: u64,
peek: bool,
emit: &mut dyn FnMut(ProcessRequest),
) -> Taken {
let mut rel = Releases::default();
let taken = with_table(|t| take_front(t, h, fingerprint, peek, emit, &mut rel));
rel.run();
taken
}
#[derive(Debug, PartialEq, Eq)]
pub enum Got {
Nothing,
NoMailbox,
Message(Message),
Mismatch,
Blocked,
}
pub fn get(
h: u64,
proc: ProcessId,
fingerprint: u64,
peek: bool,
emit: &mut dyn FnMut(ProcessRequest),
) -> Got {
let mut rel = Releases::default();
let got = with_table(|t| {
match t.pending.remove(&proc) {
Some(Pending::Completed) => return Got::Nothing,
Some(Pending::Delivered(Delivery::Message(m))) => return Got::Message(m),
Some(Pending::Delivered(Delivery::Mismatch)) => return Got::Mismatch,
Some(Pending::Waiting(on)) => {
panic!("mailbox: process {proc} resumed while still blocked on mailbox {on:#x}")
}
None => {}
}
if !t.mailboxes.contains_key(&h) {
t.pending.insert(proc, Pending::Completed);
return Got::NoMailbox;
}
match take_front(t, h, fingerprint, peek, emit, &mut rel) {
Taken::Got(m) => {
t.pending.insert(proc, Pending::Completed);
Got::Message(m)
}
Taken::Mismatch => {
t.pending.insert(proc, Pending::Completed);
Got::Mismatch
}
Taken::Empty => {
objects::retain(h);
t.mailboxes
.get_mut(&h)
.expect("looked up above")
.getters
.push_back(Getter {
proc,
peek,
fingerprint,
});
t.pending.insert(proc, Pending::Waiting(h));
emit(ProcessRequest::SyncBlock);
Got::Blocked
}
}
});
rel.run();
got
}
pub fn second_phase(proc: ProcessId) -> bool {
with_table(|t| t.pending.contains_key(&proc))
}
pub fn stash_set(proc: ProcessId, slot: u32, value: u64) {
with_table(|t| {
t.stash.insert((proc, slot), value);
});
}
pub fn stash_get(proc: ProcessId, slot: u32) -> u64 {
with_table(|t| t.stash.get(&(proc, slot)).copied().unwrap_or(0))
}
pub fn cancel(proc: ProcessId) {
let mut rel = Releases::default();
with_table(|t| t.stash.retain(|(p, _), _| *p != proc));
with_table(|t| match t.pending.remove(&proc) {
None | Some(Pending::Completed) => {}
Some(Pending::Delivered(Delivery::Mismatch)) => {}
Some(Pending::Delivered(Delivery::Message(m))) => {
rel.owned(&m);
}
Some(Pending::Waiting(h)) => {
if let Some(m) = t.mailboxes.get_mut(&h) {
m.getters.retain(|g| g.proc != proc);
let mut kept = VecDeque::new();
for (p, msg) in std::mem::take(&mut m.putters) {
if p == proc {
rel.owned(&msg);
} else {
kept.push_back((p, msg));
}
}
m.putters = kept;
}
rel.handles.push(h);
}
});
rel.run();
}
pub fn reset_for_new_run() {
let old = with_table(std::mem::take);
let mut rel = Releases::default();
for (_, p) in old.pending {
if let Pending::Delivered(Delivery::Message(m)) = p {
rel.owned(&m);
}
}
for (h, m) in old.mailboxes {
for _ in 0..(m.getters.len() + m.putters.len()) {
rel.handles.push(h);
}
for (_, msg) in m.putters {
rel.owned(&msg);
}
for msg in m.messages {
msg.discard();
}
}
rel.run();
}
#[cfg(test)]
mod tests {
use super::*;
use crate::netlist::ClassTemplateId;
const FP: u64 = 0x11;
const OTHER_FP: u64 = 0x22;
fn mbx(bound: i64, template: u32) -> u64 {
let h = objects::alloc(ClassTemplateId(template), 1);
assert!(init(h, bound));
h
}
fn msg(v: u64) -> Message {
Message {
words: vec![v].into_boxed_slice(),
fingerprint: FP,
obj: false,
coll: false,
}
}
fn value(m: &Message) -> u64 {
m.words[0]
}
#[derive(Default)]
struct Log(Vec<ProcessRequest>);
impl Log {
fn sink(&mut self) -> impl FnMut(ProcessRequest) + '_ {
|r| self.0.push(r)
}
}
#[test]
fn messages_come_out_in_fifo_order() {
let h = mbx(0, 18_000);
let mut log = Log::default();
for v in [10, 20, 30] {
assert_eq!(
put(h, 2_000, msg(v), &mut log.sink()),
Ok(PutResult::Stored)
);
}
assert_eq!(num(h), 3);
for want in [10, 20, 30] {
let Taken::Got(m) = try_take(h, FP, false, &mut log.sink()) else {
panic!("a message was queued");
};
assert_eq!(value(&m), want);
}
assert_eq!(try_take(h, FP, false, &mut log.sink()), Taken::Empty);
assert!(log.0.is_empty());
}
#[test]
fn try_put_fails_on_a_full_bounded_mailbox_and_hands_the_message_back() {
let h = mbx(2, 18_001);
let mut log = Log::default();
assert!(try_put(h, msg(1), &mut log.sink()).is_ok());
assert!(try_put(h, msg(2), &mut log.sink()).is_ok());
let back = try_put(h, msg(3), &mut log.sink()).unwrap_err();
assert_eq!(value(&back), 3);
assert_eq!(num(h), 2);
}
#[test]
fn an_unbounded_mailbox_is_never_full() {
let h = mbx(0, 18_002);
let mut log = Log::default();
for v in 0..1000 {
assert!(try_put(h, msg(v), &mut log.sink()).is_ok());
}
assert_eq!(num(h), 1000);
let n = mbx(-3, 18_003);
assert!(try_put(n, msg(1), &mut log.sink()).is_ok());
assert!(try_put(n, msg(2), &mut log.sink()).is_ok());
}
#[test]
fn a_get_on_an_empty_mailbox_blocks_and_a_put_delivers_to_it() {
let h = mbx(0, 18_004);
let mut log = Log::default();
assert_eq!(get(h, 2_010, FP, false, &mut log.sink()), Got::Blocked);
assert_eq!(log.0, [ProcessRequest::SyncBlock]);
assert_eq!(waiters(h), 1);
log.0.clear();
assert_eq!(
put(h, 2_011, msg(7), &mut log.sink()),
Ok(PutResult::Stored)
);
assert_eq!(log.0, [ProcessRequest::SyncWake(2_010)]);
assert_eq!(num(h), 0, "the get took the message");
match get(h, 2_010, FP, false, &mut log.sink()) {
Got::Message(m) => assert_eq!(value(&m), 7),
other => panic!("second phase must deliver: {other:?}"),
}
assert_eq!(get(h, 2_010, FP, false, &mut log.sink()), Got::Blocked);
}
#[test]
fn a_get_that_finds_a_message_completes_and_its_second_phase_is_a_no_op() {
let h = mbx(0, 18_005);
let mut log = Log::default();
put(h, 2_020, msg(5), &mut log.sink()).unwrap();
match get(h, 2_021, FP, false, &mut log.sink()) {
Got::Message(m) => assert_eq!(value(&m), 5),
other => panic!("{other:?}"),
}
assert_eq!(get(h, 2_021, FP, false, &mut log.sink()), Got::Nothing);
assert!(log.0.is_empty());
assert_eq!(num(h), 0);
}
#[test]
fn waiters_are_served_in_arrival_order() {
let h = mbx(0, 18_006);
let mut log = Log::default();
for p in [2_030, 2_031, 2_032] {
assert_eq!(get(h, p, FP, false, &mut log.sink()), Got::Blocked);
}
log.0.clear();
put(h, 2_039, msg(1), &mut log.sink()).unwrap();
assert_eq!(log.0, [ProcessRequest::SyncWake(2_030)]);
put(h, 2_039, msg(2), &mut log.sink()).unwrap();
assert_eq!(
log.0,
[
ProcessRequest::SyncWake(2_030),
ProcessRequest::SyncWake(2_031)
]
);
assert_eq!(waiters(h), 1);
}
#[test]
fn one_message_wakes_every_peeker_and_the_first_getter() {
let h = mbx(0, 18_007);
let mut log = Log::default();
assert_eq!(get(h, 2_040, FP, true, &mut log.sink()), Got::Blocked);
assert_eq!(get(h, 2_041, FP, true, &mut log.sink()), Got::Blocked);
assert_eq!(get(h, 2_042, FP, false, &mut log.sink()), Got::Blocked);
assert_eq!(get(h, 2_043, FP, true, &mut log.sink()), Got::Blocked);
log.0.clear();
put(h, 2_049, msg(9), &mut log.sink()).unwrap();
assert_eq!(
log.0,
[
ProcessRequest::SyncWake(2_040),
ProcessRequest::SyncWake(2_041),
ProcessRequest::SyncWake(2_042)
],
"two peeks copy, the get removes it, the later peek stays blocked"
);
assert_eq!(num(h), 0);
assert_eq!(waiters(h), 1);
for p in [2_040, 2_041, 2_042] {
match get(h, p, FP, p == 2_040 || p == 2_041, &mut log.sink()) {
Got::Message(m) => assert_eq!(value(&m), 9),
other => panic!("process {p}: {other:?}"),
}
}
}
#[test]
fn peek_leaves_the_message_in_the_queue() {
let h = mbx(0, 18_008);
let mut log = Log::default();
put(h, 2_050, msg(4), &mut log.sink()).unwrap();
let Taken::Got(m) = try_take(h, FP, true, &mut log.sink()) else {
panic!("a message was queued");
};
assert_eq!(value(&m), 4);
assert_eq!(num(h), 1);
}
#[test]
fn a_put_on_a_full_mailbox_blocks_until_a_get_makes_room() {
let h = mbx(1, 18_009);
let mut log = Log::default();
put(h, 2_060, msg(1), &mut log.sink()).unwrap();
assert_eq!(
put(h, 2_061, msg(2), &mut log.sink()),
Ok(PutResult::Blocked)
);
assert_eq!(log.0, [ProcessRequest::SyncBlock]);
assert_eq!(num(h), 1, "the blocked message is not queued yet");
log.0.clear();
let Taken::Got(m) = try_take(h, FP, false, &mut log.sink()) else {
panic!("a message was queued");
};
assert_eq!(value(&m), 1);
assert_eq!(log.0, [ProcessRequest::SyncWake(2_061)]);
assert_eq!(num(h), 1, "the blocked putter's message moved in");
let Taken::Got(m) = try_take(h, FP, false, &mut log.sink()) else {
panic!("a message was queued");
};
assert_eq!(value(&m), 2);
}
#[test]
fn a_blocked_putter_is_not_overtaken_by_try_put() {
let h = mbx(1, 18_010);
let mut log = Log::default();
put(h, 2_070, msg(1), &mut log.sink()).unwrap();
put(h, 2_071, msg(2), &mut log.sink()).unwrap();
try_take(h, FP, false, &mut log.sink());
assert!(
try_put(h, msg(3), &mut log.sink()).is_err(),
"the freed slot went to the blocked putter"
);
}
#[test]
fn a_type_mismatch_takes_nothing() {
let h = mbx(0, 18_011);
let mut log = Log::default();
put(h, 2_080, msg(1), &mut log.sink()).unwrap();
assert_eq!(
try_take(h, OTHER_FP, false, &mut log.sink()),
Taken::Mismatch
);
assert_eq!(
try_take(h, OTHER_FP, true, &mut log.sink()),
Taken::Mismatch
);
assert_eq!(
get(h, 2_081, OTHER_FP, false, &mut log.sink()),
Got::Mismatch
);
assert_eq!(num(h), 1, "the message is still there");
let e = mbx(0, 18_012);
assert_eq!(
get(e, 2_082, OTHER_FP, false, &mut log.sink()),
Got::Blocked
);
assert_eq!(get(e, 2_083, FP, false, &mut log.sink()), Got::Blocked);
log.0.clear();
put(e, 2_084, msg(8), &mut log.sink()).unwrap();
assert_eq!(
log.0,
[
ProcessRequest::SyncWake(2_082),
ProcessRequest::SyncWake(2_083)
]
);
assert_eq!(
get(e, 2_082, OTHER_FP, false, &mut log.sink()),
Got::Mismatch
);
assert!(matches!(
get(e, 2_083, FP, false, &mut log.sink()),
Got::Message(_)
));
}
fn obj_msg(h: u64) -> Message {
Message {
words: vec![h].into_boxed_slice(),
fingerprint: FP,
obj: true,
coll: false,
}
}
#[test]
fn a_class_message_owns_one_reference_until_it_is_taken() {
let h = mbx(0, 18_013);
let payload = objects::alloc(ClassTemplateId(18_014), 1);
let mut log = Log::default();
put(h, 2_090, obj_msg(payload), &mut log.sink()).unwrap();
assert_eq!(objects::refcount(payload), Some(1), "the message holds it");
let Taken::Got(copy) = try_take(h, FP, true, &mut log.sink()) else {
panic!("a message was queued");
};
assert_eq!(
objects::refcount(payload),
Some(2),
"a peeked copy owns its own"
);
copy.discard();
let Taken::Got(taken) = try_take(h, FP, false, &mut log.sink()) else {
panic!("a message was queued");
};
assert_eq!(
objects::refcount(payload),
Some(1),
"taking transfers the queue's reference"
);
taken.discard();
assert_eq!(objects::refcount(payload), None, "the last holder freed it");
}
#[test]
fn cancelling_a_blocked_putter_releases_its_message_and_the_mailbox() {
let h = mbx(1, 18_015);
let payload = objects::alloc(ClassTemplateId(18_016), 1);
let mut log = Log::default();
put(h, 2_100, msg(1), &mut log.sink()).unwrap();
assert_eq!(
put(h, 2_101, obj_msg(payload), &mut log.sink()),
Ok(PutResult::Blocked)
);
assert_eq!(objects::refcount(h), Some(2));
cancel(2_101);
assert_eq!(objects::refcount(payload), None, "its message was dropped");
assert_eq!(objects::refcount(h), Some(1));
assert_eq!(waiters(h), 0);
cancel(2_101);
}
#[test]
fn cancelling_a_blocked_getter_removes_it_and_a_woken_one_loses_its_message() {
let h = mbx(0, 18_017);
let mut log = Log::default();
assert_eq!(get(h, 2_110, FP, false, &mut log.sink()), Got::Blocked);
cancel(2_110);
assert_eq!(waiters(h), 0);
assert_eq!(objects::refcount(h), Some(1));
put(h, 2_111, msg(1), &mut log.sink()).unwrap();
assert_eq!(num(h), 1, "nobody is waiting any more");
let payload = objects::alloc(ClassTemplateId(18_018), 1);
let e = mbx(0, 18_019);
assert_eq!(get(e, 2_112, FP, false, &mut log.sink()), Got::Blocked);
put(e, 2_113, obj_msg(payload), &mut log.sink()).unwrap();
cancel(2_112);
assert_eq!(
objects::refcount(payload),
None,
"the lost message was freed"
);
}
#[test]
fn a_waiter_keeps_its_mailbox_alive_until_it_leaves() {
let h = mbx(0, 18_020);
let mut log = Log::default();
assert_eq!(get(h, 2_120, FP, false, &mut log.sink()), Got::Blocked);
assert_eq!(objects::refcount(h), Some(2));
put(h, 2_121, msg(1), &mut log.sink()).unwrap();
assert_eq!(objects::refcount(h), Some(1), "the woken waiter let go");
}
#[test]
fn a_non_mailbox_handle_stores_and_delivers_nothing() {
let mut log = Log::default();
assert!(!init(0, 0));
assert_eq!(num(0), 0);
assert_eq!(put(0, 2_130, msg(1), &mut log.sink()), Err(msg(1)));
assert!(try_put(0, msg(1), &mut log.sink()).is_err());
assert_eq!(try_take(0, FP, false, &mut log.sink()), Taken::Empty);
assert_eq!(get(0, 2_131, FP, false, &mut log.sink()), Got::NoMailbox);
assert_eq!(get(0, 2_131, FP, false, &mut log.sink()), Got::Nothing);
assert!(log.0.is_empty());
}
}