odem-rs-core 0.3.0

//! Module for basic scheduling operations that can be called as free functions.
//!
//! These operations are useful for the definition of synchronization
//! structures.

use crate::{Puck, config::Config, continuation::Continuation, ptr::Irc, simulator::Sim};

use core::{
	future::{Future, poll_fn},
	pin::Pin,
	task::{Context, Poll},
};

/// Returns a [Future] that can be awaited to reconstruct the currently active
/// [simulation context] from the [Context] provided implicitly by the future.
///
/// This function has to be awaited for technical reasons but doesn't cause any
/// scheduling operations to occur.
///
/// [simulation context]: Sim
pub fn sim<C: Config>() -> impl Future<Output = Option<Irc<Sim<C>>>> {
	poll_fn(|cx| Poll::Ready(Sim::from_context(cx)))
}

/// Suspends the currently active puck until rescheduled by some other puck.
pub fn sleep() -> impl Future<Output = ()> {
	let mut ready = false;
	poll_fn(move |_cx| {
		if ready {
			Poll::Ready(())
		} else {
			ready = true;
			Poll::Pending
		}
	})
}

/// Cedes control to the other pucks scheduled at the current model time,
/// independent of the priority of the active puck.
///
/// The active puck regains control after all the other pucks have completed
/// their transition. The model time is not advanced for the active puck.
pub fn defer() -> impl Future<Output = ()> {
	let mut ready = false;
	poll_fn(move |cx| {
		if ready {
			Poll::Ready(())
		} else {
			ready = true;
			cx.waker().wake_by_ref();
			Poll::Pending
		}
	})
}

/// Returns a [Waker] for the currently active puck that may be stored for later
/// activation.
///
/// This function has to be awaited for technical reasons but doesn't cause any
/// scheduling operations to occur.
///
/// [Waker]: core::task::Waker
pub fn waker() -> impl Future<Output = core::task::Waker> {
	poll_fn(|cx| Poll::Ready(cx.waker().clone()))
}

/// Returns a [Future] that can be awaited to advance to the termination or
/// abortion of a [Continuation] within the simulation.
#[inline]
pub const fn join<C, P>(puck: P) -> Join<C, P>
where
	C: ?Sized + Config,
	P: Puck<C> + AsRef<Continuation<'static, C>> + Unpin,
{
	Join {
		publisher: puck,
		subscriber: None,
	}
}

/// Future that suspends the currently active [Continuation] until another continuation
/// indicates completion or abortion, returning its result.
pub struct Join<C: ?Sized + Config, P: Puck<C> + AsRef<Continuation<'static, C>>> {
	/// The continuation being awaited.
	publisher: P,
	/// The reactivated continuation.
	///
	/// Is `None` before suspending and after reactivation, contains the
	/// continuation to reactivate in-between.
	subscriber: Option<Irc<Continuation<'static, C>>>,
}

impl<C, P> Drop for Join<C, P>
where
	C: ?Sized + Config,
	P: Puck<C> + AsRef<Continuation<'static, C>>,
{
	fn drop(&mut self) {
		// clean up if we're being aborted
		if let Some(active) = self.subscriber.take() {
			unsafe {
				self.publisher.as_ref().remove_pending(&active);
			}
		}
	}
}

impl<C, P> Future for Join<C, P>
where
	C: ?Sized + Config,
	P: Puck<C> + AsRef<Continuation<'static, C>>,
{
	type Output = P::Output;

	fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
		// if we've been activated, we are no longer subscribed
		self.subscriber = None;

		// check if the puck has already terminated
		match self.publisher.result() {
			Some(result) => Poll::Ready(result),
			None => {
				// determine the active continuation and subscribe if needed
				let active = self.publisher.sim().active().into_inner();
				self.subscriber = Some(active.clone());
				self.publisher.as_ref().insert_pending(active);

				// wait for reactivation
				Poll::Pending
			}
		}
	}
}