odem_rs_sync/

control.rs

//! This module contains support for awaiting arbitrary state events in the form
//! of control expressions.
//!
//! Control expressions are types that implement the [Publisher] and [Expr]
//! trait, which allows them to [calculate] a value and [notify] dependent
//! continuations of changes in their value.
//! Once a continuation has subscribed, it is automatically activated by the control
//! expression whenever its internal state changes.
//!
//! This is used for implementing a mechanism that allows processes to await
//! specific conditions regarding the system state, which is made
//! ergonomic through the [until!]-macro.
//!
//! The simplest control expressions are [Control]-variables that capture a
//! single, [copyable] value and automatically notify subscribed continuations when it
//! changes.
//!
//! [calculate]: Expr::get
//! [notify]: Subscriber::notify
//! [until!]: crate::until
//! [copyable]: Copy

use core::{
	cell::Cell,
	fmt,
	future::{Future, IntoFuture},
	pin::Pin,
	task::Waker,
	task::{Context, Poll},
};

use intrusive_collections::UnsafeRef;

use crate::{
	Publisher, Subscriber,
	chain::{Chain, Link},
};

pub mod imp;

/* ******************************************************* Control Expression */

/// An object-safe trait for repeatedly evaluating a complex expression.
pub trait Expr {
	/// The result type.
	type Output;

	/// The routine used for evaluation.
	fn get(&self) -> Self::Output;
}

/* ********************************************************* Control Variable */

/// Marks a variable as a state variable which can be used in conjunction with
/// the [until!]-macro.
///
/// Whenever a control variable changes its value, the runtime system checks
/// whether any other parts of the model are currently waiting for the variable
/// to attain a certain value.
///
/// [until!]: crate::until!
#[derive(Default)]
pub struct Control<T: ?Sized> {
	/// A [Chain] of waiting continuations.
	queue: Chain<UnsafeRef<Predicate>>,
	/// The inner value.
	value: Cell<T>,
}

impl<T> Control<T> {
	/// Creates a new control variable and initializes its value.
	#[inline]
	pub fn new(value: T) -> Self {
		Self {
			value: Cell::new(value),
			queue: Chain::new(),
		}
	}

	/// Assigns a new value to the control variable.
	///
	/// This can lead to potential activations of waiting processes.
	pub fn set(&self, val: T) {
		self.value.set(val);
		self.notify();
	}

	/// Returns a copy of the stored inner value.
	#[inline]
	pub fn get(&self) -> T
	where
		T: Copy,
	{
		self.value.get()
	}

	/// Swaps the values of two control variables. The difference to
	/// [`core::mem::swap`] is that this function doesn't require an exclusive
	/// reference.
	pub fn swap(&self, other: &Self) {
		self.value.swap(&other.value);
		self.notify();
	}

	/// Replaces the current value of the control variable with a new one and
	/// returns it.
	///
	/// This can lead to potential activations of waiting processes.
	pub fn replace(&self, val: T) -> T {
		let res = self.value.replace(val);
		self.notify();
		res
	}

	/// Updates the contained value using a function and returns the new value.
	#[inline]
	pub fn update<F>(&self, f: F)
	where
		F: FnOnce(T) -> T,
		T: Copy,
	{
		self.set(f(self.get()))
	}

	/// Takes the value of the control variable, leaving `Default::default()` in
	/// its place.
	#[inline]
	pub fn take(&self) -> T
	where
		T: Default,
	{
		self.replace(T::default())
	}

	/// Returns a mutable reference to the underlying data.
	///
	/// This call borrows the control variable mutably (at compile-time) which
	/// guarantees that we possess the only reference.
	#[inline]
	pub fn get_mut(&mut self) -> &mut T {
		self.value.get_mut()
	}

	/// Assigns a new value to the control variable **without notifying pending
	/// control conditions**.
	///
	/// Use this method if you want to change the inner value without scheduling
	/// continuations that depend on it. Once finished, you may call [notify] directly.
	///
	/// [notify]: Self::notify
	#[inline]
	pub fn unchecked_set(&self, val: T) {
		self.value.set(val);
	}
}

impl<T: ?Sized> Publisher for Control<T> {
	type Link = Link<UnsafeRef<Predicate>>;

	#[inline]
	unsafe fn subscribe(&self, link: Pin<&Self::Link>) {
		unsafe {
			self.queue.subscribe(link);
		}
	}

	#[inline]
	unsafe fn unsubscribe(&self, link: Pin<&Self::Link>) {
		unsafe {
			self.queue.unsubscribe(link);
		}
	}
}

impl<T: ?Sized> Subscriber for Control<T> {
	#[inline]
	fn notify(&self) {
		// notify the set of waiting continuations with fulfilled conditions
		self.queue.notify_all().filter(Expr::get).go();
	}
}

impl<T: Copy> Expr for Control<T> {
	type Output = T;

	#[inline]
	fn get(&self) -> T {
		self.value.get()
	}
}

impl<T: fmt::Debug> fmt::Debug for Control<T> {
	#[inline]
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		f.debug_tuple("Control")
			.field(unsafe { &*self.value.as_ptr() })
			.finish()
	}
}

impl<T: fmt::Display> fmt::Display for Control<T> {
	#[inline]
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		unsafe { &*self.value.as_ptr() }.fmt(f)
	}
}

/* ******************************************************* Control Expression */

/// Container for complex control expressions that combines a set of [Control]
/// variables with a closure that depends on the values of the control set.
#[derive(Copy, Clone)]
pub struct ControlExpr<C, F> {
	/// The set of (potential) control variables that the closure depends on.
	depends: C,
	/// A function closure accepting the current values of the control variables
	/// and calculating the result.
	closure: F,
}

impl<C, F> ControlExpr<C, F>
where
	C: Publisher,
	F: Fn(&C) -> bool,
{
	/// Creates a new control expression given a set of [Control] variables and
	/// a closure that depends on them.
	#[inline]
	pub const fn new(ctrl: C, code: F) -> Self {
		ControlExpr {
			depends: ctrl,
			closure: code,
		}
	}
}

impl<C, F> Publisher for ControlExpr<C, F>
where
	C: Publisher,
{
	type Link = C::Link;

	#[inline]
	unsafe fn subscribe(&self, link: Pin<&Self::Link>) {
		unsafe {
			self.depends.subscribe(link);
		}
	}

	#[inline]
	unsafe fn unsubscribe(&self, link: Pin<&Self::Link>) {
		unsafe {
			self.depends.unsubscribe(link);
		}
	}
}

impl<C, F, R> Expr for ControlExpr<C, F>
where
	F: Fn(&C) -> R,
{
	type Output = R;

	fn get(&self) -> R {
		(self.closure)(&self.depends)
	}
}

impl<C, F> IntoFuture for ControlExpr<C, F>
where
	C: Publisher,
	F: Fn(&C) -> bool,
	C::Link: From<UnsafeRef<Predicate>>,
{
	type Output = ();
	type IntoFuture = Until<C, F>;

	#[inline]
	fn into_future(self) -> Self::IntoFuture {
		Until::new(self)
	}
}

/* ************************************************************* Until Future */

/// Notify dispatcher that includes the boolean condition being awaited.
///
/// The basic idea is to reduce the number of spurious reactivations of waiting
/// continuations by checking if the condition being awaited evaluated to `true` for the
/// specific state-change triggering the re-evaluation and not re-awakening the
/// continuation if it doesn't.
///
/// Even if it does, the re-awakened continuation still needs to check the condition
/// since its truth value may have changed between the scheduling and the
/// activation, but this technique still has the potential to cut down on a lot
/// of superfluous scheduling operations.
#[pin_project::pin_project(PinnedDrop)]
pub struct Until<C: Publisher, F> {
	/// A predicate of the boolean condition.
	#[pin]
	poll: Predicate<ControlExpr<C, F>>,
	/// Space for the link needed during polling.
	#[pin]
	link: Option<C::Link>,
}

impl<C: Publisher, F> Until<C, F> {
	/// Creates a new future that blocks on being awaited until the passed
	/// condition is fulfilled.
	#[inline]
	pub const fn new(cond: ControlExpr<C, F>) -> Self {
		Until {
			poll: Predicate {
				note: Cell::new(None),
				cond,
			},
			link: None,
		}
	}
}

impl<C, F> Future for Until<C, F>
where
	C: Publisher,
	F: Fn(&C) -> bool,
	C::Link: From<UnsafeRef<Predicate>>,
{
	type Output = ();

	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
		// test if the condition evaluates to true and terminate if it does
		if self.poll.get() {
			return Poll::Ready(());
		}

		let mut projection = self.project();

		// clone the waker
		projection.poll.note.set(Some(cx.waker().clone()));

		// unsubscribe from the control set
		// SAFETY: we will only unsubscribe if we have been subscribed before
		// using the original link
		if let Some(link) = projection.link.as_ref().as_pin_ref() {
			unsafe {
				projection.poll.cond.unsubscribe(link);
			}
		}

		// fill the option with the link
		// SAFETY: the lifetime-extension is justified because we are sure to
		// unsubscribe during drop; drop happens because of pin
		projection.link.set(Some(
			unsafe {
				UnsafeRef::from_raw(&*core::mem::transmute::<
					Pin<&Predicate<dyn Expr<Output = bool> + '_>>,
					Pin<&Predicate<dyn Expr<Output = bool> + 'static>>,
				>(projection.poll.as_ref()))
			}
			.into(),
		));

		// subscribe to the control set
		// SAFETY: pinning guarantees that the link's address is stable
		// until this future drops, which will also unsubscribe the link
		unsafe {
			projection
				.poll
				.cond
				.subscribe(projection.link.as_ref().as_pin_ref().unwrap_unchecked());
		}

		// await the next reactivation
		Poll::Pending
	}
}

#[pin_project::pinned_drop]
impl<C: Publisher, F> PinnedDrop for Until<C, F> {
	#[inline]
	fn drop(self: Pin<&mut Self>) {
		// unsubscribe from a control set
		if let Some(link) = self.as_ref().project_ref().link.as_pin_ref() {
			unsafe { self.poll.cond.unsubscribe(link) }
		}
	}
}

/// A boolean predicate with an attached [Waker] to notify.
pub struct Predicate<C: ?Sized = dyn Expr<Output = bool>> {
	/// The [Waker] to notify.
	note: Cell<Option<Waker>>,
	/// The boolean condition as a dynamically dispatched trait.
	cond: C,
}

impl<C: Expr<Output = bool> + ?Sized> Expr for Predicate<C> {
	type Output = bool;

	#[inline]
	fn get(&self) -> bool {
		self.cond.get()
	}
}

impl<C: ?Sized> Subscriber for Predicate<C> {
	#[inline]
	fn notify(&self) {
		// only wake the process once
		if let Some(waker) = self.note.take() {
			waker.wake();
		}
	}
}

/* *************************************************************** Expr Impls */

// implement Expr for primitive data types
macro_rules! impl_primitive_expr {
	($($L:ty),*) => {$(
		impl Expr for $L {
			type Output = $L;

			#[inline(always)]
			fn get(&self) -> $L { *self }
		}
	)*};
}

impl_primitive_expr!(
	(),
	bool,
	char,
	f32,
	f64,
	u8,
	u16,
	u32,
	u64,
	u128,
	usize,
	i8,
	i16,
	i32,
	i64,
	i128,
	isize
);

impl<T: ?Sized + Expr> Expr for &'_ T {
	type Output = T::Output;

	#[inline(always)]
	fn get(&self) -> Self::Output {
		(**self).get()
	}
}

impl<T: ?Sized + Expr> Expr for UnsafeRef<T> {
	type Output = T::Output;

	#[inline(always)]
	fn get(&self) -> Self::Output {
		(**self).get()
	}
}

/* ******************************************************************* Macros */

#[cfg(test)]
mod tests {
	use super::{Control, imp::*};
	use crate::{Publisher, until};
	use core::pin::pin;
	use odem_rs_core::{
		job::Job,
		simulator::{Sim, Simulator},
	};
	use std::boxed::Box;

	#[derive(Publisher)]
	struct Foo(#[subscribe] Control<bool>);

	impl Foo {
		fn is_false(&self) -> bool {
			!self.0.get()
		}
	}

	#[test]
	fn simple_control_expr() {
		async fn sim_main(sim: &Sim) -> f64 {
			until!(true).await;
			sim.now()
		}

		let time = Simulator::default()
			.run(sim_main)
			.expect("no errors during execution");

		assert_eq!(time, 0.0);
	}

	#[test]
	fn complex_control_expr() {
		async fn sim_main(sim: &Sim) -> f64 {
			let t1 = false;
			let t2 = Foo(Control::new(false));
			let t3 = Control::new(false);
			let t4 = Box::new(Control::new(false));
			let t5 = &t4;

			until!(t1 || t2.is_false() || t3 || t4 || t5).await;
			sim.now()
		}

		let time = Simulator::default()
			.run(sim_main)
			.expect("no errors during execution");
		assert_eq!(time, 0.0);
	}

	#[test]
	fn simple_scenario() {
		async fn sim_main(sim: &Sim) -> f64 {
			let t1 = Control::new(false);
			let job = pin!(Job::new(async {
				sim.advance(1.0).await;
				t1.set(true);
			}));

			sim.activate(job);
			until!(t1).await;
			sim.now()
		}

		let time = Simulator::default()
			.run(sim_main)
			.expect("no errors during execution");
		assert_eq!(time, 1.0);
	}

	#[test]
	fn complex_scenario() {
		async fn sim_main(sim: &Sim) -> f64 {
			let t1 = Control::new(0);
			let t2 = Control::new(false);

			let job = pin!(Job::new(async {
				loop {
					sim.advance(1.0).await;
					t1.update(|inner| inner + 1);
					t2.update(|inner| !inner);
				}
			}));

			sim.activate(job);
			until!(t1 > 5 && t2).await;
			sim.now()
		}

		let time = Simulator::default()
			.run(sim_main)
			.expect("no errors during execution");
		assert_eq!(time, 7.0);
	}
}