NVM-opt tree with compression

betree/Cargo.toml

@@ -21,7 +21,12 @@ harness = false
 name = "tree"
 harness = false
 
+#[workspace.dependencies]
+#rkyv = { version = "0.7.42", features = ["validation"] }
+#betree-tests = { version = "0.1.0"}
+
 [dependencies]
+once_cell = "1.10.0"
 futures = { version = "0.3", features = ["thread-pool"] }
 serde = { version = "1.0", features = [ "derive" ] }
 bincode = "1.0"
@@ -42,6 +47,7 @@ core_affinity = "0.5"
 async-trait = "0.1"
 
 lz4-sys = "1.9"
+lz4 = "1.23.1"
 zstd = { version = "0.9", default-features = false }
 zstd-safe = { version = "4.0", default-features = false, features = ["experimental"] }
 
@@ -60,6 +66,20 @@ rand = { version = "0.8", features = ["std_rng"] }
 
 pmdk = { path = "./pmdk", optional = true }
 
+rkyv = { version = "0.7.42", features = ["validation"] }
+#rkyv = { path = "../rkyv/rkyv"}
+#rkyv_dyn = { path = "../rkyv/rkyv_dyn"}
+#rkyv_dyn_test = { path = "../rkyv/rkyv_dyn_test"}
+#rkyv_test = { path = "../rkyv/rkyv_test"}
+#rkyv_typename = { path = "../rkyv/rkyv_typename"}
+#betree-tests = { path = "./tests"}
+
+bytecheck = { version = "0.7.0" }
+extend = { version = "1.2.0" }
+
+chrono = "0.4"
+
+lazy_static = "1.4"
 [dev-dependencies]
 rand_xorshift = "0.3"
 quickcheck = "1"

betree/pmdk/src/lib.rs

@@ -4,6 +4,7 @@
 
 include!(concat!(env!("OUT_DIR"), "/bindings.rs"));
 
+use std::slice;
 use std::os::raw::c_void;
 
 #[derive(Debug)]
@@ -14,6 +15,11 @@ pub struct PMem {
 unsafe impl Send for PMem {}
 unsafe impl Sync for PMem {}
 
+unsafe fn voidp_to_ref<'a, T>(p: *const c_void) -> &'a T
+{
+    unsafe { &*(p as *const T) }
+}
+
 impl PMem {
     pub fn create(filepath : &str, len: u64, mapped_len : &mut u64, is_pmem : &mut i32) -> Result<Self, std::io::Error> {
         let mut ptr = unsafe {
@@ -73,6 +79,15 @@ impl PMem {
         Ok(())
     }
 
+    pub unsafe fn get_slice(&self, offset: usize, len: usize) -> Result<&'static [u8], std::io::Error>{
+        if self.ptr.is_null() {
+            return Err(std::io::Error::new(std::io::ErrorKind::Other,
+                        format!("File handle is missing for the PMEM file.")));
+        }
+
+        Ok(slice::from_raw_parts(voidp_to_ref::<u8>(self.ptr.add(offset)), len))
+    }
+
     pub unsafe fn write(&self, offset: usize, data: &[u8], len: usize) -> Result<(), std::io::Error>{
         if self.ptr.is_null() {
             return Err(std::io::Error::new(std::io::ErrorKind::Other,

betree/src/buffer.rs

@@ -249,6 +249,10 @@ pub struct BufWrite {
 }
 
 impl BufWrite {
+    pub fn get_size(&self) -> u32 {
+        self.size
+    }
+
     /// Create an empty [BufWrite] with the specified capacity.
     /// The backing storage is zeroed.
     pub fn with_capacity(capacity: Block<u32>) -> Self {
@@ -290,6 +294,23 @@ impl io::Write for BufWrite {
     }
 }
 
+
+// Implementation of WriteBuf trait for BufWrite with methods for buffer manipulation
+unsafe impl zstd::stream::raw::WriteBuf for BufWrite {
+    fn as_slice(&self) -> &[u8] {
+        self.as_ref()
+    }
+    fn capacity(&self) -> usize {
+        self.buf.capacity.to_bytes() as usize
+    }
+    fn as_mut_ptr(&mut self) -> *mut u8 {
+        unsafe { self.buf.ptr.as_mut().unwrap() }
+    }
+    unsafe fn filled_until(&mut self, n: usize) {
+        self.size = n as u32
+    }
+}
+
 impl io::Seek for BufWrite {
     fn seek(&mut self, seek: io::SeekFrom) -> io::Result<u64> {
         use io::SeekFrom::*;

betree/src/checksum.rs

@@ -1,13 +1,23 @@
 //! This module provides a `Checksum` trait for verifying data integrity.
 
 use crate::size::{Size, StaticSize};
-use serde::{de::DeserializeOwned, Deserialize, Serialize};
+// Serde-related functionality is used using namespace resolution operator
+//use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use std::{error::Error, fmt, hash::Hasher, iter::once};
 use twox_hash;
 
+use rkyv::{
+    archived_root,
+    ser::{serializers::AllocSerializer, ScratchSpace, Serializer},
+    vec::{ArchivedVec, VecResolver},
+    with::{ArchiveWith, DeserializeWith, SerializeWith},
+    Archive, Archived, Deserialize, Fallible, Infallible, Serialize, AlignedVec,
+};
+
+
 /// A checksum to verify data integrity.
 pub trait Checksum:
-    Serialize + DeserializeOwned + Size + Clone + Send + Sync + fmt::Debug + 'static
+serde::Serialize + serde::de::DeserializeOwned + Size + Clone + Send + Sync + fmt::Debug + 'static
 {
     /// Builds a new `Checksum`.
     type Builder: Builder<Self>;
@@ -27,7 +37,7 @@ pub trait Checksum:
 
 /// A checksum builder
 pub trait Builder<C: Checksum>:
-    Serialize + DeserializeOwned + Clone + Send + Sync + fmt::Debug + 'static
+serde::Serialize + serde::de::DeserializeOwned + Clone + Send + Sync + fmt::Debug + 'static
 {
     /// The internal state of the checksum.
     type State: State<Checksum = C>;
@@ -67,7 +77,8 @@ impl Error for ChecksumError {
 /// `XxHash` contains a digest of `xxHash`
 /// which is an "extremely fast non-cryptographic hash algorithm"
 /// (<https://github.com/Cyan4973/xxHash>)
-#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(serde::Serialize, serde::Deserialize, Clone, Copy, Debug, PartialEq, Eq, rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)]
+#[archive(check_bytes)]
 pub struct XxHash(u64);
 
 impl StaticSize for XxHash {
@@ -97,7 +108,7 @@ impl Checksum for XxHash {
 }
 
 /// The corresponding `Builder` for `XxHash`.
-#[derive(Clone, Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct XxHashBuilder;
 
 impl Builder<XxHash> for XxHashBuilder {

betree/src/compression/lz4.rs

@@ -1,11 +1,22 @@
-use super::{ CompressionConfiguration, CompressionState, DecompressionState, DecompressionTag, DEFAULT_BUFFER_SIZE, Result };
+use super::{ CompressionBuilder, CompressionState, DecompressionState, DecompressionTag, DEFAULT_BUFFER_SIZE, Result };
 use crate::size::StaticSize;
 use crate::buffer::{Buf, BufWrite};
 
+use crate::{
+    vdev::Block,
+};
+use std::io::Write;
+
 use serde::{Deserialize, Serialize};
-use std::io::{self, Read};
+use zstd_safe::WriteBuf;
+use std::io::{self, BufReader, Read};
+
+use std::{
+    mem,
+};
+use std::sync::{Arc, Mutex};
 
-// use lz4_sys::{ Lz4
+use lz4::{Encoder, Decoder, EncoderBuilder, ContentChecksum, BlockSize, BlockMode};
 
 /// LZ4 compression. (<https://github.com/lz4/lz4>)
 #[derive(Debug, Serialize, Deserialize, Clone, Copy)]
@@ -25,24 +36,27 @@ pub struct Lz4Compression {
 pub struct Lz4Decompression;
 
 impl StaticSize for Lz4 {
-    fn size() -> usize {
+    fn static_size() -> usize {
         1
     }
 }
 
-impl CompressionConfiguration for Lz4 {
-    fn new_compression(&self) -> Result<Box<dyn CompressionState>> {
-        let encoder = EncoderBuilder::new()
+impl CompressionBuilder for Lz4 {
+    fn new_compression(&self) -> Result<Arc<std::sync::RwLock<dyn CompressionState>>> {
+        let mut encoder = EncoderBuilder::new()
             .level(u32::from(self.level))
             .checksum(ContentChecksum::NoChecksum)
             .block_size(BlockSize::Max4MB)
             .block_mode(BlockMode::Linked)
             .build(BufWrite::with_capacity(DEFAULT_BUFFER_SIZE))?;
 
-        Ok(Box::new(Lz4Compression { config: self.clone(), encoder }))
+        Ok(Arc::new(std::sync::RwLock::new(Lz4Compression { config: self.clone(), encoder })))
+
     }
 
-    fn decompression_tag(&self) -> DecompressionTag { DecompressionTag::Lz4 }
+    fn decompression_tag(&self) -> DecompressionTag {
+        DecompressionTag::Lz4
+    }
 }
 
 impl Lz4 {
@@ -53,30 +67,95 @@ impl Lz4 {
 
 impl io::Write for Lz4Compression {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        self.encoder.write(buf)
+        unimplemented!()
     }
 
     fn write_all(&mut self, buf: &[u8]) -> io::Result<()> {
-        self.encoder.write_all(buf)
+        unimplemented!()
     }
 
     fn flush(&mut self) -> io::Result<()> {
-        self.encoder.flush()
+        unimplemented!()
     }
 }
 
+use std::time::Instant;
+use speedy::{Readable, Writable};
+const DATA_OFF: usize = mem::size_of::<u32>();
+
+use lz4_sys::{LZ4F_compressBound, LZ4FPreferences, LZ4FCompressionContext, LZ4F_createCompressionContext};
+use std::ptr;
+use lz4_sys::LZ4FFrameInfo;
+
 impl CompressionState for Lz4Compression {
-    fn finish(&mut self) -> Buf {
-        let (v, result) = self.encoder.finish();
-        result.unwrap();
-        v.into_buf()
+    fn finish_ext(&mut self, data: &[u8]) -> Result<Vec<u8>>
+    {
+        let size = data.len();
+        let mut buf = BufWrite::with_capacity(Block::round_up_from_bytes(size as u32));
+
+        let mut encoder = EncoderBuilder::new()
+        .level(u32::from(self.config.level))
+        .checksum(ContentChecksum::NoChecksum)
+        .block_size(BlockSize::Max4MB)
+        .block_mode(BlockMode::Linked)
+        .build(buf)?;
+
+        encoder.write_all(data.as_ref())?;
+        let (compressed_data, result) = encoder.finish();
+
+        if let Err(e) = result {
+            panic!("Compression failed: {:?}", e);
+        }
+
+        let mut buf_opt = BufWrite::with_capacity(Block::round_up_from_bytes(compressed_data.as_slice().len() as u32));
+        buf_opt.write_all(compressed_data.as_slice());
+
+        Ok(buf_opt.as_slice().to_vec())
+    }
+
+    fn finish(&mut self, data: Buf) -> Result<Buf> {
+        let size = data.as_ref().len();
+
+        let mut buf: BufWrite = BufWrite::with_capacity(Block::round_up_from_bytes( (size as u32)));
+
+        let mut encoder = EncoderBuilder::new()
+        .level(u32::from(self.config.level))
+        .checksum(ContentChecksum::NoChecksum)
+        .block_size(BlockSize::Max4MB)
+        .block_mode(BlockMode::Linked)
+        .build(buf)?;
+
+        encoder.write_all(data.as_ref())?;
+        let (compressed_data, result) = encoder.finish();
+
+        if let Err(e) = result {
+            panic!("Compression failed: {:?}", e);
+        }
+
+        let mut buf_opt = BufWrite::with_capacity(Block::round_up_from_bytes(compressed_data.as_slice().len() as u32));
+        buf_opt.write_all(compressed_data.as_slice());
+
+        Ok(buf_opt.into_buf())
     }
 }
 
 impl DecompressionState for Lz4Decompression {
-    fn decompress(&mut self, data: &[u8]) -> Result<Box<[u8]>> {
-        let mut output = Vec::with_capacity(DEFAULT_BUFFER_SIZE.to_bytes() as usize);
-        Decoder::new(&data[..])?.read_to_end(&mut output)?;
-        Ok(output.into_boxed_slice())
+    fn decompress_ext(&mut self, data: &[u8]) -> Result<Vec<u8>>
+    {
+        let size = data.as_ref().len() as u32;
+        let mut buf = BufWrite::with_capacity(Block::round_up_from_bytes(size));
+        let mut decoder = Decoder::new(data.as_ref())?;
+
+        io::copy(&mut decoder, &mut buf)?;
+        Ok(buf.as_slice().to_vec())
+    }
+
+    fn decompress(&mut self, data: Buf) -> Result<Buf> {
+        let size = data.as_ref().len() as u32;
+        let mut buf = BufWrite::with_capacity(Block::round_up_from_bytes(size));
+        let mut decoder = Decoder::new(data.as_ref())?;
+
+        io::copy(&mut decoder, &mut buf)?;
+        Ok(buf.into_buf())
     }
 }