Merge pull request #86 from orxfun/support-for-concurrent-queue

Support for concurrent queue

Author: orxfun

.github/workflows/ci.yml

@@ -18,13 +18,15 @@ jobs:
         toolchain: ["stable"]
 
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v5
 
     - name: Install toolchain
-      uses: dtolnay/rust-toolchain@master
+      uses: dtolnay/rust-toolchain@stable
       with:
         toolchain: ${{ matrix.toolchain }}
 
+    - name: Install clippy
+      run: rustup component add clippy
     - name: Install 32bit target
       run: rustup target add i686-unknown-linux-musl
     - name: Install wasm target

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "orx-split-vec"
-version = "3.19.0"
+version = "3.20.0"
 edition = "2024"
 authors = ["orxfun <orx.ugur.arikan@gmail.com>"]
 description = "An efficient dynamic capacity vector with pinned element guarantees."
@@ -12,7 +12,7 @@ categories = ["data-structures", "rust-patterns", "no-std"]
 [dependencies]
 orx-iterable = { version = "1.3.0", default-features = false }
 orx-pseudo-default = { version = "2.1.0", default-features = false }
-orx-pinned-vec = { version = "3.17.0", default-features = false }
+orx-pinned-vec = { version = "3.18.0", default-features = false }
 orx-concurrent-iter = { version = "3.1.0", default-features = false }
 
 [[bench]]
@@ -24,4 +24,4 @@ criterion = "0.7.0"
 rand = { version = "0.9.2", default-features = false }
 rand_chacha = { version = "0.9", default-features = false }
 test-case = "3.3.1"
-orx-concurrent-bag = "3.0.0"
+# orx-concurrent-bag = "3.1.0"

README.md

@@ -108,6 +108,7 @@ Naturally, it has certain specific differences and operations. For instance, we
 
 ```rust
 use orx_split_vec::*;
+use orx_pseudo_default::PseudoDefault;
 
 #[derive(Clone)]
 struct MyCustomGrowth;

src/concurrent_iter/tests/con_iter_owned/mod.rs

@@ -1,4 +1,4 @@
-mod con_iter;
-mod into;
+// mod con_iter;
+// mod into;
 mod par;
 mod transformations;

src/concurrent_iter/tests/con_iter_ref/mod.rs

@@ -1,4 +1,4 @@
-mod con_iter;
-mod into;
+// mod con_iter;
+// mod into;
 mod par;
 mod transformations;

src/concurrent_pinned_vec.rs src/concurrent_pinned_vec/con_pinvec.rssrc/concurrent_pinned_vec.rs renamed to src/concurrent_pinned_vec/con_pinvec.rs

@@ -1,12 +1,13 @@
 use crate::{
     Doubling, Fragment, GrowthWithConstantTimeAccess, SplitVec,
     common_traits::iterator::{IterOfSlicesOfCon, SliceBorrowAsMut, SliceBorrowAsRef},
+    concurrent_pinned_vec::iter_ptr::IterPtrOfCon,
     fragment::transformations::{fragment_from_raw, fragment_into_raw},
 };
 use alloc::vec::Vec;
-use core::cell::UnsafeCell;
 use core::ops::RangeBounds;
 use core::sync::atomic::{AtomicUsize, Ordering};
+use core::{cell::UnsafeCell, ops::Range};
 use orx_pinned_vec::ConcurrentPinnedVec;
 
 struct FragmentData {
@@ -189,6 +190,11 @@ impl<T, G: GrowthWithConstantTimeAccess> ConcurrentPinnedVec<T> for ConcurrentSp
     where
         Self: 'a;
 
+    type PtrIter<'a>
+        = IterPtrOfCon<'a, T, G>
+    where
+        Self: 'a;
+
     unsafe fn into_inner(mut self, len: usize) -> Self::P {
         let mut fragments = Vec::with_capacity(self.max_num_fragments);
         let mut take_fragment = |fragment| fragments.push(fragment);
@@ -424,4 +430,8 @@ impl<T, G: GrowthWithConstantTimeAccess> ConcurrentPinnedVec<T> for ConcurrentSp
         self.maximum_capacity = (0..self.data.len()).map(|f| self.capacity_of(f)).sum();
         self.pinned_vec_len = 0;
     }
+
+    unsafe fn ptr_iter_unchecked(&self, range: Range<usize>) -> Self::PtrIter<'_> {
+        IterPtrOfCon::new(self.capacity(), &self.data, self.growth.clone(), range)
+    }
 }

src/concurrent_pinned_vec/iter_ptr.rs

@@ -0,0 +1,96 @@
+use crate::{
+    GrowthWithConstantTimeAccess, concurrent_pinned_vec::iter_ptr_slices::IterPtrOfConSlices,
+};
+use core::{cell::UnsafeCell, ops::Range};
+
+pub struct IterPtrOfCon<'a, T, G>
+where
+    G: GrowthWithConstantTimeAccess,
+{
+    slices: IterPtrOfConSlices<'a, T, G>,
+    len_of_remaining_slices: usize,
+    current_ptr: *const T,
+    current_last: *const T,
+}
+
+impl<'a, T, G> IterPtrOfCon<'a, T, G>
+where
+    G: GrowthWithConstantTimeAccess,
+{
+    pub fn new(
+        capacity: usize,
+        fragments: &'a [UnsafeCell<*mut T>],
+        growth: G,
+        range: Range<usize>,
+    ) -> Self {
+        let len_of_remaining_slices = range.len();
+        let slices = IterPtrOfConSlices::new(capacity, fragments, growth, range);
+        Self {
+            slices,
+            len_of_remaining_slices,
+            current_ptr: core::ptr::null(),
+            current_last: core::ptr::null(),
+        }
+    }
+
+    fn remaining(&self) -> usize {
+        let remaining_current = match self.current_ptr.is_null() {
+            true => 0,
+            // SAFETY: whenever current_ptr is not null, we know that current_last is also not
+            // null which is >= current_ptr.
+            false => unsafe { self.current_last.offset_from(self.current_ptr) as usize + 1 },
+        };
+
+        self.len_of_remaining_slices + remaining_current
+    }
+
+    fn next_slice(&mut self) -> Option<*mut T> {
+        self.slices.next().and_then(|(ptr, len)| {
+            debug_assert!(len > 0);
+            self.len_of_remaining_slices -= len;
+            // SAFETY: pointers are not null since slice is not empty
+            self.current_ptr = ptr;
+            self.current_last = unsafe { ptr.add(len - 1) };
+            self.next()
+        })
+    }
+}
+
+impl<'a, T, G> Iterator for IterPtrOfCon<'a, T, G>
+where
+    G: GrowthWithConstantTimeAccess,
+{
+    type Item = *mut T;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.current_ptr {
+            ptr if ptr.is_null() => self.next_slice(),
+            ptr if ptr == self.current_last => {
+                self.current_ptr = core::ptr::null_mut();
+                Some(ptr as *mut T)
+            }
+            ptr => {
+                // SAFETY: current_ptr is not the last element, hance current_ptr+1 is in bounds
+                self.current_ptr = unsafe { self.current_ptr.add(1) };
+
+                // SAFETY: ptr is valid and its value can be taken.
+                // Drop will skip this position which is now uninitialized.
+                Some(ptr as *mut T)
+            }
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let len = self.remaining();
+        (len, Some(len))
+    }
+}
+
+impl<'a, T, G> ExactSizeIterator for IterPtrOfCon<'a, T, G>
+where
+    G: GrowthWithConstantTimeAccess,
+{
+    fn len(&self) -> usize {
+        self.remaining()
+    }
+}

src/concurrent_pinned_vec/iter_ptr_slices.rs

@@ -0,0 +1,157 @@
+use crate::{
+    GrowthWithConstantTimeAccess,
+    range_helpers::{range_end, range_start},
+};
+use core::cmp::min;
+use core::{cell::UnsafeCell, iter::FusedIterator, ops::Range};
+
+pub struct IterPtrOfConSlices<'a, T, G>
+where
+    G: GrowthWithConstantTimeAccess,
+{
+    fragments: &'a [UnsafeCell<*mut T>],
+    growth: G,
+    sf: usize,
+    si: usize,
+    si_end: usize,
+    ef: usize,
+    ei: usize,
+    f: usize,
+}
+
+impl<'a, T, G> IterPtrOfConSlices<'a, T, G>
+where
+    G: GrowthWithConstantTimeAccess,
+{
+    fn empty() -> Self {
+        Self {
+            fragments: &[],
+            growth: G::pseudo_default(),
+            sf: 0,
+            si: 0,
+            si_end: 0,
+            ef: 0,
+            ei: 0,
+            f: 1,
+        }
+    }
+
+    fn single_slice(
+        fragments: &'a [UnsafeCell<*mut T>],
+        growth: G,
+        f: usize,
+        begin: usize,
+        end: usize,
+    ) -> Self {
+        Self {
+            fragments,
+            growth,
+            sf: f,
+            si: begin,
+            si_end: end,
+            ef: f,
+            ei: 0,
+            f,
+        }
+    }
+
+    pub fn new(
+        capacity: usize,
+        fragments: &'a [UnsafeCell<*mut T>],
+        growth: G,
+        range: Range<usize>,
+    ) -> Self {
+        let fragment_and_inner_indices = |i| growth.get_fragment_and_inner_indices_unchecked(i);
+
+        let a = range_start(&range);
+        let b = min(capacity, range_end(&range, capacity));
+
+        match b.saturating_sub(a) {
+            0 => Self::empty(),
+            _ => {
+                let (sf, si) = fragment_and_inner_indices(a);
+                let (ef, ei) = fragment_and_inner_indices(b - 1);
+
+                match sf == ef {
+                    true => Self::single_slice(fragments, growth, sf, si, ei + 1),
+                    false => {
+                        let si_end = growth.fragment_capacity_of(sf);
+                        Self {
+                            fragments,
+                            growth,
+                            sf,
+                            si,
+                            si_end,
+                            ef,
+                            ei,
+                            f: sf,
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    #[inline(always)]
+    fn remaining_len(&self) -> usize {
+        (1 + self.ef).saturating_sub(self.f)
+    }
+
+    #[inline(always)]
+    fn get_ptr_fi(&self, f: usize, i: usize) -> *mut T {
+        let p = unsafe { *self.fragments[f].get() };
+        unsafe { p.add(i) }
+    }
+
+    #[inline(always)]
+    fn capacity_of(&self, f: usize) -> usize {
+        self.growth.fragment_capacity_of(f)
+    }
+}
+
+impl<'a, T, G> Iterator for IterPtrOfConSlices<'a, T, G>
+where
+    G: GrowthWithConstantTimeAccess,
+{
+    type Item = (*mut T, usize);
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.f {
+            f if f == self.sf => {
+                self.f += 1;
+                let len = self.si_end - self.si;
+                let p = self.get_ptr_fi(self.sf, self.si);
+                Some((p, len))
+            }
+            f if f < self.ef => {
+                self.f += 1;
+                let len = self.capacity_of(f);
+                let p = self.get_ptr_fi(f, 0);
+                Some((p, len))
+            }
+            f if f == self.ef => {
+                self.f += 1;
+                let len = self.ei + 1;
+                let p = self.get_ptr_fi(self.ef, 0);
+                Some((p, len))
+            }
+            _ => None,
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let len = self.remaining_len();
+        (len, Some(len))
+    }
+}
+
+impl<'a, T, G> FusedIterator for IterPtrOfConSlices<'a, T, G> where G: GrowthWithConstantTimeAccess {}
+
+impl<'a, T, G> ExactSizeIterator for IterPtrOfConSlices<'a, T, G>
+where
+    G: GrowthWithConstantTimeAccess,
+{
+    fn len(&self) -> usize {
+        self.remaining_len()
+    }
+}

src/concurrent_pinned_vec/mod.rs

@@ -0,0 +1,5 @@
+mod con_pinvec;
+mod iter_ptr;
+mod iter_ptr_slices;
+
+pub use con_pinvec::ConcurrentSplitVec;

src/new_split_vec/new.rs

@@ -32,6 +32,7 @@ where
     ///
     /// ```
     /// use orx_split_vec::*;
+    /// use orx_pseudo_default::PseudoDefault;
     ///
     /// #[derive(Clone)]
     /// pub struct DoubleEverySecondFragment(usize); // any custom growth strategy