odem_rs_sync/

chain.rs

//! A module containing a data structure for intrusively linked lists of
//! [notifyable] objects that is used to construct [Chains] of [Continuations] on the
//! stack.
//!
//! [Chains]: Chain
//! [notifyable]: Subscriber
//! [Continuations]: odem_rs_core::continuation::Continuation

use core::{
	cell::Cell,
	fmt,
	marker::PhantomData,
	pin::{Pin, pin},
	task::Waker,
};
use intrusive_collections::{LinkedList, LinkedListLink, UnsafeRef, intrusive_adapter};

use crate::{Publisher, Subscriber};

/* ******************************************************************** Chain */

/// A shared list of intrusively [linked] objects that can be [notified].
///
/// [linked]: Link
/// [notified]: Subscriber
pub struct Chain<N = Waker> {
	inner: Cell<LinkedList<LinkAdapter<N>>>,
}

impl<N> Chain<N> {
	/// Creates a new, empty chain.
	pub fn new() -> Self {
		Chain {
			inner: Cell::new(LinkedList::new(LinkAdapter::NEW)),
		}
	}

	/// Removes all the queued-up objects without notifying them.
	pub fn clear(&mut self) {
		self.inner.get_mut().clear();
	}

	/// Returns `true` if this chain is empty.
	pub fn is_empty(&self) -> bool {
		let inner = self.inner.take();
		let res = inner.is_empty();
		self.inner.set(inner);
		res
	}

	/// Constructs a Notifier for the first linked object.
	///
	/// At most one object is notified once the [`go`]-method is called and the
	/// chain isn't empty, but the caller may execute additional methods that
	/// change which object is notified and what happens in that case.
	///
	/// The notified object (if any) is automatically removed from the chain.
	///
	/// [`go`]: NotifyOne::go
	pub fn notify_one(&self) -> NotifyOne<'_, N>
	where
		N: Subscriber,
	{
		NotifyOne::new(self)
	}

	/// Constructs a Notifier for all the linked objects.
	///
	/// Potentially all the objects in the chain are notified when the [`go`]
	/// method is called, but the caller may execute additional methods that
	/// change which objects are notified and what happens in that case.
	///
	/// All notified objects are automatically removed from the chain.
	///
	/// [`go`]: NotifyAll::go
	pub fn notify_all(&self) -> NotifyAll<'_, N>
	where
		N: Subscriber,
	{
		NotifyAll::new(self)
	}

	/// Removes the first element of the chain.
	///
	/// Returns `None` if the chain is empty.
	fn pop_front(&self) -> Option<UnsafeRef<Link<N>>> {
		let mut inner = self.inner.take();
		let res = inner.pop_front();
		self.inner.set(inner);
		res
	}

	/// Removes the last element of the chain.
	///
	/// Returns `None` if the chain is empty.
	fn pop_back(&self) -> Option<UnsafeRef<Link<N>>> {
		let mut inner = self.inner.take();
		let res = inner.pop_back();
		self.inner.set(inner);
		res
	}
}

// Extension methods for `Waker`-based chains.
impl Chain<Waker> {
	/// Enters the current process into a FIFO-queue, to be activated later.
	///
	/// The reactivation follows queue-ordering, i.e. first-in first-out.
	pub async fn fifo(&self) {
		use odem_rs_core::ops::{sleep, waker};

		// Construct a link containing the currently active waker.
		let link = pin!(Link::new(waker().await));

		// Subscribe to the chain in queue-order.
		let mut inner = self.inner.take();
		unsafe {
			inner.push_back(UnsafeRef::from_raw(&*link));
		}
		self.inner.set(inner);

		// Make sure to unsubscribe, even when canceled early.
		scopeguard::defer! {
			unsafe {
				self.unsubscribe(link.as_ref());
			}
		}

		// Await reactivation.
		sleep().await;
	}

	/// Enters the current process into a LIFO-queue, to be activated later.
	///
	/// The reactivation follows stack-ordering, i.e. last-in first-out.
	pub async fn lifo(&self) {
		use odem_rs_core::ops::{sleep, waker};

		// Construct a link containing the currently active waker.
		let link = pin!(Link::new(waker().await));

		// Subscribe to the chain in stack-order.
		let mut inner = self.inner.take();
		unsafe {
			inner.push_front(UnsafeRef::from_raw(&*link));
		}
		self.inner.set(inner);

		// Make sure to unsubscribe, even when panicking.
		scopeguard::defer! {
			unsafe {
				self.unsubscribe(link.as_ref());
			}
		}

		// Await reactivation.
		sleep().await;
	}
}

impl<N: Subscriber> Subscriber for Chain<N> {
	/// Notifies all the linked objects, removing them all from the chain.
	fn notify(&self) {
		self.notify_all().go();
	}
}

impl<N: Subscriber> Publisher for Chain<N> {
	type Link = Link<N>;

	unsafe fn subscribe(&self, link: Pin<&Self::Link>) {
		let mut inner = self.inner.take();
		unsafe {
			inner.push_back(UnsafeRef::from_raw(&*link));
		}
		self.inner.set(inner);
	}

	unsafe fn unsubscribe(&self, link: Pin<&Self::Link>) {
		if link.hook.is_linked() {
			let mut inner = self.inner.take();
			unsafe {
				inner.cursor_mut_from_ptr(&*link).remove();
			}
			self.inner.set(inner);
		}
	}
}

impl<N: fmt::Debug> fmt::Debug for Chain<N> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		let inner = self.inner.take();
		let res = inner.fmt(f);
		self.inner.set(inner);
		res
	}
}

impl<N> Default for Chain<N> {
	fn default() -> Self {
		Self::new()
	}
}

impl<N> Drop for Chain<N> {
	fn drop(&mut self) {
		self.clear();
	}
}

/* *********************************************************** Intrusive Link */

/// A wrapper type to augment objects that may be notified into intrusive
/// links that can be stored in intrusive collections.
#[pin_project::pin_project]
pub struct Link<N: ?Sized> {
	/// Marks the structure as non-Send and non-Sync.
	_mark: PhantomData<*const ()>,
	/// Contains the intrusive link.
	#[pin]
	hook: LinkedListLink,
	/// The object to be notified.
	pub note: N,
}

impl<N> Link<N> {
	/// Initializes a new link with the notifiable object.
	pub const fn new(waker: N) -> Self {
		Link {
			hook: LinkedListLink::new(),
			note: waker,
			_mark: PhantomData,
		}
	}

	/// Returns whether this link is in use or not.
	pub fn is_linked(&self) -> bool {
		self.hook.is_linked()
	}

	/// Strips the outer link and returns the inner `N`.
	pub fn into_inner(self) -> N {
		self.note
	}
}

impl<N> From<N> for Link<N> {
	fn from(value: N) -> Self {
		Self::new(value)
	}
}

impl<N: ?Sized + Subscriber> Subscriber for Link<N> {
	fn notify(&self) {
		self.note.notify();
	}
}

impl<N: fmt::Debug> fmt::Debug for Link<N> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		f.debug_struct("Link")
			.field("hook", &self.hook)
			.field("note", &self.note)
			.finish()
	}
}

impl fmt::Debug for Link<dyn Subscriber> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		f.debug_struct("Link").field("hook", &self.hook).finish()
	}
}

#[doc(hidden)]
mod adapter {
	#![allow(missing_docs)]
	use super::*;

	intrusive_adapter! {
		/// An adapter for the [Link]-structure, allowing it to be inserted into an
		/// intrusive [LinkedList].
		pub LinkAdapter<N> = UnsafeRef<Link<N>>: Link<N> {
			hook: LinkedListLink
		}
	}
}

pub use adapter::*;

/* ***************************************************** Notification Helpers */

/// Trait used to trigger the notification agent after the
/// notify-configuration is complete.
pub trait Trigger {
	/// The result-type of the triggered notification operation.
	type Output;

	/// The notification operation itself.
	fn trigger(self) -> Self::Output;
}

/* *************************************************************** Notify One */

/// Notification builder that seeks, notifies and removes the first fitting
/// object in a [Chain].
#[must_use = "value must be consumed with go() in order to run"]
pub struct NotifyOne<'c, N, P = (), F = (), const FIRST: bool = true> {
	/// A reference to the [Chain].
	chain: &'c Chain<N>,
	/// A boolean function that is used to skip elements in the chain that are
	/// not ready to be notified.
	filter: P,
	/// A function that is called at most once for the selected chain element.
	///
	/// This function is called before the actual notification is submitted.
	action: F,
}

impl<'c, N: Subscriber> NotifyOne<'c, N, (), ()> {
	/// Creates a notification configuration that notifies the first object in
	/// the chain.
	const fn new(chain: &'c Chain<N>) -> Self {
		NotifyOne {
			chain,
			filter: (),
			action: (),
		}
	}
}

impl<'c, N: Subscriber, P, F, const FIRST: bool> NotifyOne<'c, N, P, F, FIRST> {
	/// Configures the notification configuration with a boolean predicate that
	/// allows skipping the first sequence of objects that aren't ready to be
	/// notified.
	pub fn filter<X: Fn(&N) -> bool>(self, filter: X) -> NotifyOne<'c, N, X, F> {
		NotifyOne {
			chain: self.chain,
			filter,
			action: self.action,
		}
	}

	/// Configures what happens after a suitable object has been found for
	/// notification but before it is actually notified.
	pub fn then<X: FnOnce(&N) -> R, R>(self, action: X) -> NotifyOne<'c, N, P, X> {
		NotifyOne {
			chain: self.chain,
			filter: self.filter,
			action,
		}
	}

	/// Selects the first item in the chain to be notified.
	pub fn first(self) -> NotifyOne<'c, N, P, F, true> {
		NotifyOne {
			chain: self.chain,
			filter: self.filter,
			action: self.action,
		}
	}

	/// Selects the last item in the chain to be notified.
	pub fn last(self) -> NotifyOne<'c, N, P, F, false> {
		NotifyOne {
			chain: self.chain,
			filter: self.filter,
			action: self.action,
		}
	}

	/// Triggers the notification.
	pub fn go(self) -> <Self as Trigger>::Output
	where
		Self: Trigger,
	{
		Trigger::trigger(self)
	}
}

impl<N, P, F, R, const FIRST: bool> Trigger for NotifyOne<'_, N, P, F, FIRST>
where
	N: Subscriber,
	P: Fn(&N) -> bool,
	F: FnOnce(&N) -> R,
{
	type Output = Option<R>;

	fn trigger(self) -> Self::Output {
		// take a local copy of the chain and replace it with an empty list
		let mut chain = self.chain.inner.replace(LinkedList::new(LinkAdapter::NEW));

		// get a mutable cursor for iteration and removal
		let mut cur = if FIRST {
			chain.front_mut()
		} else {
			chain.back_mut()
		};
		let mut out = None;

		// iterate over the elements
		while let Some(link) = cur.get() {
			// check the predicate
			if (self.filter)(&link.note) {
				// found the element: call the function, notify and remove
				let res = (self.action)(&link.note);
				link.notify();
				cur.remove();

				// store the result and break off the loop
				out = Some(res);
				break;
			}

			// move to the next/prev element
			if FIRST {
				cur.move_next();
			} else {
				cur.move_prev();
			}
		}

		// restore the chain
		self.chain.inner.replace(chain);

		out
	}
}

impl<N, P, const FIRST: bool> Trigger for NotifyOne<'_, N, P, (), FIRST>
where
	N: Subscriber,
	P: Fn(&N) -> bool,
{
	type Output = ();

	fn trigger(self) -> Self::Output {
		// take a local copy of the chain and replace it with an empty list
		let mut chain = self.chain.inner.replace(LinkedList::new(LinkAdapter::NEW));

		// get a mutable cursor for iteration and removal
		let mut cur = if FIRST {
			chain.front_mut()
		} else {
			chain.back_mut()
		};

		// iterate over the elements
		while let Some(link) = cur.get() {
			// check the predicate
			if (self.filter)(&link.note) {
				// found the element: notify and remove
				link.notify();
				cur.remove();
				break;
			}

			// move to the next/prev element
			if FIRST {
				cur.move_next();
			} else {
				cur.move_prev();
			}
		}

		// restore the chain
		self.chain.inner.replace(chain);
	}
}

impl<N, F, R, const FIRST: bool> Trigger for NotifyOne<'_, N, (), F, FIRST>
where
	N: Subscriber,
	F: FnOnce(&N) -> R,
{
	type Output = Option<R>;

	fn trigger(self) -> Self::Output {
		// immediately select the first link to notifiy
		if FIRST {
			self.chain.pop_front()
		} else {
			self.chain.pop_back()
		}
		.map(move |link| {
			// run the provided closure before notifying
			let res = (self.action)(&link.note);
			link.notify();
			res
		})
	}
}

impl<N, const FIRST: bool> Trigger for NotifyOne<'_, N, (), (), FIRST>
where
	N: Subscriber,
{
	type Output = ();

	fn trigger(self) -> Self::Output {
		// immediately select the first link to notify
		if let Some(link) = if FIRST {
			self.chain.pop_front()
		} else {
			self.chain.pop_back()
		} {
			link.notify();
		}
	}
}

/* *************************************************************** Notify All */

/// Notification builder that locates, notifies and removes all fitting objects
/// in a [Chain].
#[must_use = "value must be consumed with go() in order to run"]
pub struct NotifyAll<'c, N, P = (), F = (), const REV: bool = false> {
	/// A reference to the [Chain].
	chain: &'c Chain<N>,
	/// A boolean function that is used to skip elements that are not ready to
	/// be notified.
	filter: P,
	/// A function that is called for every fitting element immediately prior
	/// to its notification.
	///
	/// This function may be used to aggregate data over the sequence of fitting
	/// objects.
	action: F,
}

impl<'c, N: Subscriber> NotifyAll<'c, N, (), ()> {
	/// Creates a notification configuration that notifies all the objects
	/// contained in the chain.
	///
	/// All successfully notified objects are removed from the chain.
	const fn new(chain: &'c Chain<N>) -> Self {
		NotifyAll {
			chain,
			filter: (),
			action: (),
		}
	}
}

impl<'c, N: Subscriber, P, F, const REV: bool> NotifyAll<'c, N, P, F, REV> {
	/// Configures the notification configuration with a boolean predicate that
	/// allows skipping the objects that aren't ready to be notified.
	pub fn filter<X: Fn(&N) -> bool>(self, filter: X) -> NotifyAll<'c, N, X, F> {
		NotifyAll {
			chain: self.chain,
			filter,
			action: self.action,
		}
	}

	/// Configures what happens for every suitable object that has been found
	/// for notification right before it is actually notified.
	pub fn then<X: FnMut(&N)>(self, action: X) -> NotifyAll<'c, N, P, X> {
		NotifyAll {
			chain: self.chain,
			filter: self.filter,
			action,
		}
	}

	/// Triggers the notification.
	pub fn go(self) -> <Self as Trigger>::Output
	where
		Self: Trigger,
	{
		Trigger::trigger(self)
	}
}

impl<'c, N: Subscriber, P, F> NotifyAll<'c, N, P, F, true> {
	/// Reverses the order when triggering.
	pub fn rev(self) -> NotifyAll<'c, N, P, F, false> {
		NotifyAll {
			chain: self.chain,
			filter: self.filter,
			action: self.action,
		}
	}
}

impl<'c, N: Subscriber, P, F> NotifyAll<'c, N, P, F, false> {
	/// Reverses the order when triggering.
	pub fn rev(self) -> NotifyAll<'c, N, P, F, true> {
		NotifyAll {
			chain: self.chain,
			filter: self.filter,
			action: self.action,
		}
	}
}

impl<N, P, F, const REV: bool> Trigger for NotifyAll<'_, N, P, F, REV>
where
	N: Subscriber,
	P: Fn(&N) -> bool,
	F: FnMut(&N),
{
	type Output = ();

	fn trigger(mut self) -> Self::Output {
		// take a local copy of the chain and replace it with an empty list
		let mut chain = self.chain.inner.replace(LinkedList::new(LinkAdapter::NEW));

		// get a mutable cursor for iteration and removal
		let mut cur = if REV {
			chain.back_mut()
		} else {
			chain.front_mut()
		};

		// iterate over the elements
		while let Some(link) = cur.get() {
			// check the predicate
			if (self.filter)(&link.note) {
				// located an element: call the function, notify and remove
				(self.action)(&link.note);
				link.notify();
				cur.remove();
			} else if !REV {
				cur.move_next();
			}

			if REV {
				cur.move_prev();
			}
		}

		// restore the chain
		self.chain.inner.replace(chain);
	}
}

impl<N, P, const REV: bool> Trigger for NotifyAll<'_, N, P, (), REV>
where
	N: Subscriber,
	P: Fn(&N) -> bool,
{
	type Output = ();

	fn trigger(self) -> Self::Output {
		// take a local copy of the chain and replace it with an empty list
		let mut chain = self.chain.inner.replace(LinkedList::new(LinkAdapter::NEW));

		// get a mutable cursor for iteration and removal
		let mut cur = if REV {
			chain.back_mut()
		} else {
			chain.front_mut()
		};

		// iterate over the elements
		while let Some(link) = cur.get() {
			// check the predicate
			if (self.filter)(&link.note) {
				// located an element: remove and notify
				cur.remove().unwrap().notify();
			} else if !REV {
				cur.move_next();
			}

			if REV {
				cur.move_prev();
			}
		}

		// restore the chain
		self.chain.inner.replace(chain);
	}
}

impl<N, F, const REV: bool> Trigger for NotifyAll<'_, N, (), F, REV>
where
	N: Subscriber,
	F: FnMut(&N),
{
	type Output = ();

	fn trigger(mut self) -> Self::Output {
		while let Some(link) = if REV {
			self.chain.pop_back()
		} else {
			self.chain.pop_front()
		} {
			(self.action)(&link.note);
			link.notify();
		}
	}
}

impl<N: Subscriber, const REV: bool> Trigger for NotifyAll<'_, N, (), (), REV> {
	type Output = ();

	fn trigger(self) -> Self::Output {
		while let Some(link) = if REV {
			self.chain.pop_back()
		} else {
			self.chain.pop_front()
		} {
			link.notify();
		}
	}
}

/* *********************************************************** Waker Notifier */

impl Subscriber for Waker {
	fn notify(&self) {
		self.wake_by_ref();
	}
}