vendor/crossbeam-deque-0.7.4/tests/fifo.rs

   1 extern crate crossbeam_deque as deque;
   2 extern crate crossbeam_utils as utils;
   3 extern crate rand;
   4
   5 use std::sync::atomic::Ordering::SeqCst;
   6 use std::sync::atomic::{AtomicBool, AtomicUsize};
   7 use std::sync::{Arc, Mutex};
   8
   9 use deque::Steal::{Empty, Success};
  10 use deque::Worker;
  11 use rand::Rng;
  12 use utils::thread::scope;
  13
  14 #[test]
  15 fn smoke() {
  16     let w = Worker::new_fifo();
  17     let s = w.stealer();
  18     assert_eq!(w.pop(), None);
  19     assert_eq!(s.steal(), Empty);
  20
  21     w.push(1);
  22     assert_eq!(w.pop(), Some(1));
  23     assert_eq!(w.pop(), None);
  24     assert_eq!(s.steal(), Empty);
  25
  26     w.push(2);
  27     assert_eq!(s.steal(), Success(2));
  28     assert_eq!(s.steal(), Empty);
  29     assert_eq!(w.pop(), None);
  30
  31     w.push(3);
  32     w.push(4);
  33     w.push(5);
  34     assert_eq!(s.steal(), Success(3));
  35     assert_eq!(s.steal(), Success(4));
  36     assert_eq!(s.steal(), Success(5));
  37     assert_eq!(s.steal(), Empty);
  38
  39     w.push(6);
  40     w.push(7);
  41     w.push(8);
  42     w.push(9);
  43     assert_eq!(w.pop(), Some(6));
  44     assert_eq!(s.steal(), Success(7));
  45     assert_eq!(w.pop(), Some(8));
  46     assert_eq!(w.pop(), Some(9));
  47     assert_eq!(w.pop(), None);
  48 }
  49
  50 #[test]
  51 fn is_empty() {
  52     let w = Worker::new_fifo();
  53     let s = w.stealer();
  54
  55     assert!(w.is_empty());
  56     w.push(1);
  57     assert!(!w.is_empty());
  58     w.push(2);
  59     assert!(!w.is_empty());
  60     let _ = w.pop();
  61     assert!(!w.is_empty());
  62     let _ = w.pop();
  63     assert!(w.is_empty());
  64
  65     assert!(s.is_empty());
  66     w.push(1);
  67     assert!(!s.is_empty());
  68     w.push(2);
  69     assert!(!s.is_empty());
  70     let _ = s.steal();
  71     assert!(!s.is_empty());
  72     let _ = s.steal();
  73     assert!(s.is_empty());
  74 }
  75
  76 #[test]
  77 fn spsc() {
  78     const STEPS: usize = 50_000;
  79
  80     let w = Worker::new_fifo();
  81     let s = w.stealer();
  82
  83     scope(|scope| {
  84         scope.spawn(|_| {
  85             for i in 0..STEPS {
  86                 loop {
  87                     if let Success(v) = s.steal() {
  88                         assert_eq!(i, v);
  89                         break;
  90                     }
  91                 }
  92             }
  93
  94             assert_eq!(s.steal(), Empty);
  95         });
  96
  97         for i in 0..STEPS {
  98             w.push(i);
  99         }
 100     })
 101     .unwrap();
 102 }
 103
 104 #[test]
 105 fn stampede() {
 106     const THREADS: usize = 8;
 107     const COUNT: usize = 50_000;
 108
 109     let w = Worker::new_fifo();
 110
 111     for i in 0..COUNT {
 112         w.push(Box::new(i + 1));
 113     }
 114     let remaining = Arc::new(AtomicUsize::new(COUNT));
 115
 116     scope(|scope| {
 117         for _ in 0..THREADS {
 118             let s = w.stealer();
 119             let remaining = remaining.clone();
 120
 121             scope.spawn(move |_| {
 122                 let mut last = 0;
 123                 while remaining.load(SeqCst) > 0 {
 124                     if let Success(x) = s.steal() {
 125                         assert!(last < *x);
 126                         last = *x;
 127                         remaining.fetch_sub(1, SeqCst);
 128                     }
 129                 }
 130             });
 131         }
 132
 133         let mut last = 0;
 134         while remaining.load(SeqCst) > 0 {
 135             if let Some(x) = w.pop() {
 136                 assert!(last < *x);
 137                 last = *x;
 138                 remaining.fetch_sub(1, SeqCst);
 139             }
 140         }
 141     })
 142     .unwrap();
 143 }
 144
 145 #[test]
 146 fn stress() {
 147     const THREADS: usize = 8;
 148     const COUNT: usize = 50_000;
 149
 150     let w = Worker::new_fifo();
 151     let done = Arc::new(AtomicBool::new(false));
 152     let hits = Arc::new(AtomicUsize::new(0));
 153
 154     scope(|scope| {
 155         for _ in 0..THREADS {
 156             let s = w.stealer();
 157             let done = done.clone();
 158             let hits = hits.clone();
 159
 160             scope.spawn(move |_| {
 161                 let w2 = Worker::new_fifo();
 162
 163                 while !done.load(SeqCst) {
 164                     if let Success(_) = s.steal() {
 165                         hits.fetch_add(1, SeqCst);
 166                     }
 167
 168                     let _ = s.steal_batch(&w2);
 169
 170                     if let Success(_) = s.steal_batch_and_pop(&w2) {
 171                         hits.fetch_add(1, SeqCst);
 172                     }
 173
 174                     while let Some(_) = w2.pop() {
 175                         hits.fetch_add(1, SeqCst);
 176                     }
 177                 }
 178             });
 179         }
 180
 181         let mut rng = rand::thread_rng();
 182         let mut expected = 0;
 183         while expected < COUNT {
 184             if rng.gen_range(0, 3) == 0 {
 185                 while let Some(_) = w.pop() {
 186                     hits.fetch_add(1, SeqCst);
 187                 }
 188             } else {
 189                 w.push(expected);
 190                 expected += 1;
 191             }
 192         }
 193
 194         while hits.load(SeqCst) < COUNT {
 195             while let Some(_) = w.pop() {
 196                 hits.fetch_add(1, SeqCst);
 197             }
 198         }
 199         done.store(true, SeqCst);
 200     })
 201     .unwrap();
 202 }
 203
 204 #[test]
 205 fn no_starvation() {
 206     const THREADS: usize = 8;
 207     const COUNT: usize = 50_000;
 208
 209     let w = Worker::new_fifo();
 210     let done = Arc::new(AtomicBool::new(false));
 211     let mut all_hits = Vec::new();
 212
 213     scope(|scope| {
 214         for _ in 0..THREADS {
 215             let s = w.stealer();
 216             let done = done.clone();
 217             let hits = Arc::new(AtomicUsize::new(0));
 218             all_hits.push(hits.clone());
 219
 220             scope.spawn(move |_| {
 221                 let w2 = Worker::new_fifo();
 222
 223                 while !done.load(SeqCst) {
 224                     if let Success(_) = s.steal() {
 225                         hits.fetch_add(1, SeqCst);
 226                     }
 227
 228                     let _ = s.steal_batch(&w2);
 229
 230                     if let Success(_) = s.steal_batch_and_pop(&w2) {
 231                         hits.fetch_add(1, SeqCst);
 232                     }
 233
 234                     while let Some(_) = w2.pop() {
 235                         hits.fetch_add(1, SeqCst);
 236                     }
 237                 }
 238             });
 239         }
 240
 241         let mut rng = rand::thread_rng();
 242         let mut my_hits = 0;
 243         loop {
 244             for i in 0..rng.gen_range(0, COUNT) {
 245                 if rng.gen_range(0, 3) == 0 && my_hits == 0 {
 246                     while let Some(_) = w.pop() {
 247                         my_hits += 1;
 248                     }
 249                 } else {
 250                     w.push(i);
 251                 }
 252             }
 253
 254             if my_hits > 0 && all_hits.iter().all(|h| h.load(SeqCst) > 0) {
 255                 break;
 256             }
 257         }
 258         done.store(true, SeqCst);
 259     })
 260     .unwrap();
 261 }
 262
 263 #[test]
 264 fn destructors() {
 265     const THREADS: usize = 8;
 266     const COUNT: usize = 50_000;
 267     const STEPS: usize = 1000;
 268
 269     struct Elem(usize, Arc<Mutex<Vec<usize>>>);
 270
 271     impl Drop for Elem {
 272         fn drop(&mut self) {
 273             self.1.lock().unwrap().push(self.0);
 274         }
 275     }
 276
 277     let w = Worker::new_fifo();
 278     let dropped = Arc::new(Mutex::new(Vec::new()));
 279     let remaining = Arc::new(AtomicUsize::new(COUNT));
 280
 281     for i in 0..COUNT {
 282         w.push(Elem(i, dropped.clone()));
 283     }
 284
 285     scope(|scope| {
 286         for _ in 0..THREADS {
 287             let remaining = remaining.clone();
 288             let s = w.stealer();
 289
 290             scope.spawn(move |_| {
 291                 let w2 = Worker::new_fifo();
 292                 let mut cnt = 0;
 293
 294                 while cnt < STEPS {
 295                     if let Success(_) = s.steal() {
 296                         cnt += 1;
 297                         remaining.fetch_sub(1, SeqCst);
 298                     }
 299
 300                     let _ = s.steal_batch(&w2);
 301
 302                     if let Success(_) = s.steal_batch_and_pop(&w2) {
 303                         cnt += 1;
 304                         remaining.fetch_sub(1, SeqCst);
 305                     }
 306
 307                     while let Some(_) = w2.pop() {
 308                         cnt += 1;
 309                         remaining.fetch_sub(1, SeqCst);
 310                     }
 311                 }
 312             });
 313         }
 314
 315         for _ in 0..STEPS {
 316             if let Some(_) = w.pop() {
 317                 remaining.fetch_sub(1, SeqCst);
 318             }
 319         }
 320     })
 321     .unwrap();
 322
 323     let rem = remaining.load(SeqCst);
 324     assert!(rem > 0);
 325
 326     {
 327         let mut v = dropped.lock().unwrap();
 328         assert_eq!(v.len(), COUNT - rem);
 329         v.clear();
 330     }
 331
 332     drop(w);
 333
 334     {
 335         let mut v = dropped.lock().unwrap();
 336         assert_eq!(v.len(), rem);
 337         v.sort();
 338         for pair in v.windows(2) {
 339             assert_eq!(pair[0] + 1, pair[1]);
 340         }
 341     }
 342 }