ya_ring_buf/

lib.rs

//! Yet Another Ring Buffer.
//!
//! SPSC channel backed by a ring buffer.
//!
//! Features include:
//!
//! * Lock free, non-blocking push and pop.
//! * If desired, can integrate blocking, futures-based, or custom waiting. If undesired, no-op waiters add
//!   no overhead.
//! * `veiw` functions for safely accessing the buffer directly, ex. for doing IO directly from
//!   or to the buffer.
//! * `no_std` compatible (currently requires `alloc`)
#![cfg_attr(not(feature = "std"), no_std)]

extern crate alloc;

#[cfg(any(test, doctest))]
mod tests;
pub mod waiter;

use alloc::alloc::Layout;
#[cfg(not(loom))]
use alloc::sync::Arc;
#[cfg(not(loom))]
use core::sync::atomic::{
	AtomicUsize,
	Ordering,
};
use core::{
	mem::MaybeUninit,
	ops::Range,
	ptr::NonNull,
};

#[cfg(loom)]
use loom::sync::{
	atomic::{
		AtomicUsize,
		Ordering,
	},
	Arc,
};
use waiter::OptionalWaiter;

/// Creates a new ring buffer with the specified capacity and no waiters.
///
/// Returns the producer and consumer ends. The buffer will not have any capability to wait
/// for the other end - i.e. non-blocking operations only.
///
/// See [`with_waiters`] for more documentation.
///
/// # Panics
///
/// If capacity is zero.
pub fn new_waitless<T>(
	capacity: usize,
) -> (RingBufferProducer<T, (), ()>, RingBufferConsumer<T, (), ()>) {
	with_waiters(capacity, (), ())
}

/// Creates a new ring buffer with the specified capacity and blocking waiters.
///
/// Returns the producer and consumer ends. The wait methods will block until
/// available.
///
/// See [`with_waiters`] for more documentation.
///
/// Requires the 'std' feature.
///
/// # Panics
///
/// If capacity is zero.
#[cfg(feature = "std")]
pub fn new_blocking<T>(
	capacity: usize,
) -> (
	RingBufferProducer<T, waiter::BlockingWaiter, waiter::BlockingWaiter>,
	RingBufferConsumer<T, waiter::BlockingWaiter, waiter::BlockingWaiter>,
) {
	with_waiters(
		capacity,
		waiter::BlockingWaiter::new(),
		waiter::BlockingWaiter::new(),
	)
}

/// Creates a new ring buffer with the specified capacity and asynchronous waiters.
///
/// Returns the producer and consumer ends. The wait methods will return a future that
/// will be fulfilled when there are items or slots available.
///
/// See [`with_waiters`] for more documentation.
///
/// Requires the 'async' feature.
///
/// # Panics
///
/// If capacity is zero.
#[cfg(feature = "async")]
pub fn new_async<T>(
	capacity: usize,
) -> (
	RingBufferProducer<T, waiter::AsyncWaiter, waiter::AsyncWaiter>,
	RingBufferConsumer<T, waiter::AsyncWaiter, waiter::AsyncWaiter>,
) {
	with_waiters(
		capacity,
		waiter::AsyncWaiter::new(),
		waiter::AsyncWaiter::new(),
	)
}

/// Creates a new ring buffer with the specified capacity and specified waiters.
///
/// Returns the producer and consumer ends. The ends may be sent to other threads,
/// but because most methods take `&mut self` for safety, they cannot be shared by
/// threads.
///
/// If one side of the buffer is dropped, the other side's operations will eventually
/// fail with a [`ClosedError`] (immediately for producers, and after the existing items
/// have been drained for consumers).
///
/// Each end has a waiter, which abstracts how each buffer end should notify the
/// other end that items or slots are available and it may resume activity.
/// This crate includes two such waiters: [`BlockingWaiter`](waiter::BlockingWaiter),
/// which blocks the current thread, and [`AsyncWaiter`](waiter::AyncWaiter), which
/// returns a future. Custom implementations may do other things - for example, signal
/// an mio waker. If no waiter is needed, the unit type `()` can be used, which removes
/// the `wait` methods and adds no overhead to the buffer.
///
/// # Panics
///
/// If capacity is zero.
pub fn with_waiters<T, PW: OptionalWaiter, CW: OptionalWaiter>(
	mut capacity: usize,
	producer_waiter: PW,
	consumer_waiter: CW,
) -> (RingBufferProducer<T, PW, CW>, RingBufferConsumer<T, PW, CW>) {
	assert!(capacity != 0, "Cannot create a 0 capacity ring buffer");
	// The ring buffer can't fill the last slot, so add 1 to the capacity.
	capacity = capacity.saturating_add(1);
	let layout = Layout::array::<T>(capacity).unwrap();
	let buf = if core::mem::size_of::<T>() == 0 {
		NonNull::dangling()
	} else {
		unsafe {
			NonNull::new(alloc::alloc::alloc(layout) as *mut T)
				.unwrap_or_else(|| alloc::alloc::handle_alloc_error(layout))
		}
	};

	let shared = Arc::new(RingBufferShared::<T, PW, CW> {
		buf,
		cap: capacity,
		head: AtomicUsize::new(0),
		tail: AtomicUsize::new(0),
		closed_count: AtomicUsize::new(0),
		producer_waiter,
		consumer_waiter,
		_ph: core::marker::PhantomData::default(),
	});

	let producer = RingBufferProducer {
		shared: shared.clone(),
	};
	let consumer = RingBufferConsumer { shared };
	(producer, consumer)
}

/// Creates a new ring buffer with the specified capacity and `Default::default()` for waiters.
///
/// This is equivalent to `with_waiters(capacity, Default::default(), Default::default())`.
///
/// Returns the producer and consumer ends.
///
/// See [`with_waiters`] for more documentation.
///
/// # Panics
///
/// If capacity is zero.
pub fn with_default_waiters<T, PW: OptionalWaiter + Default, CW: OptionalWaiter + Default>(
	capacity: usize,
) -> (RingBufferProducer<T, PW, CW>, RingBufferConsumer<T, PW, CW>) {
	with_waiters(capacity, Default::default(), Default::default())
}

/// Shared state
struct RingBufferShared<T, PW: OptionalWaiter, CW: OptionalWaiter> {
	buf: NonNull<T>,
	cap: usize,
	/// Next slot to read
	head: AtomicUsize,
	/// Next slot to write
	tail: AtomicUsize,
	// non-zero if one of the sides has been dropped.
	// TODO: Maybe see if we can remove this? Can't just use Arc::strong_count,
	// because we need to notify the waiter after we drop the count, which may race
	// with another thread dropping.
	closed_count: AtomicUsize,
	producer_waiter: PW,
	consumer_waiter: CW,
	_ph: core::marker::PhantomData<T>,
}
impl<T, PW: OptionalWaiter, CW: OptionalWaiter> RingBufferShared<T, PW, CW> {
	fn head_tail(&self) -> (usize, usize) {
		let head = self.head.load(Ordering::Acquire);
		let tail = self.tail.load(Ordering::Acquire);
		debug_assert!(head < self.cap);
		debug_assert!(tail < self.cap);
		(head, tail)
	}

	unsafe fn consume(&self, head: usize, tail: usize, amnt: usize, drop: bool) {
		let (new_head, [drop1, drop2], _) = self.incr_head(head, tail, amnt);
		debug_assert!(new_head < self.cap);
		let set_on_drop = SetOnDrop(&self.head, new_head, &self.producer_waiter);

		if drop && core::mem::needs_drop::<T>() {
			for i in drop1 {
				self.buf.as_ptr().add(i).drop_in_place();
			}
			for i in drop2 {
				self.buf.as_ptr().add(i).drop_in_place();
			}
		}

		core::mem::drop(set_on_drop);
	}

	/// Increment the head pointer by an amount, clamped to the available capacity.
	///
	/// Returns:
	///
	/// 1. The new head pointer.
	/// 2. Ranges of the indices for slots between the old and new head pointers, i.e. slots requested for reading.
	/// 3. Ranges of the indices for slots after the new head pointer but before the tail pointer, i.e. remaining
	///    slots to read.
	fn incr_head(
		&self,
		head: usize,
		tail: usize,
		amnt: usize,
	) -> (usize, [Range<usize>; 2], [Range<usize>; 2]) {
		debug_assert!(head < self.cap);
		debug_assert!(tail < self.cap);

		if head <= tail {
			// No wrapping, limited by tail
			let new_head = head.saturating_add(amnt).min(tail);
			return (new_head, [head..new_head, 0..0], [new_head..tail, 0..0]);
		}

		if head.saturating_add(amnt) < self.cap {
			// Not enough to wrap
			let new_head = head + amnt;
			return (
				new_head,
				[head..new_head, 0..0],
				[new_head..self.cap, 0..tail],
			);
		}

		// Wrap
		let new_head = tail.min(amnt - (self.cap - head));
		return (
			new_head,
			[head..self.cap, 0..new_head],
			[new_head..tail, 0..0],
		);
	}

	/// Increment the tail pointer by an amount, clamped to the available capacity.
	///
	/// Returns:
	///
	/// 1. The new tail pointer.
	/// 2. Ranges of the indices for slots before the new tail pointer, i.e. slots that would be made available.
	/// 3. Ranges of the indices for slots after the new tail pointer but before the head pointer, i.e.
	///    slots that could still be written.
	fn incr_tail(
		&self,
		head: usize,
		tail: usize,
		amnt: usize,
	) -> (usize, [Range<usize>; 2], [Range<usize>; 2]) {
		debug_assert!(head < self.cap);
		debug_assert!(tail < self.cap);

		if head == 0 {
			// Head at zero, keep last slot open and don't wrap.
			let new_tail = tail.saturating_add(amnt).min(self.cap - 1);
			return (
				new_tail,
				[tail..new_tail, 0..0],
				[new_tail..self.cap - 1, 0..0],
			);
		}

		if head <= tail && tail.saturating_add(amnt) >= self.cap {
			// Wrap around. Head != 0.
			let new_tail = (amnt - (self.cap - tail)).min(head - 1);
			return (
				new_tail,
				[tail..self.cap, 0..new_tail],
				[new_tail..head - 1, 0..0],
			);
		}

		if head <= tail {
			// No wrapping, limited by cap
			return (
				tail + amnt,
				[tail..tail + amnt, 0..0],
				[tail + amnt..self.cap, 0..head - 1],
			);
		}

		// head > tail, limited by head
		let new_tail = (tail + amnt).min(head - 1);
		return (new_tail, [tail..new_tail, 0..0], [new_tail..head - 1, 0..0]);
	}
}
unsafe impl<T: Send, PW: OptionalWaiter, CW: OptionalWaiter> Send for RingBufferShared<T, PW, CW> {}
unsafe impl<T: Send, PW: OptionalWaiter, CW: OptionalWaiter> Sync for RingBufferShared<T, PW, CW> {}
impl<T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::Drop for RingBufferShared<T, PW, CW> {
	fn drop(&mut self) {
		if core::mem::needs_drop::<T>() {
			#[cfg(loom)]
			let (head, tail) = self.head_tail();
			#[cfg(not(loom))]
			let (head, tail) = (*self.head.get_mut(), *self.tail.get_mut());
			unsafe {
				self.consume(head, tail, usize::MAX, true);
			}
		}

		if core::mem::size_of::<T>() != 0 {
			let layout = Layout::array::<T>(self.cap).unwrap();
			unsafe {
				alloc::alloc::dealloc(self.buf.as_ptr() as *mut _, layout);
			}
		}
	}
}

impl<T, PW: OptionalWaiter, CW: OptionalWaiter> core::fmt::Debug for RingBufferShared<T, PW, CW> {
	fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
		f.debug_struct("RingBufferShared")
			.field("buf", &self.buf)
			.field("cap", &self.cap)
			.field("head", &self.head)
			.field("tail", &self.tail)
			.finish()
	}
}

/// Producer side of the ring buffer.
pub struct RingBufferProducer<T, PW: OptionalWaiter, CW: OptionalWaiter> {
	shared: Arc<RingBufferShared<T, PW, CW>>,
}
impl<T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::Drop for RingBufferProducer<T, PW, CW> {
	fn drop(&mut self) {
		let v = self.shared.closed_count.fetch_add(1, Ordering::Release);
		debug_assert!(v <= 1);
		self.shared.consumer_waiter.notify();
	}
}

impl<T, PW: OptionalWaiter, CW: OptionalWaiter> RingBufferProducer<T, PW, CW> {
	/// Pushes a single item into the ring buffer.
	///
	/// If there was enough space, returns `Ok(())`.
	///
	/// If the buffer is full, returns `PushError::Full(v)`. If the receiver
	/// had been dropped, returns `PushError::Closed(v)`. The passed-in value
	/// is returned so that pushing may be attempted later.
	pub fn push(&mut self, v: T) -> Result<(), PushError<T>> {
		self.extend(core::iter::once(v))
			.map_err(|ClosedError(mut it)| PushError::Closed(it.next().unwrap()))
			.and_then(|mut it| it.next().map(|v| Err(PushError::Full(v))).unwrap_or(Ok(())))
	}

	/// Pushes a sequence of items into the ring buffer.
	///
	/// Will stop when the iterator runs out of items or the ring buffer
	/// runs out of space, whichever comes first. Returns the iterator
	/// with the remaining items. It's recommended to use a `FusedIterator`.
	///
	/// In the case of the iterator panicing, already-pushed items may not
	/// appear to the consumer, and may leak instead.
	///
	/// This can be slightly more efficient than pushing individually, as the
	/// atomic operations for updating the counters will not update until all
	/// the items are pushed. If `T` is `Copy`, consider using [`RingBufferProducer::push_slice`]
	/// instead.
	///
	/// Returns `Err(ClosedError(it))` if the receiver had been dropped,
	/// returning the iterator without touching it.
	pub fn extend<I: Iterator<Item = T>>(&mut self, mut it: I) -> Result<I, ClosedError<I>> {
		if self.shared.closed_count.load(Ordering::Acquire) > 0 {
			return Err(ClosedError(it));
		}

		let (head, mut tail) = self.shared.head_tail();
		let limit = head.checked_sub(1).unwrap_or(self.shared.cap - 1);
		if tail == limit {
			return Ok(it);
		}

		// TODO: if next panics, objects that did get added will leak. Should drop them.
		while tail != limit {
			let v = match it.next() {
				Some(v) => v,
				None => break,
			};

			unsafe {
				self.shared.buf.as_ptr().add(tail).write(v);
			}
			tail = (tail + 1) % self.shared.cap;
		}

		debug_assert!(tail < self.shared.cap);
		self.shared.tail.store(tail, Ordering::Release);
		self.shared.consumer_waiter.notify();

		Ok(it)
	}

	/// Gets a view into the currently writable portion of this ring buffer.
	///
	/// This returns a wrapper over two slices, which, when concatenated, form the
	/// available space in order. The user should write items to the slices in order,
	/// then call [`UninitWriteView::produce`] to make them available to the consumer.
	///
	/// As opposed to [`RingBufferProducer::view`], this does not do any initialization
	/// of the available slots, and so the returned slices have elements of `MaybeUninit<T>`.
	/// Its up to the user to write to the slots via `MaybeUninit::write` and call
	/// `UninitWriteView::produce` with the correct amount.
	///
	/// This can be used, for example, to read bytes directly to the ring buffer, via
	/// [`std::io::ReadBuf`], without having to use another buffer and paying for the
	/// overhead of initializing the values.
	pub fn uninitialized_view(&mut self) -> Result<UninitWriteView<T, PW, CW>, ClosedError<()>> {
		if self.shared.closed_count.load(Ordering::Acquire) > 0 {
			return Err(ClosedError(()));
		}

		let (head, tail) = self.shared.head_tail();

		let (_, _, [slice1, slice2]) = self.shared.incr_tail(head, tail, 0);
		let slices = unsafe {
			(
				core::slice::from_raw_parts_mut(
					(self.shared.buf.as_ptr() as *mut MaybeUninit<T>).add(slice1.start),
					slice1.end - slice1.start,
				),
				core::slice::from_raw_parts_mut(
					(self.shared.buf.as_ptr() as *mut MaybeUninit<T>).add(slice2.start),
					slice2.end - slice2.start,
				),
			)
		};

		Ok(UninitWriteView {
			slices,
			shared: &*self.shared,
			head,
			tail,
		})
	}

	/// Gets how many items that can be written.
	///
	/// If the consumer side is being used on another thread, the actual amount
	/// may increase after this function has returned.
	pub fn available_len(&self) -> usize {
		let (head, tail) = self.shared.head_tail();
		let (_, _, [slice1, slice2]) = self.shared.incr_tail(head, tail, 0);
		(slice1.end - slice1.start) + (slice2.end - slice2.start)
	}

	/// Returns the maximum number of elements that the buffer can hold.
	pub fn capacity(&self) -> usize {
		self.shared.cap - 1
	}
}
impl<T: Default, PW: OptionalWaiter, CW: OptionalWaiter> RingBufferProducer<T, PW, CW> {
	/// Gets a view into the currently writable portion of this ring buffer.
	///
	/// This returns a wrapper over two slices, which, when concatenated, form the
	/// available space in order. The user should write items to the slices in order,
	/// then call [`UninitWriteView::produce`] to make them available to the consumer.
	///
	/// As opposed to [`RingBufferProducer::uninitialized_view`], this initializes
	/// each element with `Default::default`, meaning there will be no uninitialized
	/// values in the buffer and the interface can be used in safe code.
	///
	/// This can be used, for example, to read bytes directly to the ring buffer, via
	/// [`std::io::Read::read`], without having to use another buffer.
	pub fn view(&mut self) -> Result<WriteView<T, PW, CW>, ClosedError<()>> {
		self.sized_view(usize::MAX)
	}

	/// Gets a view into the currently writable portion of this ring buffer, limited by a size.
	///
	/// Works similar to [`RingBufferProducer::view`], but caps the number of elements in the view
	/// to the specified number. This can be used to avoid default-initializing an excessive number of
	/// elements.
	pub fn sized_view(&mut self, size: usize) -> Result<WriteView<T, PW, CW>, ClosedError<()>> {
		if self.shared.closed_count.load(Ordering::Acquire) > 0 {
			return Err(ClosedError(()));
		}

		let (head, tail) = self.shared.head_tail();

		let (_, [slice1, slice2], _) = self.shared.incr_tail(head, tail, size);

		unsafe {
			for i in slice1.clone() {
				self.shared.buf.as_ptr().add(i).write(Default::default());
			}
			for i in slice2.clone() {
				self.shared.buf.as_ptr().add(i).write(Default::default());
			}
		}

		let slices = unsafe {
			(
				core::slice::from_raw_parts_mut(
					self.shared.buf.as_ptr().add(slice1.start),
					slice1.end - slice1.start,
				),
				core::slice::from_raw_parts_mut(
					self.shared.buf.as_ptr().add(slice2.start),
					slice2.end - slice2.start,
				),
			)
		};

		Ok(WriteView {
			slices,
			shared: &*self.shared,
			head,
			tail,
		})
	}
}
impl<T: Copy, PW: OptionalWaiter, CW: OptionalWaiter> RingBufferProducer<T, PW, CW> {
	/// Pushes a slice of copyable values.
	///
	/// This memcpy's the slice into the buffer, so it should be
	/// more efficient than [`RingBufferProducer::push`] or [`RingBufferProducer::extend`].
	///
	/// Returns the amount of items pushed, which may be less than the length
	/// of the slice is the buffer fills up.
	///
	/// If the receiver had been dropped, returns `Err(ClosedError(()))` without
	/// touching the values slice.
	pub fn push_slice(&mut self, values: &[T]) -> Result<usize, ClosedError<()>> {
		if self.shared.closed_count.load(Ordering::Acquire) > 0 {
			return Err(ClosedError(()));
		}
		let (head, tail) = self.shared.head_tail();

		let (new_tail, [range1, range2], _) = self.shared.incr_tail(head, tail, values.len());
		debug_assert!((range1.end - range1.start) + (range2.end - range2.start) <= values.len());
		unsafe {
			self.shared
				.buf
				.as_ptr()
				.add(range1.start)
				.copy_from_nonoverlapping(values.as_ptr(), range1.end - range1.start);
			self.shared
				.buf
				.as_ptr()
				.add(range2.start)
				.copy_from_nonoverlapping(
					values[range1.end - range1.start..].as_ptr(),
					range2.end - range2.start,
				);
		}

		debug_assert!(tail < self.shared.cap);
		self.shared.tail.store(new_tail, Ordering::Release);
		Ok((range1.end - range1.start) + (range2.end - range2.start))
	}
}
impl<T, PW: OptionalWaiter, CW: OptionalWaiter> RingBufferProducer<T, PW, CW>
where
	PW: for<'a> waiter::Waiter<'a>,
{
	/// Waits for space to become available in the buffer, or for the consumer to be closed.
	///
	/// This method is only available if the buffer was created with a producer waker.
	///
	/// How the wait happens is up to the waiter - it may block, return a future, etc.
	/// However, it should return immediately if space is currently available.
	pub fn wait<'a>(&'a mut self) -> <PW as waiter::Waiter<'a>>::Future {
		let checker = waiter::WakeCheck {
			head: &self.shared.head,
			tail: &self.shared.tail,
			closed_count: &self.shared.closed_count,
			cap: self.shared.cap,
			is_producer: true,
		};
		self.shared.producer_waiter.wait(checker)
	}

	/// Gets the waiter.
	///
	/// Can be used to alter settings of the waiter, for example, by setting a timeout.
	pub fn waiter(&mut self) -> &PW {
		&self.shared.producer_waiter
	}
}

/// Consumer side of the ring buffer.
pub struct RingBufferConsumer<T, PW: OptionalWaiter, CW: OptionalWaiter> {
	shared: Arc<RingBufferShared<T, PW, CW>>,
}
impl<T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::Drop for RingBufferConsumer<T, PW, CW> {
	fn drop(&mut self) {
		let v = self.shared.closed_count.fetch_add(1, Ordering::Release);
		debug_assert!(v <= 1);
		self.shared.producer_waiter.notify();
	}
}

impl<T, PW: OptionalWaiter, CW: OptionalWaiter> RingBufferConsumer<T, PW, CW> {
	/// Tries to pop an item from the buffer.
	///
	/// Returns `Ok(Some(_))` if the buffer had an item, `Ok(None)` if
	/// the buffer was empty but the sender is still available, or `Err(ClosedError(()))`
	/// if the buffer is empty and the sender had been dropped.
	pub fn pop(&mut self) -> Result<Option<T>, ClosedError<()>> {
		let (head, tail) = self.shared.head_tail();
		if head == tail {
			if self.shared.closed_count.load(Ordering::Acquire) > 0 {
				return Err(ClosedError(()));
			}
			return Ok(None);
		}
		let v = unsafe { self.shared.buf.as_ptr().add(head).read() };
		unsafe {
			self.shared.consume(head, tail, 1, false);
		}
		Ok(Some(v))
	}

	/// Gets a view into the currently readable portion of this ring buffer.
	///
	/// This returns a wrapper over two slices, which, when concatenated, form the
	/// available space in order. The user should read the items, then call
	/// [`ReadView::consume`] or [`ReadView::consume_all`] to remove the items
	/// from the buffer.
	///
	/// If the buffer is empty and the sender had been dropped, returns
	/// `Err(ClosedError(()))`.
	///
	/// This can be used, for example, to write bytes directly from the ring buffer, via
	/// [`std::io::Write::write_all`], without having to use another buffer.
	pub fn view(&mut self) -> Result<ReadView<T, PW, CW>, ClosedError<()>> {
		let (head, tail) = self.shared.head_tail();

		let (_, _, [slice1, slice2]) = self.shared.incr_head(head, tail, 0);

		if slice1.end - slice1.start == 0 {
			debug_assert_eq!(slice1.end - slice1.start, 0);
			if self.shared.closed_count.load(Ordering::Acquire) > 0 {
				return Err(ClosedError(()));
			}
		}

		let slices = unsafe {
			(
				core::slice::from_raw_parts_mut(
					self.shared.buf.as_ptr().add(slice1.start),
					slice1.end - slice1.start,
				),
				core::slice::from_raw_parts_mut(
					self.shared.buf.as_ptr().add(slice2.start),
					slice2.end - slice2.start,
				),
			)
		};

		Ok(ReadView {
			slices,
			shared: &*self.shared,
			head,
			tail,
		})
	}

	/// Gets how many items that can be read.
	///
	/// If the producer side is being used on another thread, the actual amount
	/// may increase after this function has returned.
	pub fn available_len(&self) -> usize {
		let (head, tail) = self.shared.head_tail();
		let (_, _, [slice1, slice2]) = self.shared.incr_head(head, tail, 0);
		(slice1.end - slice1.start) + (slice2.end - slice2.start)
	}

	/// Returns the maximum number of elements that the buffer can hold.
	pub fn capacity(&self) -> usize {
		self.shared.cap - 1
	}
}
impl<T, PW: OptionalWaiter, CW: OptionalWaiter> RingBufferConsumer<T, PW, CW>
where
	CW: for<'a> waiter::Waiter<'a>,
{
	/// Waits for an item to be produced into the buffer, or for the producer to be closed.
	///
	/// This method is only available if the buffer was created with a producer waker.
	///
	/// How the wait happens is up to the waiter - it may block, return a future, etc.
	/// Ideally, it should return immediately if space is already available.
	pub fn wait<'a>(&'a mut self) -> <CW as waiter::Waiter<'a>>::Future {
		let checker = waiter::WakeCheck {
			head: &self.shared.head,
			tail: &self.shared.tail,
			closed_count: &self.shared.closed_count,
			cap: self.shared.cap,
			is_producer: false,
		};
		self.shared.consumer_waiter.wait(checker)
	}

	/// Gets the waiter.
	///
	/// Can be used to alter settings of the waiter, for example, by setting a timeout.
	pub fn waiter(&mut self) -> &CW {
		&self.shared.consumer_waiter
	}
}
impl<T: Copy, PW: OptionalWaiter, CW: OptionalWaiter> RingBufferConsumer<T, PW, CW> {
	/// Pops as many items as possible, copying them into a destination slice.
	///
	/// Returns the amount of items copied, which may be zero.
	pub fn pop_to_slice(&mut self, dest: &mut [T]) -> Result<usize, ClosedError<()>> {
		let v = self.view()?;
		let s1amnt = dest.len().min(v.0.len());
		let s2amnt = v.1.len().min(dest.len() - s1amnt);
		dest[..s1amnt].copy_from_slice(&v.0[..s1amnt]);
		dest[s1amnt..s1amnt + s2amnt].copy_from_slice(&v.1[..s2amnt]);
		v.consume(s1amnt + s2amnt);
		Ok(s1amnt + s2amnt)
	}
}

/// Ring buffer available contents, as two slices.
///
/// The two slices are similar to how `VecDeque::to_slices` work -  when concatenated, they
/// contain the available contents of the buffer at the time of calling.
///
/// After the contents have been examined, you may call `consume` or `consume_all` to remove the items
/// from the buffer and free up their slots for the producer.
pub struct ReadView<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> {
	pub slices: (&'a mut [T], &'a mut [T]),
	shared: &'a RingBufferShared<T, PW, CW>,
	head: usize,
	tail: usize,
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::Deref for ReadView<'a, T, PW, CW> {
	type Target = (&'a mut [T], &'a mut [T]);

	fn deref(&self) -> &Self::Target {
		&self.slices
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::DerefMut
	for ReadView<'a, T, PW, CW>
{
	fn deref_mut(&mut self) -> &mut Self::Target {
		&mut self.slices
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> ReadView<'a, T, PW, CW> {
	/// Consumes the first `amnt` items from the buffer, dropping them and
	/// freeing the slots for the producer to fill.
	///
	/// `amnt` is clamped to the number of items in this view.
	pub fn consume(self, amnt: usize) {
		let Self {
			slices: _,
			shared,
			head,
			tail,
		} = self;
		unsafe { shared.consume(head, tail, amnt, true) }
	}

	/// Consumes all items in this view from the buffer.
	pub fn consume_all(self) {
		self.consume(usize::MAX);
	}

	/// Iterates through each available element, in order.
	///
	/// Equivalent to chaining the two slice's iterators.
	pub fn iter(&mut self) -> core::iter::Chain<core::slice::IterMut<T>, core::slice::IterMut<T>> {
		let i1 = self.slices.0.iter_mut();
		let i2 = self.slices.1.iter_mut();
		i1.chain(i2)
	}

	/// Checks if the view is empty.
	///
	/// Equivalent to testing if the first slices is empty. If the first slice is empty,
	/// the second one will be as well.
	pub fn is_empty(&self) -> bool {
		self.slices.0.is_empty()
	}

	/// Gets how many items are in the view.
	///
	/// This is the sum of the lengths of the two slices.
	pub fn len(&self) -> usize {
		self.slices.0.len() + self.slices.1.len()
	}

	/// Gets an item from the view, at the `i`'th place, or `None` if out of bounds.
	///
	/// Mimics `slice::get`.
	pub fn get(&self, i: usize) -> Option<&T> {
		if i > self.len() {
			None
		} else if i > self.slices.0.len() {
			Some(&self.slices.1[i - self.0.len()])
		} else {
			Some(&self.slices.0[i])
		}
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::Index<usize>
	for ReadView<'a, T, PW, CW>
{
	type Output = T;

	fn index(&self, i: usize) -> &Self::Output {
		if i > self.slices.0.len() {
			&self.slices.1[i - self.0.len()]
		} else {
			&self.slices.0[i]
		}
	}
}

/// View into the writable portion of the buffer.
pub struct WriteView<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> {
	pub slices: (&'a mut [T], &'a mut [T]),
	shared: &'a RingBufferShared<T, PW, CW>,
	head: usize,
	tail: usize,
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::Deref for WriteView<'a, T, PW, CW> {
	type Target = (&'a mut [T], &'a mut [T]);

	fn deref(&self) -> &Self::Target {
		&self.slices
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::DerefMut
	for WriteView<'a, T, PW, CW>
{
	fn deref_mut(&mut self) -> &mut Self::Target {
		&mut self.slices
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> WriteView<'a, T, PW, CW> {
	/// Produces the specified number of items, making them available to the receiver.
	///
	/// The remaining elements are dropped.
	///
	/// `amnt` is clamped to the number of items in this view.
	pub fn produce(self, amnt: usize) {
		let len = self.len();
		let amnt = amnt.min(len);

		let shared = self.shared;
		let head = self.head;
		let tail = self.tail;
		core::mem::forget(self);

		let (new_tail, _, _) = shared.incr_tail(head, tail, amnt);
		debug_assert!(new_tail < shared.cap);
		shared.tail.store(new_tail, Ordering::Release);
		shared.consumer_waiter.notify();

		let (_, [drop1, drop2], _) = shared.incr_tail(head, new_tail, len - amnt);
		debug_assert!(new_tail < shared.cap);
		if core::mem::needs_drop::<T>() {
			unsafe {
				for i in drop1 {
					shared.buf.as_ptr().add(i).drop_in_place();
				}
				for i in drop2 {
					shared.buf.as_ptr().add(i).drop_in_place();
				}
			}
		}
	}

	/// Iterates through each available slot, in order.
	///
	/// Equivalent to chaining the two slice's iterators.
	pub fn iter(&mut self) -> core::iter::Chain<core::slice::IterMut<T>, core::slice::IterMut<T>> {
		let i1 = self.slices.0.iter_mut();
		let i2 = self.slices.1.iter_mut();
		i1.chain(i2)
	}

	/// Checks if the view is empty.
	///
	/// Equivalent to testing if the first slices is empty. If the first slice is empty,
	/// the second one will be as well.
	pub fn is_empty(&self) -> bool {
		self.slices.0.is_empty()
	}

	/// Gets the length of the view.
	///
	/// This is the sum of the lengths of both slices.
	pub fn len(&self) -> usize {
		self.0.len() + self.1.len()
	}

	/// Gets an item from the view, at the `i`'th place, or `None` if out of bounds.
	///
	/// Mimics `slice::get_mut`.
	pub fn get(&mut self, i: usize) -> Option<&mut T> {
		if i > self.len() {
			None
		} else if i > self.slices.0.len() {
			Some(&mut self.slices.1[i - self.0.len()])
		} else {
			Some(&mut self.slices.0[i])
		}
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::Drop for WriteView<'a, T, PW, CW> {
	fn drop(&mut self) {
		let len = self.len();
		let (_, [drop1, drop2], _) = self.shared.incr_tail(self.head, self.tail, len);
		if core::mem::needs_drop::<T>() {
			unsafe {
				for i in drop1 {
					self.shared.buf.as_ptr().add(i).drop_in_place();
				}
				for i in drop2 {
					self.shared.buf.as_ptr().add(i).drop_in_place();
				}
			}
		}
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::Index<usize>
	for WriteView<'a, T, PW, CW>
{
	type Output = T;

	fn index(&self, i: usize) -> &Self::Output {
		if i > self.slices.0.len() {
			&self.slices.1[i - self.0.len()]
		} else {
			&self.slices.0[i]
		}
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::IndexMut<usize>
	for WriteView<'a, T, PW, CW>
{
	fn index_mut(&mut self, i: usize) -> &mut T {
		if i > self.slices.0.len() {
			&mut self.slices.1[i - self.0.len()]
		} else {
			&mut self.slices.0[i]
		}
	}
}

/// View into the writable portion of the buffer, with uninitialized elements.
///
/// Using this is unsafe, as an accurate number of items produces must be given,
/// otherwise this will produce uninitialized values to the consumer.
pub struct UninitWriteView<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> {
	pub slices: (&'a mut [MaybeUninit<T>], &'a mut [MaybeUninit<T>]),
	shared: &'a RingBufferShared<T, PW, CW>,
	head: usize,
	tail: usize,
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::Deref
	for UninitWriteView<'a, T, PW, CW>
{
	type Target = (&'a mut [MaybeUninit<T>], &'a mut [MaybeUninit<T>]);

	fn deref(&self) -> &Self::Target {
		&self.slices
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::DerefMut
	for UninitWriteView<'a, T, PW, CW>
{
	fn deref_mut(&mut self) -> &mut Self::Target {
		&mut self.slices
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> UninitWriteView<'a, T, PW, CW> {
	/// Produces the specified number of items, making them available to the receiver.
	///
	/// The remaining elements are assumed to be uninitialzed and will not be accessed.
	///
	/// `amnt` is clamped to the number of items in this view.
	///
	/// # Safety
	///
	/// The elements being produced must have been initialized.
	pub unsafe fn produce(self, amnt: usize) {
		let Self {
			slices: _,
			shared,
			head,
			tail,
		} = self;

		let (new_tail, _, _) = shared.incr_tail(head, tail, amnt);

		debug_assert!(new_tail < self.shared.cap);
		self.shared.tail.store(new_tail, Ordering::Release);
	}

	/// Iterates through each available slot, in order.
	///
	/// Equivalent to chaining the two slice's iterators.
	pub fn iter(
		&mut self,
	) -> core::iter::Chain<core::slice::IterMut<MaybeUninit<T>>, core::slice::IterMut<MaybeUninit<T>>>
	{
		let i1 = self.slices.0.iter_mut();
		let i2 = self.slices.1.iter_mut();
		i1.chain(i2)
	}

	/// Checks if the view is empty.
	///
	/// Equivalent to testing if the first slices is empty. If the first slice is empty,
	/// the second one will be as well.
	pub fn is_empty(&self) -> bool {
		self.slices.0.is_empty()
	}

	/// Gets the length of the view.
	///
	/// This is the sum of the lengths of both slices.
	pub fn len(&self) -> usize {
		self.0.len() + self.1.len()
	}

	/// Gets an item from the view, at the `i`'th place, or `None` if out of bounds.
	///
	/// Mimics `slice::get_mut`.
	pub fn get(&mut self, i: usize) -> Option<&mut MaybeUninit<T>> {
		if i > self.len() {
			None
		} else if i > self.slices.0.len() {
			Some(&mut self.slices.1[i - self.0.len()])
		} else {
			Some(&mut self.slices.0[i])
		}
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::Index<usize>
	for UninitWriteView<'a, T, PW, CW>
{
	type Output = MaybeUninit<T>;

	fn index(&self, i: usize) -> &Self::Output {
		if i > self.slices.0.len() {
			&self.slices.1[i - self.0.len()]
		} else {
			&self.slices.0[i]
		}
	}
}
impl<'a, T, PW: OptionalWaiter, CW: OptionalWaiter> core::ops::IndexMut<usize>
	for UninitWriteView<'a, T, PW, CW>
{
	fn index_mut(&mut self, i: usize) -> &mut MaybeUninit<T> {
		if i > self.slices.0.len() {
			&mut self.slices.1[i - self.0.len()]
		} else {
			&mut self.slices.0[i]
		}
	}
}

/// Struct for ensuring an AtomicUsize is always set and a waiter is notified,
/// even if panicking.
struct SetOnDrop<'a, W: OptionalWaiter>(&'a AtomicUsize, usize, &'a W);
impl<'a, W: OptionalWaiter> core::ops::Drop for SetOnDrop<'a, W> {
	fn drop(&mut self) {
		self.0.store(self.1, Ordering::Release);
		self.2.notify()
	}
}

/// Error produced when trying to do an operation but the other side is closed.
///
/// For [`RingBufferProducer`], most functions will start returning this error
/// as soon as the receiver has been dropped. The error may contain the value
/// that was being pushed, so that the sender can do something with it.
///
/// For [`RingBufferConsumer`], most functions will start returning this error
/// only after all the data in the buffer has been consumed and it is now empty.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClosedError<T>(pub T);
impl<T> core::fmt::Display for ClosedError<T> {
	fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
		f.write_str("Other side of buffer dropped")
	}
}

/// Error produced while trying to push a single value to the buffer.
///
/// The error contains the value that was being pushed, so that the sender
/// can try again later.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PushError<T> {
	/// Buffer is full.
	Full(T),
	/// Receiver was dropped. Equivalent to [`ClosedError`].
	Closed(T),
}
impl<T> core::fmt::Display for PushError<T> {
	fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
		match self {
			Self::Full(_) => f.write_str("Buffer full"),
			Self::Closed(_) => f.write_str("Other side of buffer dropped"),
		}
	}
}
impl<T> From<ClosedError<T>> for PushError<T> {
	fn from(v: ClosedError<T>) -> Self {
		Self::Closed(v.0)
	}
}