use std::marker::PhantomData;
use std::time::Duration;
use crate::distributed::node_id;
use crate::host::{self, api};
use crate::mailbox::Catching;
use crate::serializer::{Bincode, Serializer};
use crate::supervisor::{Supervisable, Supervisor, SupervisorConfig};
use crate::timer::TimerRef;
use crate::{Mailbox, MailboxResult, Process, ProcessConfig, Tag};
pub fn process_id() -> u64 {
unsafe { api::process::process_id() }
}
pub trait AbstractProcess {
type Arg: serde::Serialize + serde::de::DeserializeOwned;
type State;
fn init(this: ProcessRef<Self>, arg: Self::Arg) -> Self::State;
fn terminate(_state: Self::State) {}
fn handle_link_trapped(_state: &mut Self::State, _tag: Tag) {}
}
pub trait MessageHandler<M, S = Bincode>: AbstractProcess
where
S: Serializer<M>,
{
fn handle(state: &mut Self::State, message: M);
}
pub trait RequestHandler<M, S = Bincode>: AbstractProcess
where
S: Serializer<M>,
{
type Response;
fn handle(state: &mut Self::State, request: M) -> Self::Response;
}
pub trait StartProcess<T>
where
T: AbstractProcess,
{
fn start(arg: T::Arg, name: Option<&str>) -> ProcessRef<T>;
fn start_config(arg: T::Arg, name: Option<&str>, config: &ProcessConfig) -> ProcessRef<T>;
fn start_link(arg: T::Arg, name: Option<&str>) -> ProcessRef<T>;
fn start_link_config(arg: T::Arg, name: Option<&str>, config: &ProcessConfig) -> ProcessRef<T>;
fn start_node(arg: T::Arg, name: Option<&str>, node: u64) -> ProcessRef<T>;
fn start_node_config(
arg: T::Arg,
name: Option<&str>,
node: u64,
config: &ProcessConfig,
) -> ProcessRef<T>;
}
impl<T> StartProcess<T> for T
where
T: AbstractProcess,
{
#[track_caller]
fn start(arg: T::Arg, name: Option<&str>) -> ProcessRef<T> {
start::<T>(arg, name, None, None, None).unwrap()
}
#[track_caller]
fn start_config(arg: T::Arg, name: Option<&str>, config: &ProcessConfig) -> ProcessRef<T> {
start::<T>(arg, name, None, Some(config), None).unwrap()
}
#[track_caller]
fn start_link(arg: T::Arg, name: Option<&str>) -> ProcessRef<T> {
start::<T>(arg, name, Some(Tag::new()), None, None).unwrap()
}
#[track_caller]
fn start_link_config(arg: T::Arg, name: Option<&str>, config: &ProcessConfig) -> ProcessRef<T> {
start::<T>(arg, name, Some(Tag::new()), Some(config), None).unwrap()
}
#[track_caller]
fn start_node(
arg: <T as AbstractProcess>::Arg,
name: Option<&str>,
node: u64,
) -> ProcessRef<T> {
start::<T>(arg, name, None, None, Some(node)).unwrap()
}
#[track_caller]
fn start_node_config(
arg: <T as AbstractProcess>::Arg,
name: Option<&str>,
node: u64,
config: &ProcessConfig,
) -> ProcessRef<T> {
start::<T>(arg, name, None, Some(config), Some(node)).unwrap()
}
}
pub trait SelfReference<T> {
fn process(&self) -> ProcessRef<T>;
}
impl<T> SelfReference<T> for T
where
T: AbstractProcess,
{
fn process(&self) -> ProcessRef<T> {
unsafe { ProcessRef::new(node_id(), process_id()) }
}
}
pub(crate) trait StartFailableProcess<T>
where
T: AbstractProcess,
{
fn start_link_or_fail(arg: T::Arg, name: Option<&str>) -> MailboxResult<(ProcessRef<T>, Tag)>;
fn start_link_config_or_fail(
arg: T::Arg,
name: Option<&str>,
config: &ProcessConfig,
) -> MailboxResult<(ProcessRef<T>, Tag)>;
}
impl<T> StartFailableProcess<T> for T
where
T: AbstractProcess,
{
fn start_link_or_fail(arg: T::Arg, name: Option<&str>) -> MailboxResult<(ProcessRef<T>, Tag)> {
let tag = Tag::new();
match start::<T>(arg, name, Some(tag), None, None) {
MailboxResult::Message(proc) => MailboxResult::Message((proc, tag)),
MailboxResult::DeserializationFailed(err) => MailboxResult::DeserializationFailed(err),
MailboxResult::TimedOut => MailboxResult::TimedOut,
MailboxResult::LinkDied(tag) => MailboxResult::LinkDied(tag),
}
}
fn start_link_config_or_fail(
arg: T::Arg,
name: Option<&str>,
config: &ProcessConfig,
) -> MailboxResult<(ProcessRef<T>, Tag)> {
let tag = Tag::new();
match start::<T>(arg, name, Some(tag), Some(config), None) {
MailboxResult::Message(proc) => MailboxResult::Message((proc, tag)),
MailboxResult::DeserializationFailed(err) => MailboxResult::DeserializationFailed(err),
MailboxResult::TimedOut => MailboxResult::TimedOut,
MailboxResult::LinkDied(tag) => MailboxResult::LinkDied(tag),
}
}
}
fn start<T>(
arg: T::Arg,
name: Option<&str>,
link: Option<Tag>,
config: Option<&ProcessConfig>,
node: Option<u64>,
) -> MailboxResult<ProcessRef<T>>
where
T: AbstractProcess,
{
let tag = if let Some(tag) = link {
tag
} else {
Tag::new()
};
let name = name.map(|name| name.to_owned());
let parent = <Process<(), Bincode>>::new(node_id(), process_id());
let process = if let Some(node) = node {
Process::<(), Bincode>::spawn_node(
node,
(parent, tag, arg, name, T::init as usize as i32),
starter::<T>,
)
} else if let Some(config) = config {
if link.is_some() {
Process::<(), Bincode>::spawn_link_config_tag(
config,
(parent, tag, arg, name, T::init as usize as i32),
tag,
starter::<T>,
)
} else {
Process::<(), Bincode>::spawn_config(
config,
(parent, tag, arg, name, T::init as usize as i32),
starter::<T>,
)
}
} else if link.is_some() {
Process::<(), Bincode>::spawn_link_tag(
(parent, tag, arg, name, T::init as usize as i32),
tag,
starter::<T>,
)
} else {
Process::<(), Bincode>::spawn(
(parent, tag, arg, name, T::init as usize as i32),
starter::<T>,
)
};
let mailbox: Mailbox<(), Bincode, Catching> = unsafe { Mailbox::new() };
match mailbox.tag_receive(&[tag]) {
MailboxResult::Message(_) => MailboxResult::Message(ProcessRef {
process,
phantom: PhantomData,
}),
MailboxResult::DeserializationFailed(err) => MailboxResult::DeserializationFailed(err),
MailboxResult::TimedOut => MailboxResult::TimedOut,
MailboxResult::LinkDied(tag) => MailboxResult::LinkDied(tag),
}
}
fn starter<T>(
(parent, tag, capture, name, entry): (Process<(), Bincode>, Tag, T::Arg, Option<String>, i32),
_: Mailbox<(), Bincode>,
) where
T: AbstractProcess,
{
let entry: fn(this: ProcessRef<T>, arg: T::Arg) -> T::State =
unsafe { std::mem::transmute(entry) };
let this = unsafe { ProcessRef::new(node_id(), process_id()) };
let name = if let Some(name) = name {
let name = format!("{} + ProcessRef + {}", name, std::any::type_name::<T>());
unsafe {
host::api::registry::put(
name.as_ptr(),
name.len(),
this.process.node_id(),
this.process.id(),
)
};
Some(name)
} else {
None
};
let mut state = entry(this, capture);
parent.tag_send(tag, ());
let mailbox: Mailbox<Sendable, Bincode, Catching> = unsafe { Mailbox::new() };
loop {
let dispatcher = mailbox.tag_receive(&[]);
match dispatcher {
MailboxResult::Message(dispatcher) => match dispatcher {
Sendable::Message(handler) => {
let handler: fn(state: &mut T::State) = unsafe { std::mem::transmute(handler) };
handler(&mut state);
}
Sendable::Request(handler, sender) => {
let handler: fn(state: &mut T::State, sender: Process<()>) =
unsafe { std::mem::transmute(handler) };
handler(&mut state, sender);
}
Sendable::Shutdown(sender) => {
let tag = unsafe { host::api::message::get_tag() };
let tag = Tag::from(tag);
T::terminate(state);
sender.tag_send(tag, ());
break;
}
},
MailboxResult::LinkDied(tag) => T::handle_link_trapped(&mut state, tag),
_ => unreachable!(),
}
}
if let Some(name) = name {
unsafe { host::api::registry::remove(name.as_ptr(), name.len()) };
}
}
pub trait Message<M, S>
where
S: Serializer<M>,
{
fn send(&self, message: M);
fn send_after(&self, message: M, duration: Duration) -> TimerRef;
}
pub trait Request<M, S>
where
S: Serializer<M>,
{
type Result;
fn request(&self, request: M) -> Self::Result {
self.request_timeout_(request, None).unwrap()
}
fn request_timeout(&self, request: M, timeout: Duration) -> MailboxResult<Self::Result> {
self.request_timeout_(request, Some(timeout))
}
#[doc(hidden)]
fn request_timeout_(
&self,
request: M,
timeout: Option<Duration>,
) -> MailboxResult<Self::Result>;
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(bound = "")]
pub struct ProcessRef<T>
where
T: ?Sized,
{
process: Process<()>,
phantom: PhantomData<T>,
}
impl<T> ProcessRef<T> {
unsafe fn new(node_id: u64, process_id: u64) -> Self {
let process = <Process<()>>::new(node_id, process_id);
ProcessRef {
process,
phantom: PhantomData,
}
}
pub fn id(&self) -> u64 {
self.process.id()
}
pub fn lookup(name: &str) -> Option<Self> {
let name = format!("{} + ProcessRef + {}", name, std::any::type_name::<T>());
let mut id = 0;
let mut node_id = 0;
let result =
unsafe { host::api::registry::get(name.as_ptr(), name.len(), &mut node_id, &mut id) };
if result == 0 {
unsafe { Some(Self::new(node_id, id)) }
} else {
None
}
}
pub fn link(&self) {
unsafe { host::api::process::link(0, self.process.id()) };
}
pub fn unlink(&self) {
unsafe { host::api::process::unlink(self.process.id()) };
}
pub fn kill(&self) {
unsafe { host::api::process::kill(self.process.id()) };
}
}
impl<T> Clone for ProcessRef<T> {
fn clone(&self) -> Self {
ProcessRef {
process: self.process,
phantom: PhantomData,
}
}
}
impl<T> ProcessRef<T>
where
T: AbstractProcess,
{
pub fn shutdown(&self) {
self.shutdown_timeout_(None).unwrap()
}
pub fn shutdown_timeout(&self, timeout: Duration) -> MailboxResult<()> {
self.shutdown_timeout_(Some(timeout))
}
fn shutdown_timeout_(&self, timeout: Option<Duration>) -> MailboxResult<()> {
let tag = Tag::new();
unsafe { host::api::message::create_data(tag.id(), 0) };
let this: Process<()> = Process::this();
Bincode::encode(&Sendable::Shutdown(this)).unwrap();
let timeout_ms = match timeout {
Some(timeout) => timeout.as_millis() as u64,
None => u64::MAX,
};
let result =
host::send_receive_skip_search(self.process.node_id(), self.process.id(), timeout_ms);
if result == 9027 {
return MailboxResult::TimedOut;
}
MailboxResult::Message(())
}
}
#[derive(serde::Serialize, serde::Deserialize)]
enum Sendable {
Message(i32),
Request(i32, Process<()>),
Shutdown(Process<()>),
}
impl<M, S, T> Message<M, S> for ProcessRef<T>
where
T: AbstractProcess,
T: MessageHandler<M, S>,
S: Serializer<M>,
{
fn send(&self, message: M) {
fn unpacker<TU, MU, SU>(this: &mut TU::State)
where
TU: MessageHandler<MU, SU>,
SU: Serializer<MU>,
{
let message: MU = SU::decode().unwrap();
<TU as MessageHandler<MU, SU>>::handle(this, message);
}
unsafe { host::api::message::create_data(Tag::none().id(), 0) };
let handler = unpacker::<T, M, S> as usize as i32;
let handler_message = Sendable::Message(handler);
Bincode::encode(&handler_message).unwrap();
S::encode(&message).unwrap();
host::send(self.process.node_id(), self.process.id());
}
fn send_after(&self, message: M, duration: Duration) -> TimerRef {
fn unpacker<TU, MU, SU>(this: &mut TU::State)
where
TU: MessageHandler<MU, SU>,
SU: Serializer<MU>,
{
let message: MU = SU::decode().unwrap();
<TU as MessageHandler<MU, SU>>::handle(this, message);
}
unsafe { host::api::message::create_data(Tag::none().id(), 0) };
let handler = unpacker::<T, M, S> as usize as i32;
let handler_message = Sendable::Message(handler);
Bincode::encode(&handler_message).unwrap();
S::encode(&message).unwrap();
let timer_id =
unsafe { host::api::timer::send_after(self.process.id(), duration.as_millis() as u64) };
TimerRef::new(timer_id)
}
}
impl<M, S, T> Request<M, S> for ProcessRef<T>
where
T: AbstractProcess,
T: RequestHandler<M, S>,
S: Serializer<M> + Serializer<<T as RequestHandler<M, S>>::Response>,
{
type Result = <T as RequestHandler<M, S>>::Response;
fn request_timeout_(
&self,
request: M,
timeout: Option<Duration>,
) -> MailboxResult<Self::Result> {
fn unpacker<TU, MU, SU>(
this: &mut TU::State,
sender: Process<<TU as RequestHandler<MU, SU>>::Response, SU>,
) where
TU: RequestHandler<MU, SU>,
SU: Serializer<MU> + Serializer<<TU as RequestHandler<MU, SU>>::Response>,
{
let message: MU = SU::decode().unwrap();
let tag = unsafe { host::api::message::get_tag() };
let tag = Tag::from(tag);
let result = <TU as RequestHandler<MU, SU>>::handle(this, message);
sender.tag_send(tag, result);
}
let tag = Tag::new();
unsafe { host::api::message::create_data(tag.id(), 0) };
let this: Process<()> = Process::new(node_id(), process_id());
let handler = unpacker::<T, M, S> as usize as i32;
let handler_message = Sendable::Request(handler, this);
Bincode::encode(&handler_message).unwrap();
S::encode(&request).unwrap();
let timeout_ms = match timeout {
Some(timeout) => timeout.as_millis() as u64,
None => u64::MAX,
};
let result =
host::send_receive_skip_search(self.process.node_id(), self.process.id(), timeout_ms);
if result == 9027 {
return MailboxResult::TimedOut;
};
MailboxResult::Message(S::decode().unwrap())
}
}
#[derive(Debug)]
pub(crate) struct Subscriber {
process: Process<(), Bincode>,
tag: Tag,
}
impl Subscriber {
pub fn new(process: Process<(), Bincode>, tag: Tag) -> Self {
Self { process, tag }
}
pub fn notify(&self) {
self.process.tag_send(self.tag, ());
}
}
impl<T> ProcessRef<T>
where
T: Supervisor,
T: AbstractProcess<State = SupervisorConfig<T>>,
{
pub fn block_until_shutdown(&self) {
fn unpacker<TU>(this: &mut TU::State, sender: Process<(), Bincode>)
where
TU: Supervisor,
TU: AbstractProcess<State = SupervisorConfig<TU>>,
{
let tag = unsafe { host::api::message::get_tag() };
let tag = Tag::from(tag);
this.subscribe_shutdown(Subscriber::new(sender, tag));
}
let tag = Tag::new();
unsafe { host::api::message::create_data(tag.id(), 0) };
let this: Process<()> = Process::this();
let handler = unpacker::<T> as usize as i32;
let handler_message = Sendable::Request(handler, this);
Bincode::encode(&handler_message).unwrap();
unsafe {
host::api::message::send_receive_skip_search(self.process.id(), 0);
};
}
}
#[derive(serde::Serialize, serde::Deserialize)]
struct GetChildren;
impl<T> RequestHandler<GetChildren> for T
where
T: Supervisor,
T: AbstractProcess<State = SupervisorConfig<T>>,
{
type Response = <<T as Supervisor>::Children as Supervisable<T>>::Processes;
fn handle(state: &mut Self::State, _: GetChildren) -> Self::Response {
state.get_children()
}
}
impl<T> ProcessRef<T>
where
T: Supervisor,
T: AbstractProcess<State = SupervisorConfig<T>>,
{
pub fn children(&self) -> <<T as Supervisor>::Children as Supervisable<T>>::Processes {
self.request(GetChildren)
}
}
impl<T> PartialEq for ProcessRef<T> {
fn eq(&self, other: &Self) -> bool {
self.process == other.process
}
}
impl<T> Eq for ProcessRef<T> {}
impl<T> std::hash::Hash for ProcessRef<T> {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
self.process.hash(state);
}
}
impl<T> std::fmt::Debug for ProcessRef<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ProcessRef")
.field("uuid", &self.id())
.finish()
}
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use lunatic_test::test;
use super::*;
use crate::sleep;
struct TestServer(i32);
#[derive(serde::Serialize, serde::Deserialize)]
struct Inc(i32);
#[derive(serde::Serialize, serde::Deserialize)]
struct Count;
#[derive(serde::Serialize, serde::Deserialize)]
struct Panic;
impl AbstractProcess for TestServer {
type Arg = ();
type State = Self;
fn init(_: ProcessRef<Self>, _: ()) -> Self {
TestServer(0)
}
}
impl MessageHandler<Inc> for TestServer {
fn handle(state: &mut Self::State, message: Inc) {
state.0 += message.0;
}
}
impl RequestHandler<Count> for TestServer {
type Response = i32;
fn handle(state: &mut Self::State, _: Count) -> Self::Response {
state.0
}
}
impl MessageHandler<Panic> for TestServer {
fn handle(_: &mut Self::State, _: Panic) {
panic!("fail");
}
}
#[test]
fn spawn_test() {
let child = TestServer::start((), None);
child.send(Inc(33));
child.send(Inc(55));
let result = child.request(Count);
assert_eq!(result, 88);
}
#[test]
#[should_panic]
fn spawn_link_test() {
let child = TestServer::start_link((), None);
child.send(Panic);
sleep(Duration::from_millis(100));
}
}