Merge branch 'tests'

5 files changed, 82 insertions, 15 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 81ffad4..9aff7f2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1032,8 +1032,8 @@ dependencies = [
 
 [[package]]
 name = "treebitmap"
-version = "0.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
+version = "0.5.0"
+source = "git+https://github.com/JakubOnderka/treebitmap#207c371a501780a94a8cd375fe15f877b110d9e2"
 
 [[package]]
 name = "typenum"
@@ -1253,7 +1253,7 @@ dependencies = [
  "ring 0.16.11 (registry+https://github.com/rust-lang/crates.io-index)",
  "spin 0.5.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "subtle 2.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
- "treebitmap 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "treebitmap 0.5.0 (git+https://github.com/JakubOnderka/treebitmap)",
  "x25519-dalek 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "zerocopy 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -1440,7 +1440,7 @@ dependencies = [
 "checksum term 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)" = "fa63644f74ce96fbeb9b794f66aff2a52d601cbd5e80f4b97123e3899f4570f1"
 "checksum termcolor 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "bb6bfa289a4d7c5766392812c0a1f4c1ba45afa1ad47803c11e1f407d846d75f"
 "checksum thread_local 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)" = "c6b53e329000edc2b34dbe8545fd20e55a333362d0a321909685a19bd28c3f1b"
-"checksum treebitmap 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "6bf423939ac9ccf4083788879b883a7149176586f9cf8b0fb1fd88b66ad692b5"
+"checksum treebitmap 0.5.0 (git+https://github.com/JakubOnderka/treebitmap)" = "<none>"
 "checksum typenum 1.11.2 (registry+https://github.com/rust-lang/crates.io-index)" = "6d2783fe2d6b8c1101136184eb41be8b1ad379e4657050b8aaff0c79ee7575f9"
 "checksum unicode-segmentation 1.6.0 (registry+https://github.com/rust-lang/crates.io-index)" = "e83e153d1053cbb5a118eeff7fd5be06ed99153f00dbcd8ae310c5fb2b22edc0"
 "checksum unicode-xid 0.0.3 (registry+https://github.com/rust-lang/crates.io-index)" = "36dff09cafb4ec7c8cf0023eb0b686cb6ce65499116a12201c9e11840ca01beb"
diff --git a/Cargo.toml b/Cargo.toml
index b884216..e89676a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -15,7 +15,7 @@ zerocopy = "0.2.7"
 byteorder = "1.3.1"
 digest = "0.8.0"
 arraydeque = "0.4.5"
-treebitmap = "^0.4"
+treebitmap = { git = "https://github.com/JakubOnderka/treebitmap" }
 hjul = "0.2.1"
 ring = "0.16.7"
 rand = "^0.7"
@@ -46,4 +46,4 @@ start_up = []
 [dev-dependencies]
 pnet = "0.25.0"
 proptest = "0.9.4"
-rand_chacha = "0.2.1"
\ No newline at end of file
+rand_chacha = "0.2.1"
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 {