-
Notifications
You must be signed in to change notification settings - Fork 27
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Shared rb, heap and static rb trait, fix doc tests
- Loading branch information
Showing
12 changed files
with
261 additions
and
135 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,30 +1 @@ | ||
| use crate::{local::LocalRb, storage::Storage, utils::uninit_array}; | ||
| use core::mem::MaybeUninit; | ||
|
|
||
| #[cfg(feature = "alloc")] | ||
| use alloc::{collections::TryReserveError, vec::Vec}; | ||
|
|
||
| impl<T, const N: usize> Default for LocalRb<[MaybeUninit<T>; N]> { | ||
| fn default() -> Self { | ||
| uninit_array().into_rb() | ||
| } | ||
| } | ||
|
|
||
| #[cfg(feature = "alloc")] | ||
| impl<T> LocalRb<Vec<MaybeUninit<T>>> { | ||
| /// Creates a new instance of a ring buffer. | ||
| /// | ||
| /// *Panics if allocation failed or `capacity` is zero.* | ||
| pub fn new(capacity: usize) -> Self { | ||
| Self::try_new(capacity).unwrap() | ||
| } | ||
|
|
||
| /// Creates a new instance of a ring buffer returning an error if allocation failed. | ||
| /// | ||
| /// *Panics if `capacity` is zero.* | ||
| pub fn try_new(capacity: usize) -> Result<Self, TryReserveError> { | ||
| let mut data = Vec::new(); | ||
| data.try_reserve_exact(capacity)?; | ||
| Ok(data.into_rb()) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,113 @@ | ||
| use crate::{ | ||
| consumer::Consumer, | ||
| observer::Observer, | ||
| producer::Producer, | ||
| raw::RawRb, | ||
| ring_buffer::RingBuffer, | ||
| storage::{Shared, Storage}, | ||
| stored::StoredRb, | ||
| }; | ||
| use core::{ | ||
| mem::ManuallyDrop, | ||
| ptr, | ||
| sync::atomic::{AtomicUsize, Ordering}, | ||
| }; | ||
| use crossbeam_utils::CachePadded; | ||
|
|
||
| /// Ring buffer that could be shared between threads. | ||
| /// | ||
| /// Implements [`Sync`] *if `T` implements [`Send`]*. And therefore its [`Producer`] and [`Consumer`] implement [`Send`]. | ||
| /// | ||
| /// Note that there is no explicit requirement of `T: Send`. Instead [`SharedRb`] will work just fine even with `T: !Send` | ||
| /// until you try to send its [`Producer`] or [`Consumer`] to another thread. | ||
| #[cfg_attr( | ||
| feature = "std", | ||
| doc = r##" | ||
| ``` | ||
| use std::{thread, sync::Arc}; | ||
| use ringbuf::{SharedRb, storage::Heap, prelude::*}; | ||
| let rb = SharedRb::<Heap<i32>>::new(256); | ||
| let (mut prod, mut cons) = Split::<Arc<_>>::split(rb); | ||
| thread::spawn(move || { | ||
| prod.try_push(123).unwrap(); | ||
| }) | ||
| .join(); | ||
| thread::spawn(move || { | ||
| assert_eq!(cons.try_pop().unwrap(), 123); | ||
| }) | ||
| .join(); | ||
| ``` | ||
| "## | ||
| )] | ||
| pub struct SharedRb<S: Storage> { | ||
| storage: Shared<S>, | ||
| read: CachePadded<AtomicUsize>, | ||
| write: CachePadded<AtomicUsize>, | ||
| } | ||
|
|
||
| impl<S: Storage> StoredRb for SharedRb<S> { | ||
| type Storage = S; | ||
|
|
||
| unsafe fn from_raw_parts(storage: S, read: usize, write: usize) -> Self { | ||
| Self { | ||
| storage: Shared::new(storage), | ||
| read: CachePadded::new(AtomicUsize::new(read)), | ||
| write: CachePadded::new(AtomicUsize::new(write)), | ||
| } | ||
| } | ||
|
|
||
| unsafe fn into_raw_parts(self) -> (S, usize, usize) { | ||
| let (read, write) = (self.read_end(), self.write_end()); | ||
| let self_ = ManuallyDrop::new(self); | ||
| (ptr::read(&self_.storage).into_inner(), read, write) | ||
| } | ||
|
|
||
| fn storage(&self) -> &Shared<S> { | ||
| &self.storage | ||
| } | ||
| } | ||
|
|
||
| impl<S: Storage> RawRb for SharedRb<S> { | ||
| #[inline] | ||
| fn read_end(&self) -> usize { | ||
| self.read.load(Ordering::Acquire) | ||
| } | ||
|
|
||
| #[inline] | ||
| fn write_end(&self) -> usize { | ||
| self.write.load(Ordering::Acquire) | ||
| } | ||
|
|
||
| #[inline] | ||
| unsafe fn set_read_end(&self, value: usize) { | ||
| self.read.store(value, Ordering::Release) | ||
| } | ||
|
|
||
| #[inline] | ||
| unsafe fn set_write_end(&self, value: usize) { | ||
| self.write.store(value, Ordering::Release) | ||
| } | ||
| } | ||
|
|
||
| impl<S: Storage> Observer for SharedRb<S> { | ||
| type Item = S::Item; | ||
|
|
||
| type Raw = Self; | ||
|
|
||
| fn as_raw(&self) -> &Self::Raw { | ||
| self | ||
| } | ||
| } | ||
|
|
||
| impl<S: Storage> Producer for SharedRb<S> {} | ||
|
|
||
| impl<S: Storage> Consumer for SharedRb<S> {} | ||
|
|
||
| impl<S: Storage> RingBuffer for SharedRb<S> {} | ||
|
|
||
| impl<S: Storage> Drop for SharedRb<S> { | ||
| fn drop(&mut self) { | ||
| self.clear(); | ||
| } | ||
| } |
Oops, something went wrong.