parallel_task/accessors/
limit_queue.rs

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

parallel_task/accessors/
limit_queue.rs
