Add concurrent unit tests

al8n · Sep 14, 2024 · bb63228 · bb63228
1 parent e5e09d2
commit bb63228
Show file tree

Hide file tree

Showing 4 changed files with 210 additions and 12 deletions.
diff --git a/src/swmr/generic.rs b/src/swmr/generic.rs
@@ -182,7 +182,7 @@ where
   fn compare(&self, p: &Pointer<K, V>) -> cmp::Ordering {
     let kr: K::Ref<'_> = TypeRef::from_slice(p.as_key_slice());
     let or: K::Ref<'_> = TypeRef::from_slice(self.as_key_slice());
-    KeyRef::compare(&kr, &or)
+    KeyRef::compare(&kr, &or).reverse()
   }
 }
 
@@ -221,7 +221,7 @@ where
 {
   fn compare(&self, p: &Pointer<K, V>) -> cmp::Ordering {
     let kr = TypeRef::from_slice(p.as_key_slice());
-    KeyRef::compare(&kr, self.key)
+    KeyRef::compare(&kr, self.key).reverse()
   }
 }
 
@@ -260,7 +260,7 @@ where
 {
   fn compare(&self, p: &Pointer<K, V>) -> cmp::Ordering {
     let kr = <K::Ref<'_> as TypeRef<'_>>::from_slice(p.as_key_slice());
-    KeyRef::compare(&kr, self.key)
+    KeyRef::compare(&kr, self.key).reverse()
   }
 }
 
@@ -1374,7 +1374,7 @@ where
   ///   where
   ///     Q: ?Sized + Ord + Comparable<Self>,
   ///   {
-  ///     Comparable::compare(a, self)
+  ///     Comparable::compare(a, self).reverse()
   ///   }
   ///
   ///   fn compare_binary(this: &[u8], other: &[u8]) -> cmp::Ordering {

diff --git a/src/swmr/generic/tests.rs b/src/swmr/generic/tests.rs
@@ -1,4 +1,4 @@
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, thread::spawn};
 
 use arbitrary::Arbitrary;
 use dbutils::leb128::{decode_u64_varint, encode_u64_varint, encoded_u64_varint_len};
@@ -115,7 +115,7 @@ impl<'a> KeyRef<'a, Person> for PersonRef<'a> {
   where
     Q: ?Sized + Ord + Comparable<Self>,
   {
-    Comparable::compare(a, self)
+    Comparable::compare(a, self).reverse()
   }
 
   fn compare_binary(this: &[u8], other: &[u8]) -> cmp::Ordering {
@@ -1609,3 +1609,130 @@ fn reserved_map_file() {
 
   reserved(&mut wal);
 }
+
+fn concurrent_basic(mut w: GenericOrderWal<u32, [u8; 4]>) {
+  let readers = (0..100u32).map(|i| (i, w.reader())).collect::<Vec<_>>();
+
+  let handles = readers.into_iter().map(|(i, reader)| {
+    spawn(move || loop {
+      if let Some(p) = reader.get(&i) {
+        assert_eq!(p.key(), i);
+        assert_eq!(p.value(), i.to_le_bytes());
+        break;
+      }
+    })
+  });
+
+  spawn(move || {
+    for i in 0..100u32 {
+      w.insert(&i, &i.to_le_bytes()).unwrap();
+    }
+  });
+
+  for handle in handles {
+    handle.join().unwrap();
+  }
+}
+
+#[test]
+fn concurrent_basic_inmemory() {
+  let wal = GenericOrderWal::<u32, [u8; 4]>::new(Options::new().with_capacity(MB).with_reserved(4))
+    .unwrap();
+  concurrent_basic(wal);
+}
+
+#[test]
+fn concurrent_basic_map_anon() {
+  let wal =
+    GenericOrderWal::<u32, [u8; 4]>::map_anon(Options::new().with_capacity(MB).with_reserved(4))
+      .unwrap();
+  concurrent_basic(wal);
+}
+
+#[test]
+#[cfg_attr(miri, ignore)]
+fn concurrent_basic_map_file() {
+  let dir = tempdir().unwrap();
+  let path = dir.path().join("generic_wal_concurrent_basic_map_file");
+
+  let wal = unsafe {
+    GenericOrderWal::<u32, [u8; 4]>::map_mut(
+      &path,
+      Options::new().with_reserved(4),
+      OpenOptions::new()
+        .create_new(Some(MB))
+        .write(true)
+        .read(true),
+    )
+    .unwrap()
+  };
+
+  concurrent_basic(wal);
+
+  let wal =
+    unsafe { GenericOrderWal::<u32, [u8; 4]>::map(path, Options::new().with_reserved(4)).unwrap() };
+
+  for i in 0..100u32 {
+    assert!(wal.contains_key(&i));
+  }
+}
+
+fn concurrent_one_key(mut w: GenericOrderWal<u32, [u8; 4]>) {
+  let readers = (0..100u32).map(|i| (i, w.reader())).collect::<Vec<_>>();
+  let handles = readers.into_iter().map(|(_, reader)| {
+    spawn(move || loop {
+      if let Some(p) = reader.get(&1) {
+        assert_eq!(p.key(), 1);
+        assert_eq!(p.value(), 1u32.to_le_bytes());
+        break;
+      }
+    })
+  });
+
+  w.insert(&1, &1u32.to_le_bytes()).unwrap();
+
+  for handle in handles {
+    handle.join().unwrap();
+  }
+}
+
+#[test]
+fn concurrent_one_key_inmemory() {
+  let wal = GenericOrderWal::<u32, [u8; 4]>::new(Options::new().with_capacity(MB).with_reserved(4))
+    .unwrap();
+  concurrent_one_key(wal);
+}
+
+#[test]
+fn concurrent_one_key_map_anon() {
+  let wal =
+    GenericOrderWal::<u32, [u8; 4]>::map_anon(Options::new().with_capacity(MB).with_reserved(4))
+      .unwrap();
+  concurrent_one_key(wal);
+}
+
+#[test]
+#[cfg_attr(miri, ignore)]
+fn concurrent_one_key_map_file() {
+  let dir = tempdir().unwrap();
+  let path = dir.path().join("generic_wal_concurrent_basic_map_file");
+
+  let wal = unsafe {
+    GenericOrderWal::<u32, [u8; 4]>::map_mut(
+      &path,
+      Options::new().with_reserved(4),
+      OpenOptions::new()
+        .create_new(Some(MB))
+        .write(true)
+        .read(true),
+    )
+    .unwrap()
+  };
+
+  concurrent_one_key(wal);
+
+  let wal =
+    unsafe { GenericOrderWal::<u32, [u8; 4]>::map(path, Options::new().with_reserved(4)).unwrap() };
+
+  assert!(wal.contains_key(&1));
+}
diff --git a/src/swmr/generic/traits/impls.rs b/src/swmr/generic/traits/impls.rs
@@ -21,3 +21,80 @@ impl Type for () {
 impl TypeRef<'_> for () {
   fn from_slice(_buf: &[u8]) -> Self {}
 }
+
+impl<const N: usize> Type for [u8; N] {
+  type Ref<'a> = Self;
+
+  type Error = ();
+
+  fn encoded_len(&self) -> usize {
+    N
+  }
+
+  fn encode(&self, buf: &mut [u8]) -> Result<(), Self::Error> {
+    buf[..N].copy_from_slice(self.as_ref());
+    Ok(())
+  }
+}
+
+impl<const N: usize> TypeRef<'_> for [u8; N] {
+  #[inline]
+  fn from_slice(src: &'_ [u8]) -> Self {
+    let mut this = [0; N];
+    this.copy_from_slice(src);
+    this
+  }
+}
+
+macro_rules! impl_numbers {
+  ($($ty:ident), +$(,)?) => {
+    $(
+      impl Type for $ty {
+        type Ref<'a> = Self;
+
+        type Error = ();
+
+        #[inline]
+        fn encoded_len(&self) -> usize {
+          core::mem::size_of::<$ty>()
+        }
+
+        #[inline]
+        fn encode(&self, buf: &mut [u8]) -> Result<(), Self::Error> {
+          const SIZE: usize = core::mem::size_of::<$ty>();
+          Ok(buf[..SIZE].copy_from_slice(self.to_le_bytes().as_ref()))
+        }
+      }
+
+      impl TypeRef<'_> for $ty {
+        #[inline]
+        fn from_slice(buf: &[u8]) -> Self {
+          const SIZE: usize = core::mem::size_of::<$ty>();
+
+          $ty::from_le_bytes(buf[..SIZE].try_into().unwrap())
+        }
+      }
+
+      impl KeyRef<'_, $ty> for $ty {
+        #[inline]
+        fn compare<Q>(&self, a: &Q) -> core::cmp::Ordering
+        where
+          Q: ?Sized + Ord + Comparable<$ty> {
+          Comparable::compare(a, self).reverse()
+        }
+
+        #[inline]
+        fn compare_binary(a: &[u8], b: &[u8]) -> core::cmp::Ordering {
+          const SIZE: usize = core::mem::size_of::<$ty>();
+
+          let a = $ty::from_le_bytes(a[..SIZE].try_into().unwrap());
+          let b = $ty::from_le_bytes(b[..SIZE].try_into().unwrap());
+
+          a.cmp(&b)
+        }
+      }
+    )*
+  };
+}
+
+impl_numbers!(i8, i16, i32, i64, i128, u8, u16, u32, u64, u128,);
diff --git a/src/swmr/generic/traits/impls/string.rs b/src/swmr/generic/traits/impls/string.rs
@@ -109,12 +109,6 @@ impl Borrow<str> for Str<'_> {
   }
 }
 
-// impl<'a> Borrow<&'a str> for Str<'a> {
-//   fn borrow(&self) -> &&'a str {
-//     &self.0
-//   }
-// }
-
 impl core::ops::Deref for Str<'_> {
   type Target = str;