orx_parallel/extendable/par_extend_impl/
split_vec_doubling.rs

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

orx_parallel/extendable/par_extend_impl/
split_vec_doubling.rs
