feat(ipc): saved replies: SaveReply/ReplySaved (8 per thread, released on thread death, unforgeable); terminal READ saved until Enter (line.rs editor, reader queue), now pinged
This commit is contained in:
@@ -6,7 +6,7 @@ use domain::{message::Message, syscall::Syscall};
|
||||
use types::{
|
||||
address::{DataRef, EntryPoint, StackPointer, VirtualAddress},
|
||||
badge::Badge,
|
||||
bounded::{CapSlot, PageCount},
|
||||
bounded::{CapSlot, PageCount, SavedReply},
|
||||
permission::PermissionSet,
|
||||
syscall_abi::*,
|
||||
};
|
||||
@@ -151,6 +151,11 @@ pub fn decode(nr: usize, a: &[usize; 6]) -> Option<Syscall> {
|
||||
notification: slot(a[0])?,
|
||||
deadline: types::time::Ticks::from_raw(a[1] as u64),
|
||||
},
|
||||
SYS_SAVE_REPLY => Syscall::SaveReply,
|
||||
SYS_REPLY_SAVED => Syscall::ReplySaved {
|
||||
reply: SavedReply::new(u8::try_from(a[0]).ok()?)?,
|
||||
message: message(a)?,
|
||||
},
|
||||
SYS_DEBUG_PUTCHAR => Syscall::DebugPutchar {
|
||||
console: slot(a[0])?,
|
||||
byte: a[1] as u8,
|
||||
|
||||
@@ -13,7 +13,7 @@ use types::{
|
||||
};
|
||||
|
||||
use super::super::{Delivery, DeliveryKind, SyscallOutcome, SyscallResult};
|
||||
use super::Outgoing;
|
||||
use super::{Outgoing, ThreadRef};
|
||||
use crate::Kernel;
|
||||
|
||||
impl<P: Platform> Kernel<P> {
|
||||
@@ -226,11 +226,17 @@ impl<P: Platform> Kernel<P> {
|
||||
.ipc_mut(tid)
|
||||
.and_then(|s| s.reply_to.take())
|
||||
.ok_or(KernelError::NoReply)?;
|
||||
self.answer(pid, caller, message);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Wakes `caller` with `message` from `pid`, if it is still waiting.
|
||||
pub(super) fn answer(&mut self, pid: ProcessId, caller: ThreadRef, message: Message) {
|
||||
let waiting = self.is_live(caller)
|
||||
&& self.is_blocked_as(caller.tid, |r| *r == BlockReason::AwaitingReply);
|
||||
if !waiting {
|
||||
// The caller died; the reply has nowhere to go.
|
||||
return Ok(());
|
||||
return;
|
||||
}
|
||||
let to = self.thread_process(caller.tid);
|
||||
// A reply may always carry a capability the server can delegate.
|
||||
@@ -242,7 +248,6 @@ impl<P: Platform> Kernel<P> {
|
||||
cap,
|
||||
};
|
||||
self.wake(caller.tid, SyscallResult::Received(delivery));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn reply_recv(
|
||||
|
||||
@@ -4,6 +4,7 @@ mod endpoint;
|
||||
mod events;
|
||||
mod irq;
|
||||
mod notification;
|
||||
mod saved;
|
||||
|
||||
use domain::{
|
||||
error::KernelError,
|
||||
@@ -44,13 +45,19 @@ pub struct IpcState {
|
||||
pub(crate) outgoing: Option<Outgoing>,
|
||||
/// The caller this thread owes a reply to.
|
||||
pub(crate) reply_to: Option<ThreadRef>,
|
||||
/// Callers it kept to answer later (`SaveReply`).
|
||||
pub(crate) saved: [Option<ThreadRef>; SAVED],
|
||||
pub(crate) bound: Option<NotificationId>,
|
||||
}
|
||||
|
||||
/// Callers one thread can keep waiting.
|
||||
pub(crate) const SAVED: usize = types::bounded::SavedReply::MAX as usize + 1;
|
||||
|
||||
impl IpcState {
|
||||
pub const EMPTY: Self = Self {
|
||||
outgoing: None,
|
||||
reply_to: None,
|
||||
saved: [None; SAVED],
|
||||
bound: None,
|
||||
};
|
||||
}
|
||||
|
||||
52
crates/kernel/src/syscall/ipc/saved.rs
Normal file
52
crates/kernel/src/syscall/ipc/saved.rs
Normal file
@@ -0,0 +1,52 @@
|
||||
//! Replies kept for later (docs/ipc.md): a server that cannot answer a call
|
||||
//! yet (it waits on the user, or on another server) keeps the caller and
|
||||
//! receives others meanwhile. The kept caller is named only within the
|
||||
//! thread that kept it, so it can be neither passed on nor forged, and it
|
||||
//! is released with an error when that thread dies.
|
||||
|
||||
use domain::{error::KernelError, message::Message};
|
||||
use platform::Platform;
|
||||
use types::{
|
||||
bounded::SavedReply,
|
||||
id::{ProcessId, ThreadId},
|
||||
};
|
||||
|
||||
use crate::Kernel;
|
||||
|
||||
impl<P: Platform> Kernel<P> {
|
||||
/// Moves the caller `tid` owes a reply to into its first free slot.
|
||||
pub(crate) fn save_reply(&mut self, tid: ThreadId) -> Result<SavedReply, KernelError> {
|
||||
let s = self.ipc_mut(tid).ok_or(KernelError::NoReply)?;
|
||||
let caller = s.reply_to.ok_or(KernelError::NoReply)?;
|
||||
let (index, free) = s
|
||||
.saved
|
||||
.iter_mut()
|
||||
.enumerate()
|
||||
.find(|(_, c)| c.is_none())
|
||||
.ok_or(KernelError::NoFreeSlot)?;
|
||||
let reply = u8::try_from(index)
|
||||
.ok()
|
||||
.and_then(SavedReply::new)
|
||||
.ok_or(KernelError::NoFreeSlot)?;
|
||||
*free = Some(caller);
|
||||
s.reply_to = None;
|
||||
Ok(reply)
|
||||
}
|
||||
|
||||
/// Answers the caller kept as `reply`; it is forgotten either way.
|
||||
pub(crate) fn reply_saved(
|
||||
&mut self,
|
||||
tid: ThreadId,
|
||||
pid: ProcessId,
|
||||
reply: SavedReply,
|
||||
message: Message,
|
||||
) -> Result<(), KernelError> {
|
||||
let caller = self
|
||||
.ipc_mut(tid)
|
||||
.and_then(|s| s.saved.get_mut(usize::from(reply.value())))
|
||||
.and_then(Option::take)
|
||||
.ok_or(KernelError::NoReply)?;
|
||||
self.answer(pid, caller, message);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -138,6 +138,14 @@ impl<P: Platform> Kernel<P> {
|
||||
Syscall::ReplyRecv { endpoint, message } => {
|
||||
self.reply_recv(tid, pid, endpoint, message)
|
||||
}
|
||||
Syscall::SaveReply => complete(
|
||||
self.save_reply(tid)
|
||||
.map(|r| SyscallResult::Value(usize::from(r.value()))),
|
||||
),
|
||||
Syscall::ReplySaved { reply, message } => complete(
|
||||
self.reply_saved(tid, pid, reply, message)
|
||||
.map(|()| SyscallResult::Ok),
|
||||
),
|
||||
Syscall::CreateNotification => {
|
||||
complete(self.create_notification_cap(pid).map(SyscallResult::Slot))
|
||||
}
|
||||
|
||||
@@ -46,7 +46,11 @@ impl<P: Platform> Kernel<P> {
|
||||
n.remove_thread(*tid);
|
||||
}
|
||||
let ipc = self.ipc_mut(*tid).map(core::mem::take).unwrap_or_default();
|
||||
if let Some(caller) = ipc.reply_to {
|
||||
for caller in ipc
|
||||
.reply_to
|
||||
.into_iter()
|
||||
.chain(ipc.saved.into_iter().flatten())
|
||||
{
|
||||
self.release_caller(caller);
|
||||
}
|
||||
self.evict(*tid);
|
||||
|
||||
@@ -44,6 +44,8 @@ fn every_defined_syscall_decodes() {
|
||||
SYS_REGION_ADDRESS,
|
||||
SYS_WATCH_DEATH,
|
||||
SYS_SET_TIMER,
|
||||
SYS_SAVE_REPLY,
|
||||
SYS_REPLY_SAVED,
|
||||
];
|
||||
// slot 1, slot 2, small values, no cap
|
||||
let args = [1, 2, 1, 0, 0, NONE];
|
||||
@@ -61,6 +63,11 @@ fn retired_and_unknown_numbers_do_not_decode() {
|
||||
|
||||
#[test]
|
||||
fn out_of_range_arguments_do_not_decode() {
|
||||
assert!(
|
||||
decode(SYS_REPLY_SAVED, &[8, 0, 0, 0, 0, NONE]).is_none(),
|
||||
"a thread keeps 8 replies"
|
||||
);
|
||||
assert!(decode(SYS_REPLY_SAVED, &[7, 0, 0, 0, 0, NONE]).is_some());
|
||||
assert!(decode(SYS_RECEIVE, &[64, 0, 0, 0, 0, 0]).is_none());
|
||||
assert!(
|
||||
decode(SYS_SEND, &[1, 0, 0, 0, 0, 64]).is_none(),
|
||||
|
||||
@@ -110,6 +110,9 @@ fn error(outcome: &SyscallOutcome) -> Option<&KernelError> {
|
||||
}
|
||||
}
|
||||
|
||||
#[path = "ipc_saved.rs"]
|
||||
mod saved;
|
||||
|
||||
// ---- rendezvous ----
|
||||
|
||||
#[test]
|
||||
|
||||
192
crates/kernel/src/tests/ipc_saved.rs
Normal file
192
crates/kernel/src/tests/ipc_saved.rs
Normal file
@@ -0,0 +1,192 @@
|
||||
//! Replies kept for later: `SaveReply` and `ReplySaved`.
|
||||
|
||||
use super::*;
|
||||
use types::bounded::SavedReply;
|
||||
|
||||
/// A call from `client` that `server` receives and keeps; which one.
|
||||
fn call_and_save(t: &mut Ipc, client: ThreadId, word: usize) -> SavedReply {
|
||||
let (tb, a_send, b_recv) = (t.tb, t.a_send, t.b_recv);
|
||||
t.run(
|
||||
client,
|
||||
Syscall::Call {
|
||||
endpoint: a_send,
|
||||
message: msg([word, 0, 0, 0]),
|
||||
},
|
||||
);
|
||||
t.run(tb, Syscall::Receive { endpoint: b_recv });
|
||||
match t.run(tb, Syscall::SaveReply) {
|
||||
SyscallOutcome::Complete(SyscallResult::Value(v)) => SavedReply::new(v as u8).unwrap(),
|
||||
_ => panic!("not saved"),
|
||||
}
|
||||
}
|
||||
|
||||
fn reply_saved(t: &mut Ipc, reply: SavedReply, word: usize) -> SyscallOutcome {
|
||||
let tb = t.tb;
|
||||
t.run(
|
||||
tb,
|
||||
Syscall::ReplySaved {
|
||||
reply,
|
||||
message: msg([word, 0, 0, 0]),
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
fn client(t: &mut Ipc) -> ThreadId {
|
||||
t.k.spawn_thread_for_test(t.a, Priority::DEFAULT, TimeSlice::DEFAULT)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_saved_caller_is_answered_after_others_are_served() {
|
||||
let mut t = Ipc::new();
|
||||
let (ta, tb, b_recv) = (t.ta, t.tb, t.b_recv);
|
||||
let first = call_and_save(&mut t, ta, 1);
|
||||
// The server goes on: the next caller is received and answered first.
|
||||
let other = client(&mut t);
|
||||
let a_send = t.a_send;
|
||||
t.run(
|
||||
other,
|
||||
Syscall::Call {
|
||||
endpoint: a_send,
|
||||
message: msg([2, 0, 0, 0]),
|
||||
},
|
||||
);
|
||||
let d = delivered(&t.run(tb, Syscall::Receive { endpoint: b_recv })).unwrap();
|
||||
assert_eq!(d.words[0], 2);
|
||||
assert!(
|
||||
matches!(t.state(ta), Some(ThreadState::Blocked(_))),
|
||||
"the saved caller still waits"
|
||||
);
|
||||
t.run(
|
||||
tb,
|
||||
Syscall::Reply {
|
||||
message: msg([20, 0, 0, 0]),
|
||||
},
|
||||
);
|
||||
assert_eq!(pending_delivery(t.woken_with(other)).unwrap().words[0], 20);
|
||||
assert!(matches!(
|
||||
reply_saved(&mut t, first, 10),
|
||||
SyscallOutcome::Complete(SyscallResult::Ok)
|
||||
));
|
||||
assert!(matches!(t.state(ta), Some(ThreadState::Ready)));
|
||||
assert_eq!(pending_delivery(t.woken_with(ta)).unwrap().words[0], 10);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nothing_to_save_without_a_pending_call() {
|
||||
let mut t = Ipc::new();
|
||||
let tb = t.tb;
|
||||
assert!(matches!(
|
||||
error(&t.run(tb, Syscall::SaveReply)),
|
||||
Some(KernelError::NoReply)
|
||||
));
|
||||
// Saving takes it: there is no reply pending any more.
|
||||
let ta = t.ta;
|
||||
call_and_save(&mut t, ta, 0);
|
||||
assert!(matches!(
|
||||
error(&t.run(
|
||||
tb,
|
||||
Syscall::Reply {
|
||||
message: msg([0; 4])
|
||||
}
|
||||
)),
|
||||
Some(KernelError::NoReply)
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_saved_reply_is_answered_once() {
|
||||
let mut t = Ipc::new();
|
||||
let ta = t.ta;
|
||||
let r = call_and_save(&mut t, ta, 0);
|
||||
assert!(error(&reply_saved(&mut t, r, 1)).is_none());
|
||||
assert!(matches!(
|
||||
error(&reply_saved(&mut t, r, 2)),
|
||||
Some(KernelError::NoReply)
|
||||
));
|
||||
assert!(
|
||||
matches!(
|
||||
error(&reply_saved(&mut t, SavedReply::new(5).unwrap(), 2)),
|
||||
Some(KernelError::NoReply)
|
||||
),
|
||||
"never saved"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_receive_does_not_release_saved_callers() {
|
||||
let mut t = Ipc::new();
|
||||
let (ta, tb, b_recv) = (t.ta, t.tb, t.b_recv);
|
||||
call_and_save(&mut t, ta, 0);
|
||||
// Nothing queued: the server just waits, the saved caller with it.
|
||||
assert!(blocked(&t.run(tb, Syscall::Receive { endpoint: b_recv })));
|
||||
assert!(matches!(t.state(ta), Some(ThreadState::Blocked(_))));
|
||||
assert!(t.woken_with(ta).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_thread_keeps_eight_callers_at_most() {
|
||||
let mut t = Ipc::new();
|
||||
let (tb, a_send, b_recv) = (t.tb, t.a_send, t.b_recv);
|
||||
let ta = t.ta;
|
||||
call_and_save(&mut t, ta, 0);
|
||||
for i in 1..=usize::from(SavedReply::MAX) {
|
||||
let c = client(&mut t);
|
||||
assert_eq!(usize::from(call_and_save(&mut t, c, i).value()), i);
|
||||
}
|
||||
let late = client(&mut t);
|
||||
t.run(
|
||||
late,
|
||||
Syscall::Call {
|
||||
endpoint: a_send,
|
||||
message: msg([9, 0, 0, 0]),
|
||||
},
|
||||
);
|
||||
t.run(tb, Syscall::Receive { endpoint: b_recv });
|
||||
assert!(matches!(
|
||||
error(&t.run(tb, Syscall::SaveReply)),
|
||||
Some(KernelError::NoFreeSlot)
|
||||
));
|
||||
// Still the pending reply: it can be answered the usual way.
|
||||
t.run(
|
||||
tb,
|
||||
Syscall::Reply {
|
||||
message: msg([90, 0, 0, 0]),
|
||||
},
|
||||
);
|
||||
assert_eq!(pending_delivery(t.woken_with(late)).unwrap().words[0], 90);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn server_death_releases_its_saved_callers() {
|
||||
let mut t = Ipc::new();
|
||||
let (ta, b) = (t.ta, t.b);
|
||||
call_and_save(&mut t, ta, 0);
|
||||
let other = client(&mut t);
|
||||
call_and_save(&mut t, other, 1);
|
||||
t.k.destroy_process(b);
|
||||
for c in [ta, other] {
|
||||
assert!(matches!(t.state(c), Some(ThreadState::Ready)));
|
||||
assert!(matches!(
|
||||
t.woken_with(c),
|
||||
Some(SyscallResult::Error(KernelError::NoReply))
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_saved_reply_to_a_caller_that_died_goes_nowhere() {
|
||||
let mut t = Ipc::new();
|
||||
let other = client(&mut t);
|
||||
let r = call_and_save(&mut t, other, 0);
|
||||
// Killed while waiting for the reply (its process holds others).
|
||||
t.k.cleanup_threads(&[Some(other)]);
|
||||
let reborn = client(&mut t);
|
||||
assert!(
|
||||
error(&reply_saved(&mut t, r, 7)).is_none(),
|
||||
"answering a dead caller is not the server's error"
|
||||
);
|
||||
assert!(
|
||||
t.woken_with(reborn).is_none(),
|
||||
"and a thread that reused its id hears nothing"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user