orx_concurrent_iter/pullers/
item_puller.rs

1use crate::concurrent_iter::ConcurrentIter;
2
3/// A regular [`Iterator`] which is created from and linked to and
4/// pulls its elements from a [`ConcurrentIter`].
5///
6/// It can be created using the [`item_puller`] method of a concurrent iterator.
7///
8/// [`item_puller`]: crate::ConcurrentIter::item_puller
9///
10/// # Examples
11///
12/// The definition might sound a bit confusing.
13///
14/// The following example demonstrates how it works:
15///
16/// * We have a concurrent iterator `con_iter` over elements "0", "1", ..., "99".
17/// * We spawn two threads, say A and B.
18/// * Each thread creates an `ItemPuller`, named as `puller`, from the same `con_iter`.
19/// * The following is one possible sequence of this parallel execution.
20///   * Thread A pulls "0" and calls process("0").
21///   * Thread B pulls "1" and calls process("1"); it completes processing before A.
22///   * Thread B pulls "2" and calls process("2").
23///   * Thread A pulls "3" and calls process("3").
24///   * and so on until all 100 elements are processed by the two threads.
25/// * Notice that there is only one data source `con_iter`;
26///   both of the `puller`s are connected to the same concurrent iterator, and
27///   each element is processed only once.
28///
29/// ```
30/// use orx_concurrent_iter::*;
31///
32/// let num_threads = 2;
33/// let data: Vec<_> = (0..100).map(|x| x.to_string()).collect();
34/// let con_iter = data.con_iter();
35///
36/// let process = |_x: &String| { /* assume actual work */ };
37///
38/// std::thread::scope(|s| {
39///     for _ in 0..num_threads {
40///         s.spawn(|| {
41///             // puller implements Iterator
42///             let puller = con_iter.item_puller();
43///             for value in puller {
44///                 process(value);
45///             }
46///         });
47///     }
48/// });
49/// ```
50///
51/// This approach brings the convenience of regular Iterators to the concurrent code.
52/// The example above already demonstrates that we can now use a regular `for` loop as we are
53/// writing a sequential code, while the program is parallelized.
54///
55/// Actually, we could've written an equivalent of the above program by directly using the
56/// concurrent iterator's [`next`] method and a `while let` loop:
57///
58/// ```ignore
59/// // ...
60/// std::thread::scope(|s| {
61///     for _ in 0..num_threads {
62///         s.spawn(|| {
63///             while let Some(value) = con_iter.next() {
64///                 process(value);
65///             }
66///         });
67///     }
68/// });
69/// ```
70///
71/// [`next`]: crate::ConcurrentIter::next
72///
73/// However, the convenience of the pullers goes beyond the `for` loops.
74/// All beautiful ergonomic Iterator methods become available in concurrent programs.
75///
76/// The following example demonstrate a very simple yet efficient implementation of the
77/// parallelized version of the [`reduce`] operation.
78///
79/// Notice that the entire implementation is nothing but a chain of Iterator methods.
80///
81/// [`reduce`]: Iterator::reduce
82///
83/// ```
84/// use orx_concurrent_iter::*;
85///
86/// fn parallel_reduce<T, F>(
87///     num_threads: usize,
88///     con_iter: impl ConcurrentIter<Item = T>,
89///     reduce: F,
90/// ) -> Option<T>
91/// where
92///     T: Send,
93///     F: Fn(T, T) -> T + Sync,
94/// {
95///     std::thread::scope(|s| {
96///         (0..num_threads)
97///             .map(|_| s.spawn(|| con_iter.item_puller().reduce(&reduce))) // reduce inside each thread
98///             .filter_map(|x| x.join().unwrap()) // join threads
99///             .reduce(&reduce) // reduce thread results to final result
100///     })
101/// }
102///
103/// let sum = parallel_reduce(8, (0..0).into_con_iter(), |a, b| a + b);
104/// assert_eq!(sum, None);
105///
106/// let n = 10_000;
107/// let data: Vec<_> = (0..n).collect();
108/// let sum = parallel_reduce(8, data.con_iter().copied(), |a, b| a + b);
109/// assert_eq!(sum, Some(n * (n - 1) / 2));
110/// ```
111pub struct ItemPuller<'a, I> {
112    con_iter: &'a I,
113}
114
115impl<I: ConcurrentIter> ItemPuller<'_, I> {
116    /// Behaves exactly as `next` but additionally provides `thread_idx` to the iterator.
117    /// This information might be useful for certain concurrent iterators, such as the
118    /// [recursive concurrent iterator](https://crates.io/crates/orx-concurrent-recursive-iter).
119    ///
120    /// Assuming a program using `n` threads that accesses this iterator, `thread_idx` is
121    /// assumed to be the internal ordering within this pool of threads taking values in
122    /// `0..n`.
123    #[inline(always)]
124    pub fn next_by(&mut self, thread_idx: usize) -> Option<I::Item> {
125        self.con_iter.next_by(thread_idx)
126    }
127}
128
129impl<'i, I: ConcurrentIter> From<&'i I> for ItemPuller<'i, I> {
130    fn from(con_iter: &'i I) -> Self {
131        Self { con_iter }
132    }
133}
134
135impl<I: ConcurrentIter> Iterator for ItemPuller<'_, I> {
136    type Item = I::Item;
137
138    #[inline(always)]
139    fn next(&mut self) -> Option<Self::Item> {
140        self.con_iter.next()
141    }
142
143    fn size_hint(&self) -> (usize, Option<usize>) {
144        // lb: other threads might pull all of the elements, hence 0
145        // ub: we might pull all elements, hence ub(con_iter)
146        (0, self.con_iter.size_hint().1)
147    }
148
149    fn fold<B, F>(self, init: B, mut f: F) -> B
150    where
151        Self: Sized,
152        F: FnMut(B, Self::Item) -> B,
153    {
154        let mut acc = init;
155
156        while let Some(elem) = self.con_iter.next() {
157            acc = f(acc, elem);
158        }
159
160        acc
161    }
162
163    fn count(self) -> usize
164    where
165        Self: Sized,
166    {
167        self.fold(0, |count, _| count + 1)
168    }
169}
orx_concurrent_iter/pullers/item_puller.rs

orx_concurrent_iter/pullers/
item_puller.rs
