odem_rs_sync/

channel.rs

//! This module contains support for MPMC-channels which synchronize sender and
//! receiver continuations over a message queue.

use core::{
	cell::{Cell, UnsafeCell},
	task::Waker,
};

#[cfg(feature = "tracing")]
use tracing::trace;

use crate::{
	Subscriber,
	chain::Chain,
	error::{SendError, TryRecvError, TrySendError},
};

#[cfg(feature = "alloc")]
#[cfg_attr(docsrs, doc(cfg(feature = "alloc")))]
pub mod shared;

pub mod pinned;

pub mod backing;

pub use pinned::{Channel, ChannelExt};

pub use backing::{ArrayBuf, Rendezvous};

#[cfg(feature = "alloc")]
pub use shared::channel;

/* ******************************************************** Ring Buffer Trait */

/// The trait that qualifies a buffer to be used as a backing store for
/// [Channels](Channel).
pub trait RingBuf {
	/// The type of item stored inside the ring buffer.
	type Item;

	/// Places an item into the buffer, returning an `Ok` result if successful
	/// and the item wrapped in an `Err` result if the buffer is full.
	fn enqueue(&mut self, item: Self::Item) -> Result<(), Self::Item>;

	/// Extracts an item from the buffer, returning `None` if the buffer is
	/// empty.
	fn dequeue(&mut self) -> Option<Self::Item>;

	/// Removes all stored elements from the buffer.
	fn clear(&mut self);

	/// Returns the number of elements that are currently stored in the buffer.
	fn len(&self) -> usize;

	/// Returns whether the buffer is currently empty.
	fn is_empty(&self) -> bool {
		self.len() == 0
	}
}

/* ****************************************************** Channel Inner Parts */

/// Helper structure to allow the unsizing coercion to function in stable Rust.
///
/// The reason this indirection is needed is that unsizing doesn't work if the
/// unsized template argument is referred to more than once in the struct.
/// Using this structure allows us to separate the associated type of the
/// [`RingBuf`]-trait from the potentially unsized ring buffer.
struct Inner<T, R: ?Sized + RingBuf<Item = T>> {
	/// A chain of continuations waiting for reactivation as soon as there is a
	/// message to send or to receive.
	chain: Chain<OneShot<T, Waker>>,
	/// The number of continuations waiting to receive or to send a message.
	/// Negative values indicate continuations waiting to receive a message,
	/// positive values indicate continuations waiting to send a message,
	/// and 0 indicates that no continuations are waiting to send or receive
	/// messages.
	pending: Cell<isize>,
	/// A ring buffer containing messages to be received. May have capacity `0`,
	/// in which case the corresponding channel is a rendezvous channel.
	buffer: UnsafeCell<R>,
}

impl<R: ?Sized + RingBuf> Inner<R::Item, R> {
	/// Creates a new channel that uses the backing store passed as an argument.
	fn new(buffer: R) -> Self
	where
		R: Sized,
	{
		Self {
			chain: Chain::new(),
			pending: Cell::new(0),
			buffer: UnsafeCell::new(buffer),
		}
	}

	/// Closes the channel causing any waiting sender or receiver to return
	/// immediately with an error indicating a disconnect. Any additional send
	/// operation after the channel is closed fails immediately. Receive
	/// operations fail once the message buffer is empty.
	fn close(&self) {
		if !self.is_closed() {
			#[cfg(feature = "tracing")]
			trace!("closing channel");

			self.pending.set(isize::MIN);
			self.chain.notify_all().go();
		}
	}

	/// Returns whether this channel is closed or not.
	fn is_closed(&self) -> bool {
		self.pending.get() == isize::MIN
	}

	fn add_pending_sender(&self) {
		self.pending.set(self.pending.get() + 1);
	}

	fn sub_pending_sender(&self) {
		self.pending.set(self.pending.get() - 1);
	}

	fn add_pending_receiver(&self) {
		self.sub_pending_sender();
	}

	fn sub_pending_receiver(&self) {
		self.add_pending_sender();
	}

	fn has_pending_sender(&self) -> bool {
		self.pending.get() > 0
	}

	fn has_pending_receiver(&self) -> bool {
		self.pending.get() < 0
	}

	/// Resets the channel by clearing the message buffer and re-opening it if
	/// it has been closed before. Panics if there are still [send](Self::send)
	/// or [recv](Self::recv) operations pending.
	fn reset(&self) {
		// reset the number of pending senders/receivers
		match self.pending.get() {
			0 | isize::MIN => self.pending.set(0),
			n => panic!(
				"resetting the channel failed because there are {} pending \
				 {}-operations",
				n.unsigned_abs(),
				if n > 0 { "send" } else { "recv" }
			),
		}

		// clear the message buffer
		unsafe { &mut *self.buffer.get() }.clear();
	}

	/// Attempt to send the value without waiting.
	///
	/// This can fail for two reasons:
	/// 1. The channel is closed, i.e. [`close`] has been called.
	/// 2. The message buffer is full.
	///
	/// The `Err`-result contains the initially provided item in either case.
	///
	/// [`close`]: Channel::close
	fn try_send(&self, item: R::Item) -> Result<(), TrySendError<R::Item>> {
		// test if there are receivers waiting for a message
		if self.has_pending_receiver() {
			// test if the channel is still open
			if !self.is_closed() {
				// decrease the number of pending messages
				self.pending.set(self.pending.get() + 1);

				#[cfg(feature = "tracing")]
				trace!(
					remaining = self.pending.get().unsigned_abs(),
					"passing message to the first receiver"
				);

				// transmit the item to the first receiver
				self.chain
					.notify_one()
					.then(|c| c.give(item))
					.go()
					.expect("receiver waiting for message");

				Ok(())
			} else {
				// closed: don't accept any more messages
				Err(TrySendError::Disconnected(item)).inspect_err(|_err| {
					#[cfg(feature = "tracing")]
					trace!("message cannot be sent: {_err}");
				})
			}
		} else {
			// insert the message into the buffer
			self.give(item)
				.map_err(TrySendError::Full)
				.inspect(|_| {
					#[cfg(feature = "tracing")]
					trace!(
						buffer_len = self.buffer_len(),
						"message buffered in channel"
					);
				})
				.inspect_err(|_err| {
					#[cfg(feature = "tracing")]
					trace!("message cannot be sent: {_err}");
				})
		}
	}

	/// Attempts to receive a value of the channel without waiting.
	///
	/// This can fail for two reasons:
	/// 1. The channel is closed, i.e. [`close`] has been called.
	/// 2. The message buffer is empty.
	///
	/// [`close`]: Channel::close
	fn try_recv(&self) -> Result<R::Item, TryRecvError> {
		// attempt to extract the next message from the underlying buffer
		match self.take() {
			Some(msg) => {
				#[cfg(feature = "tracing")]
				trace!(
					buffer_len = self.buffer_len(),
					"extracted message from the channel's buffer"
				);

				// test if there are senders waiting to transmit a message
				if self.has_pending_sender() {
					// receive a message from the longest waiting sender
					let item = unsafe {
						self.chain
							.notify_one()
							.then(OneShot::take)
							.go()
							.unwrap_unchecked()
							.unwrap_unchecked()
					};

					// decrease the number of waiting senders
					self.sub_pending_sender();

					// add the message to the buffer
					self.give(item)
						.ok()
						.expect("buffer should not be full after an element has been extracted");
				}

				// return the message extracted from the buffer
				Ok(msg)
			}
			_ => {
				if self.has_pending_sender() {
					// decrease the number of waiting senders
					self.sub_pending_sender();

					#[cfg(feature = "tracing")]
					trace!(
						remaining = self.pending.get(),
						"received message from a waiting sender"
					);

					let Some(Some(msg)) = self.chain.notify_one().then(OneShot::take).go() else {
						unreachable!("pending senders but no message")
					};

					Ok(msg)
				} else {
					if self.is_closed() {
						Err(TryRecvError::Disconnected)
					} else {
						Err(TryRecvError::Empty)
					}
					.inspect_err(|_err| {
						#[cfg(feature = "tracing")]
						trace!("message could not be received: {_err}");
					})
				}
			}
		}
	}

	/// Returns the number of elements currently queued to be received.
	///
	/// This number can be negative if there are [`RecvFuture`] currently
	/// waiting to receive messages.
	fn count(&self) -> isize {
		self.buffer_len() as isize + self.pending.get()
	}

	/// Returns the number of elements currently stored in the channel's buffer.
	fn buffer_len(&self) -> usize {
		unsafe { &*self.buffer.get() }.len()
	}

	/// Places an item into the internal message buffer. If the buffer is
	/// already full, the item is returned to the caller through the `Err`
	/// variant.
	fn give(&self, item: R::Item) -> Result<(), R::Item> {
		unsafe { &mut *self.buffer.get() }.enqueue(item)
	}

	/// Takes the first item out of the message buffer. Returns `None` if the
	/// buffer is empty.
	fn take(&self) -> Option<R::Item> {
		unsafe { &mut *self.buffer.get() }.dequeue()
	}
}

/* ****************************************************** OneShot Signal Type */

/// Wrapper for [Subscriber] objects that allow transmission of a single
/// value as part of the notification.
struct OneShot<T, N> {
	/// The value being transmitted.
	payload: Cell<Option<T>>,
	/// The object to be notified when the value has been sent or received.
	notify: N,
}

impl<T, N> OneShot<T, N> {
	/// Creates a new signal with the payload either left empty or initialized
	/// with some value.
	pub const fn new(notify: N, payload: Option<T>) -> Self {
		OneShot {
			payload: Cell::new(payload),
			notify,
		}
	}

	/// Removes the payload and returns it.
	pub fn take(&self) -> Option<T> {
		self.payload.replace(None)
	}

	/// Overwrites the current payload with an item.
	pub fn give(&self, item: T) {
		self.payload.set(Some(item));
	}
}

impl<T, N: Subscriber> Subscriber for OneShot<T, N> {
	fn notify(&self) {
		self.notify.notify();
	}
}