1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
//! LimitAccessQueue is where the queue is stored and managed. This may only be accessed via the accessors.
use crate::utils::SpinWait;
use super::read_accessor::*;
use std::sync::{atomic::AtomicBool, Arc};
pub struct LimitAccessQueue<T,State> {
pub val: Vec<T>,
write_block: AtomicBool,
state: State
}
#[allow(dead_code,clippy::new_ret_no_self)]
impl<T,State> LimitAccessQueue<T,State>
where State: Default + Clone
{
pub fn new() -> (PrimaryAccessor<T,State>,SecondaryAccessor<T,State>) {
let arc_obj = Arc::new(Self {
val: Vec::new(),
write_block: AtomicBool::new(false),
state: State::default()
});
//we need to ensure the object within AtomicPtr survives on the heap and beyond
//the function stack.
let primary = ReadAccessor::new(arc_obj.clone(),ReadAccessorType::Primary);
let secondary = ReadAccessor::new(arc_obj,ReadAccessorType::Secondary);
(PrimaryAccessor::new(primary), SecondaryAccessor::new(secondary))
}
pub fn set_state(&mut self, state:State) {
self.with_write_block(|s|{
s.state = state;
});
}
pub fn get_state(&mut self) -> State {
self.with_write_block(|s|{
s.state.clone()
})
}
pub fn pop(&mut self) -> Option<T> {
self.with_write_block(|s| {
s.val.pop()
})
}
pub fn pop_count(&mut self,count:usize) -> Option<Vec<T>> {
self.with_write_block(|s| {
let mut res = Vec::new();
for idx in 0..count {
if let Some(val) = s.val.pop() {
res.push(val)
} else {
if idx == 0 {
return None;
}
break;
}
}
Some(res)
})
}
///Steals all the un-popped values from the queue. It can then be reused
/// elsewhere.
/// ```
/// use parallel_task::{
/// accessors::limit_queue::LimitAccessQueue,
/// push_workers::worker_thread::Coordination};
/// let values = (0..100_000).collect::<Vec<_>>();
/// let (mut primary, _) = LimitAccessQueue::<i32,Coordination>::new();
/// _ = primary.write(values);
/// let vec = primary.steal().unwrap(); //This step should not fail here. But unwrap not advised in production
/// assert_eq!(vec.len(), 100_000);
/// ```
pub fn steal(&mut self) -> Option<Vec<T>> {
self.with_write_block(|s| {
if s.val.is_empty() {
None
} else {
// using mem swap to expedite the process
let mut tmp:Vec<T> = Vec::with_capacity(1);
std::mem::swap(&mut tmp, &mut s.val);
Some(tmp)
}
})
}
///Steals half the un-popped values from the queue. It can then be reused
/// elsewhere.
/// ```
/// use parallel_task::{
/// accessors::limit_queue::LimitAccessQueue,
/// push_workers::worker_thread::Coordination};
/// let values = (0..100_000).collect::<Vec<_>>();
/// let (mut primary, _) = LimitAccessQueue::<i32,Coordination>::new();
/// _ = primary.write(values);
/// let vec = primary.steal_half().unwrap(); //This step should not fail here. But unwrap not advised in production
/// assert_eq!(vec.len(), 50_000);
/// ```
pub fn steal_half(&mut self) -> Option<Vec<T>> {
self.with_write_block(|s| {
if s.val.is_empty() {
None
}
else {
let res = s.val.split_off(s.val.len()/2);
Some(res)
}
})
}
pub fn is_empty(&mut self) -> bool {
self.len() == 0
}
pub fn len(&mut self) -> usize {
self.with_write_block(|s|{
if s.val.is_empty() { 0usize } else { s.val.len() }
})
}
pub fn atomic_write_block_to_true(&mut self) -> Result<bool, bool> {
self.write_block.compare_exchange(false, true, std::sync::atomic::Ordering::SeqCst, std::sync::atomic::Ordering::SeqCst)
}
pub fn with_write_block<F,Output>(&mut self, f:F) -> Output
where F: FnOnce(&mut Self) -> Output {
SpinWait::loop_while_mut(||self.atomic_write_block_to_true().is_err());
let output = f(self);
self.write_block.store(false, std::sync::atomic::Ordering::SeqCst);
output
}
pub fn push(&mut self, value:T) {
self.with_write_block(|s|
{
s.val.push(value);
});
}
pub fn write(&mut self, mut values:Vec<T>) {
self.with_write_block(|s|{
let drained = values.drain(0..);
s.val.extend(drained);
});
}
pub fn replace(&mut self, mut values:Vec<T>) {
self.with_write_block(|s|{
std::mem::swap(&mut values, &mut s.val);
});
}
pub fn is_write_blocked(&self) -> bool {
self.write_block.load(std::sync::atomic::Ordering::SeqCst)
}
// pub fn ingest_iter<I>(&mut self, mut i:I)
// where I:AccessQueueIngestor<IngestorItem = T>
// {
// while let Some(value) = i.next_chunk() {
// self.push(value);
// }
// }
}