orx_concurrent_iter/implementations/slice/
con_iter.rs

1use super::chunk_puller::ChunkPullerSlice;
2use crate::{concurrent_iter::ConcurrentIter, exact_size_concurrent_iter::ExactSizeConcurrentIter};
3use core::{
4    iter::Skip,
5    sync::atomic::{AtomicUsize, Ordering},
6};
7
8/// Concurrent iterator of a slice.
9///
10/// It can be created by calling [`into_con_iter`] on a slice.
11///
12/// Alternatively, it can be created calling [`con_iter`] on the type
13/// that owns the slice.
14///
15/// [`into_con_iter`]: crate::IntoConcurrentIter::into_con_iter
16/// [`con_iter`]: crate::ConcurrentIterable::con_iter
17///
18/// # Examples
19///
20/// ```
21/// use orx_concurrent_iter::*;
22///
23/// // &[T]: IntoConcurrentIter
24/// let vec = vec![0, 1, 2, 3];
25/// let slice = &vec[1..3];
26/// let con_iter = slice.into_con_iter();
27/// assert_eq!(con_iter.next(), Some(&1));
28/// assert_eq!(con_iter.next(), Some(&2));
29/// assert_eq!(con_iter.next(), None);
30///
31/// // Vec<T>: ConcurrentIterable
32/// let vec = vec![1, 2];
33/// let con_iter = vec.con_iter();
34/// assert_eq!(con_iter.next(), Some(&1));
35/// assert_eq!(con_iter.next(), Some(&2));
36/// assert_eq!(con_iter.next(), None);
37/// ```
38pub struct ConIterSlice<'a, T> {
39    slice: &'a [T],
40    counter: AtomicUsize,
41}
42
43impl<T> Default for ConIterSlice<'_, T> {
44    fn default() -> Self {
45        Self::new(&[])
46    }
47}
48
49impl<'a, T> ConIterSlice<'a, T> {
50    pub(crate) fn new(slice: &'a [T]) -> Self {
51        Self {
52            slice,
53            counter: 0.into(),
54        }
55    }
56
57    pub(super) fn slice(&self) -> &'a [T] {
58        self.slice
59    }
60
61    fn progress_and_get_begin_idx(&self, number_to_fetch: usize) -> Option<usize> {
62        let begin_idx = self.counter.fetch_add(number_to_fetch, Ordering::Relaxed);
63        match begin_idx < self.slice.len() {
64            true => Some(begin_idx),
65            _ => None,
66        }
67    }
68
69    pub(super) fn progress_and_get_slice(&self, chunk_size: usize) -> Option<(usize, &'a [T])> {
70        self.progress_and_get_begin_idx(chunk_size)
71            .map(|begin_idx| {
72                let end_idx = (begin_idx + chunk_size)
73                    .min(self.slice.len())
74                    .max(begin_idx);
75                (begin_idx, &self.slice[begin_idx..end_idx])
76            })
77    }
78}
79
80impl<'a, T> ConcurrentIter for ConIterSlice<'a, T>
81where
82    T: Sync,
83{
84    type Item = &'a T;
85
86    type SequentialIter = Skip<core::slice::Iter<'a, T>>;
87
88    type ChunkPuller<'i>
89        = ChunkPullerSlice<'i, 'a, T>
90    where
91        Self: 'i;
92
93    fn is_source_serialized() -> bool {
94        false
95    }
96
97    fn into_seq_iter(self) -> Self::SequentialIter {
98        let current = self.counter.load(Ordering::Acquire);
99        self.slice.iter().skip(current)
100    }
101
102    fn skip_to_end(&self) {
103        let _ = self.counter.fetch_max(self.slice.len(), Ordering::Acquire);
104    }
105
106    fn next(&self) -> Option<Self::Item> {
107        self.progress_and_get_begin_idx(1)
108            .map(|idx| &self.slice[idx])
109    }
110
111    fn next_with_idx(&self) -> Option<(usize, Self::Item)> {
112        self.progress_and_get_begin_idx(1)
113            .map(|idx| (idx, &self.slice[idx]))
114    }
115
116    fn size_hint(&self) -> (usize, Option<usize>) {
117        let num_taken = self.counter.load(Ordering::Acquire);
118        let remaining = self.slice.len().saturating_sub(num_taken);
119        (remaining, Some(remaining))
120    }
121
122    fn is_completed_when_none_returned(&self) -> bool {
123        true
124    }
125
126    fn chunk_puller(&self, chunk_size: usize) -> Self::ChunkPuller<'_> {
127        let chunk_size = chunk_size.min(self.slice.len());
128        Self::ChunkPuller::new(self, chunk_size)
129    }
130}
131
132impl<T> ExactSizeConcurrentIter for ConIterSlice<'_, T>
133where
134    T: Sync,
135{
136    fn len(&self) -> usize {
137        let num_taken = self.counter.load(Ordering::Acquire);
138        self.slice.len().saturating_sub(num_taken)
139    }
140}
orx_concurrent_iter/implementations/slice/con_iter.rs

orx_concurrent_iter/implementations/slice/
con_iter.rs
