odem_rs_sync/channel/

backing.rs

//! This module provides various ring buffer implementations that can be used
//! as backing storage for message-passing channels in discrete-event
//! simulations.
//!
//! ## Buffer Types
//!
//! - **[`Rendezvous`]**: A zero-capacity buffer that enforces direct
//!   synchronization between sender and receiver, requiring both parties to be
//!   ready before a message is exchanged.
//! - **[`ArrayBuf`]**: A fixed-size, stack-allocated circular buffer with a
//!   predefined capacity, supporting efficient enqueue and dequeue operations.
//! - **[`VecDeque`]** *(requires `alloc` feature)*: A dynamically growing
//!   buffer using Rust's `VecDeque`, providing flexible storage with amortized
//!   constant-time operations.
//! - **[`BinaryHeap`]** *(requires `alloc` feature)*: A priority queue
//!   implementation for message passing with ordered retrieval of elements
//!   based on priority.
//!
//! [`VecDeque`]: alloc::collections::VecDeque
//! [`BinaryHeap`]: alloc::collections::BinaryHeap

use core::{fmt, marker::PhantomData, mem::MaybeUninit};

use super::RingBuf;

/* ******************************************************** Rendezvous Buffer */

/// A buffer that provides an empty backing store for a channel.
///
/// This leads to a direct synchronization between sender and receiver before
/// messages are exchanged.
pub struct Rendezvous<T>(PhantomData<T>);

impl<T> Rendezvous<T> {
	/// Creates a new rendezvouz buffer.
	pub const fn new() -> Self {
		Rendezvous(PhantomData)
	}
}

impl<T> Default for Rendezvous<T> {
	fn default() -> Self {
		Rendezvous::new()
	}
}

impl<T> fmt::Debug for Rendezvous<T> {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		write!(f, "Rendezvous")
	}
}

impl<T> RingBuf for Rendezvous<T> {
	type Item = T;

	fn enqueue(&mut self, item: T) -> Result<(), T> {
		Err(item)
	}

	fn dequeue(&mut self) -> Option<T> {
		None
	}

	fn clear(&mut self) {}

	fn len(&self) -> usize {
		0
	}
}

/* ************************************************** Fixed Size Array Buffer */

/// A fixed-size array buffer that can be allocated on the stack and used as a
/// backing store for channels.
pub struct ArrayBuf<const N: usize, T> {
	/// Index of the first item stored.
	off: usize,
	/// Number of elements currently held.
	len: usize,
	/// Pre-allocated space for the stored items.
	buf: [MaybeUninit<T>; N],
}

impl<const N: usize, T> ArrayBuf<N, T> {
	/// Creates a new, empty array buffer with a user-defined - but fixed -
	/// capacity.
	pub const fn new() -> Self {
		ArrayBuf {
			off: 0,
			len: 0,
			buf: [const { MaybeUninit::uninit() }; N],
		}
	}
}

impl<const N: usize, T> Default for ArrayBuf<N, T> {
	fn default() -> Self {
		ArrayBuf::new()
	}
}

impl<const N: usize, T> Drop for ArrayBuf<N, T> {
	fn drop(&mut self) {
		for i in self.off..self.off + self.len {
			unsafe { self.buf.get_unchecked_mut(i % N).assume_init_drop() }
		}
	}
}

impl<const N: usize, T> RingBuf for ArrayBuf<N, T> {
	type Item = T;

	fn enqueue(&mut self, item: T) -> Result<(), T> {
		// test if there is still space in the buffer
		if self.len < N {
			// calculate the prospective index
			let idx = (self.off + self.len) % N;

			// fill the slot with the item
			// SAFETY: there is an empty slot due to the fulfilled condition
			// above, and it has the calculated index due to the values of
			// `off` and `len`
			unsafe { self.buf.get_unchecked_mut(idx).write(item) };

			// increase the number of elements
			self.len += 1;

			// return success
			Ok(())
		} else {
			// not enough space, so return the item
			Err(item)
		}
	}

	fn dequeue(&mut self) -> Option<T> {
		// test if there are still elements in the buffer
		if self.len > 0 {
			// read the first element
			// SAFETY: the item exists due to the condition fulfilled above,
			// it has the calculated index due to the value of `off`
			let item = unsafe { self.buf.get_unchecked_mut(self.off).assume_init_read() };

			// increase the offset and wrap it around
			self.off = (self.off + 1) % N;

			// decrease the number of stored items
			self.len -= 1;

			// return the item
			Some(item)
		} else {
			// no elements: return `None`
			None
		}
	}

	fn clear(&mut self) {
		while self.len > 0 {
			// read the first element
			// SAFETY: the item exists due to the condition fulfilled above,
			// it has the calculated index due to the value of `off`
			unsafe { self.buf.get_unchecked_mut(self.off).assume_init_read() };

			// increase the offset and wrap it around
			self.off = (self.off + 1) % N;

			// decrease the number of stored items
			self.len -= 1;
		}
	}

	fn len(&self) -> usize {
		self.len
	}
}

/* ********************************************* Variably Sized Vector Buffer */

#[cfg(feature = "alloc")]
#[cfg_attr(docsrs, doc(cfg(feature = "alloc")))]
impl<T> RingBuf for alloc::collections::VecDeque<T> {
	type Item = T;

	fn enqueue(&mut self, item: T) -> Result<(), T> {
		self.push_back(item);
		Ok(())
	}

	fn dequeue(&mut self) -> Option<T> {
		Self::pop_front(self)
	}

	fn clear(&mut self) {
		Self::clear(self);
	}

	fn len(&self) -> usize {
		self.len()
	}
}

/* **************************************************** Priority Queue Buffer */

#[cfg(feature = "alloc")]
#[cfg_attr(docsrs, doc(cfg(feature = "alloc")))]
impl<T: Ord> RingBuf for alloc::collections::BinaryHeap<T> {
	type Item = T;

	fn enqueue(&mut self, item: T) -> Result<(), T> {
		self.push(item);
		Ok(())
	}

	fn dequeue(&mut self) -> Option<T> {
		self.pop()
	}

	fn clear(&mut self) {
		self.clear();
	}

	fn len(&self) -> usize {
		self.len()
	}
}