orx_parallel/extendable/par_extend_impl/
vec_deque.rs

1use crate::extendable::par_extend_core::ParExtendCore;
2use crate::extendable::par_extend_impl::utils::{ColAndPos, IdxLen, NextN};
3use alloc::collections::VecDeque;
4use alloc::{vec, vec::Vec};
5use orx_priority_queue::{BinaryHeap, PriorityQueue};
6
7impl<T: Send> ParExtendCore<T> for VecDeque<T> {
8    type ThreadValues = Self;
9
10    type OrderedThreadValues = ColAndPos<Vec<T>>;
11
12    fn new_thread_values() -> Self::ThreadValues {
13        Default::default()
14    }
15
16    fn new_ordered_thread_values() -> Self::OrderedThreadValues {
17        Default::default()
18    }
19
20    // thread collect
21
22    fn add_thread_value(collected: &mut Self::ThreadValues, value: T) {
23        collected.push_back(value);
24    }
25
26    fn add_thread_values(collected: &mut Self::ThreadValues, values: impl IntoIterator<Item = T>) {
27        collected.extend(values)
28    }
29
30    fn add_ordered_thread_value(collected: &mut Self::OrderedThreadValues, idx: usize, value: T) {
31        collected.values.push(value);
32        collected.positions.push(IdxLen { idx, len: 1 });
33    }
34
35    fn add_ordered_thread_values(
36        collected: &mut Self::OrderedThreadValues,
37        idx: usize,
38        values: impl IntoIterator<Item = T>,
39    ) {
40        let len_begin = collected.values.len();
41        collected.values.extend(values);
42
43        let len = collected.values.len() - len_begin;
44        if len > 0 {
45            collected.positions.push(IdxLen { idx, len });
46        }
47    }
48
49    // opt: thread collect
50
51    fn add_ordered_thread_optionals(
52        collected: &mut Self::OrderedThreadValues,
53        idx: usize,
54        values: impl IntoIterator<Item = Option<T>>,
55    ) -> Option<()> {
56        let len_begin = collected.values.len();
57        for value in values {
58            collected.values.push(value?);
59        }
60
61        let len = collected.values.len() - len_begin;
62        if len > 0 {
63            collected.positions.push(IdxLen { idx, len });
64        }
65
66        Some(())
67    }
68
69    // res: thread collect
70
71    fn add_ordered_thread_fallibles<E>(
72        collected: &mut Self::OrderedThreadValues,
73        idx: usize,
74        values: impl IntoIterator<Item = Result<T, E>>,
75    ) -> Result<(), E> {
76        let len_begin = collected.values.len();
77        for value in values {
78            collected.values.push(value?);
79        }
80
81        let len = collected.values.len() - len_begin;
82        if len > 0 {
83            collected.positions.push(IdxLen { idx, len });
84        }
85
86        Ok(())
87    }
88
89    // add
90
91    fn add_one(&mut self, value: T) {
92        self.push_back(value);
93    }
94
95    // extend - merge
96
97    fn extend_merge_infallibles(&mut self, results: Vec<Self::ThreadValues>) {
98        let collected_len: usize = results.iter().map(|x| x.len()).sum();
99        self.reserve(collected_len);
100        for result in results {
101            self.extend(result);
102        }
103    }
104
105    fn extend_merge_ordered_infallibles(&mut self, results: Vec<Self::OrderedThreadValues>) {
106        let collected_len: usize = results.iter().map(|x| x.values.len()).sum();
107        self.reserve(collected_len);
108
109        let outer_len = results.len();
110        let mut all_values = Vec::with_capacity(outer_len);
111        let mut all_positions = Vec::with_capacity(outer_len);
112
113        for x in results {
114            all_values.push(x.values.into_iter());
115            all_positions.push(x.positions);
116        }
117
118        let mut queue = BinaryHeap::with_capacity(outer_len);
119        let mut pos_indices = vec![0; outer_len];
120        for (th, positions) in all_positions.iter().enumerate() {
121            if let Some(pos) = positions.first() {
122                let node = ThLen::new(th, pos.len);
123                queue.push(node, pos.idx);
124            }
125        }
126        let mut curr_t = queue.pop_node();
127
128        while let Some(ThLen { th, len }) = curr_t {
129            let chunk = NextN::new(&mut all_values[th], len);
130            self.extend(chunk);
131
132            pos_indices[th] += 1;
133            curr_t = match all_positions[th].get(pos_indices[th]) {
134                Some(pos) => {
135                    let node = ThLen::new(th, pos.len);
136                    Some(queue.push_then_pop(node, pos.idx).0)
137                }
138                None => queue.pop_node(),
139            };
140        }
141    }
142}
143
144// merge helpers
145
146#[derive(Clone)]
147struct ThLen {
148    th: usize,
149    len: usize,
150}
151
152impl ThLen {
153    #[inline(always)]
154    fn new(th: usize, len: usize) -> Self {
155        Self { th, len }
156    }
157}
orx_parallel/extendable/par_extend_impl/vec_deque.rs

orx_parallel/extendable/par_extend_impl/
vec_deque.rs
