Merge branch 'tests'

3 files changed, 76 insertions, 9 deletions

diff --git a/src/platform/linux/udp.rs b/src/platform/linux/udp.rs
index 7b4fa33..e76c2a8 100644
--- a/src/platform/linux/udp.rs
+++ b/src/platform/linux/udp.rs
@@ -118,7 +118,7 @@ fn setsockopt_int(
 
 #[allow(non_snake_case)]
 const fn CMSG_ALIGN(len: usize) -> usize {
-    (((len) + mem::size_of::<u32>() - 1) & !(mem::size_of::<u32>() - 1))
+    ((len) + mem::size_of::<u32>() - 1) & !(mem::size_of::<u32>() - 1)
 }
 
 #[allow(non_snake_case)]
diff --git a/src/wireguard/router/anti_replay.rs b/src/wireguard/router/anti_replay.rs
index b0838bd..b47dea9 100644
--- a/src/wireguard/router/anti_replay.rs
+++ b/src/wireguard/router/anti_replay.rs
@@ -18,7 +18,7 @@ const REDUNDANT_BIT_SHIFTS: usize = 5;
 const SIZE_OF_WORD: usize = mem::size_of::<Word>() * 8;
 
 const BITMAP_BITLEN: usize = 2048;
-const BITMAP_LEN: usize = (BITMAP_BITLEN / SIZE_OF_WORD);
+const BITMAP_LEN: usize = BITMAP_BITLEN / SIZE_OF_WORD;
 const BITMAP_INDEX_MASK: u64 = BITMAP_LEN as u64 - 1;
 const BITMAP_LOC_MASK: u64 = (SIZE_OF_WORD - 1) as u64;
 const WINDOW_SIZE: u64 = (BITMAP_BITLEN - SIZE_OF_WORD) as u64;
diff --git a/src/wireguard/router/queue.rs b/src/wireguard/router/queue.rs
index ec4492e..6517ba4 100644
--- a/src/wireguard/router/queue.rs
+++ b/src/wireguard/router/queue.rs
@@ -97,25 +97,92 @@ impl<J: SequentialJob> Queue<J> {
 mod tests {
     use super::*;
 
-    use std::sync::Arc;
     use std::thread;
 
+    use std::sync::Arc;
+    use std::time::Duration;
+
     use rand::thread_rng;
     use rand::Rng;
 
-    struct TestJob {}
+    #[test]
+    fn test_consume_queue() {
+        struct TestJob {
+            cnt: Arc<AtomicUsize>,
+            wait_sequential: Duration,
+        }
+
+        impl SequentialJob for TestJob {
+            fn is_ready(&self) -> bool {
+                true
+            }
+
+            fn sequential_work(self) {
+                thread::sleep(self.wait_sequential);
+                self.cnt.fetch_add(1, Ordering::SeqCst);
+            }
+        }
+
+        fn hammer(queue: &Arc<Queue<TestJob>>, cnt: Arc<AtomicUsize>) -> usize {
+            let mut jobs = 0;
+            let mut rng = thread_rng();
+            for _ in 0..10_000 {
+                if rng.gen() {
+                    let wait_sequential: u64 = rng.gen();
+                    let wait_sequential = wait_sequential % 1000;
+
+                    let wait_parallel: u64 = rng.gen();
+                    let wait_parallel = wait_parallel % 1000;
 
-    impl SequentialJob for TestJob {
-        fn is_ready(&self) -> bool {
-            true
+                    thread::sleep(Duration::from_micros(wait_parallel));
+
+                    queue.push(TestJob {
+                        cnt: cnt.clone(),
+                        wait_sequential: Duration::from_micros(wait_sequential),
+                    });
+                    jobs += 1;
+                } else {
+                    queue.consume();
+                }
+            }
+            queue.consume();
+            jobs
         }
 
-        fn sequential_work(self) {}
+        let queue = Arc::new(Queue::new());
+        let counter = Arc::new(AtomicUsize::new(0));
+
+        // repeatedly apply operations randomly from concurrent threads
+        let other = {
+            let queue = queue.clone();
+            let counter = counter.clone();
+            thread::spawn(move || hammer(&queue, counter))
+        };
+        let mut jobs = hammer(&queue, counter.clone());
+
+        // wait, consume and check empty
+        jobs += other.join().unwrap();
+        assert_eq!(queue.queue.lock().len(), 0, "elements left in queue");
+        assert_eq!(
+            jobs,
+            counter.load(Ordering::Acquire),
+            "did not consume every job"
+        );
     }
 
     /* Fuzz the Queue */
     #[test]
-    fn test_queue() {
+    fn test_fuzz_queue() {
+        struct TestJob {}
+
+        impl SequentialJob for TestJob {
+            fn is_ready(&self) -> bool {
+                true
+            }
+
+            fn sequential_work(self) {}
+        }
+
         fn hammer(queue: &Arc<Queue<TestJob>>) {
             let mut rng = thread_rng();
             for _ in 0..1_000_000 {