Remove zstd in anticipation for it being re-added through marble

Author: spacejam

.gitignore

@@ -1,4 +1,5 @@
 default.sled
+crash_*
 *db
 *conf
 *snap.*

Cargo.toml

@@ -31,8 +31,7 @@ default = []
 # test-only configurations that cause performance to drop significantly.
 # It will cause your tests to take much more time, and possibly time out etc...
 testing = ["event_log", "lock_free_delays", "light_testing"]
-light_testing = ["compression", "failpoints", "backtrace", "memshred"]
-compression = ["zstd"]
+light_testing = ["failpoints", "backtrace", "memshred"]
 lock_free_delays = []
 failpoints = []
 event_log = []
@@ -41,20 +40,20 @@ no_logs = ["log/max_level_off"]
 no_inline = []
 pretty_backtrace = ["color-backtrace"]
 docs = []
+no_zstd = []
 miri_optimizations = []
 mutex = []
 memshred = []
 
 [dependencies]
 libc = "0.2.96"
-zstd = { version = "0.11.2", optional = true }
 crc32fast = "1.2.1"
 log = "0.4.14"
-parking_lot = "0.12.0"
+parking_lot = "0.12.1"
 color-backtrace = { version = "0.5.1", optional = true }
 num-format = { version = "0.4.0", optional = true }
 backtrace = { version = "0.3.60", optional = true }
-im = "15.0.0"
+im = "15.1.0"
 
 [target.'cfg(any(target_os = "linux", target_os = "macos", target_os="windows"))'.dependencies]
 fs2 = "0.4.3"

src/config.rs

@@ -481,8 +481,8 @@ impl Config {
         );
         if self.use_compression {
             supported!(
-                cfg!(feature = "compression"),
-                "the 'compression' feature must be enabled"
+                !cfg!(feature = "no_zstd"),
+                "the 'no_zstd' feature is set, but Config.use_compression is also set to true"
             );
         }
         supported!(

src/doc/mod.rs

@@ -9,8 +9,7 @@
 //! * forward and reverse iterators
 //! * a monotonic ID generator capable of giving out 75-125+ million unique IDs
 //!   per second, never double allocating even in the presence of crashes
-//! * [zstd](https://github.com/facebook/zstd) compression (use the zstd build
-//!   feature)
+//! * [zstd](https://github.com/facebook/zstd) compression
 //! * cpu-scalable lock-free implementation
 //! * SSD-optimized log-structured storage
 //!

src/metrics.rs

@@ -320,14 +320,10 @@ impl Metrics {
         ret.push_str(&format!("hit ratio: {}%\n", hit_ratio));
 
         ret.push_str(&format!("{}\n", "-".repeat(134)));
-        ret.push_str("serialization and compression:\n");
+        ret.push_str("serialization:\n");
         ret.push_str(&p(vec![
             lat("serialize", &self.serialize),
             lat("deserialize", &self.deserialize),
-            #[cfg(feature = "compression")]
-            lat("compress", &self.compress),
-            #[cfg(feature = "compression")]
-            lat("decompress", &self.decompress),
         ]));
 
         ret.push_str(&format!("{}\n", "-".repeat(134)));

src/pagecache/heap.rs

@@ -234,18 +234,10 @@ impl Heap {
         }
     }
 
-    pub fn read(
-        &self,
-        heap_id: HeapId,
-        use_compression: bool,
-    ) -> Result<(MessageKind, Vec<u8>)> {
+    pub fn read(&self, heap_id: HeapId) -> Result<(MessageKind, Vec<u8>)> {
         log::trace!("Heap::read({:?})", heap_id);
         let (slab_id, slab_idx, original_lsn) = heap_id.decompose();
-        self.slabs[slab_id as usize].read(
-            slab_idx,
-            original_lsn,
-            use_compression,
-        )
+        self.slabs[slab_id as usize].read(slab_idx, original_lsn)
     }
 
     pub fn free(&self, heap_id: HeapId) {
@@ -300,7 +292,6 @@ impl Slab {
         &self,
         slab_idx: SlabIdx,
         original_lsn: Lsn,
-        use_compression: bool,
     ) -> Result<(MessageKind, Vec<u8>)> {
         let bs = slab_id_to_size(self.slab_id);
         let offset = u64::from(slab_idx) * bs;
@@ -332,12 +323,7 @@ impl Slab {
                 return Err(Error::corruption(None));
             }
             let buf = heap_buf[13..].to_vec();
-            let buf2 = if use_compression {
-                crate::pagecache::decompress(buf)
-            } else {
-                buf
-            };
-            Ok((MessageKind::from(heap_buf[0]), buf2))
+            Ok((MessageKind::from(heap_buf[0]), buf))
         } else {
             log::debug!(
                 "heap message CRC does not match contents. stored: {} actual: {}",

src/pagecache/logger.rs

@@ -1,11 +1,11 @@
 use std::fs::File;
 
 use super::{
-    arr_to_lsn, arr_to_u32, assert_usize, decompress, header, iobuf,
-    lsn_to_arr, pread_exact, pread_exact_or_eof, roll_iobuf, u32_to_arr, Arc,
-    BasedBuf, DiskPtr, HeapId, IoBuf, IoBufs, LogKind, LogOffset, Lsn,
-    MessageKind, Reservation, Serialize, Snapshot, BATCH_MANIFEST_PID,
-    COUNTER_PID, MAX_MSG_HEADER_LEN, META_PID, SEG_HEADER_LEN,
+    arr_to_lsn, arr_to_u32, assert_usize, header, iobuf, lsn_to_arr,
+    pread_exact, pread_exact_or_eof, roll_iobuf, u32_to_arr, Arc, BasedBuf,
+    DiskPtr, HeapId, IoBuf, IoBufs, LogKind, LogOffset, Lsn, MessageKind,
+    Reservation, Serialize, Snapshot, BATCH_MANIFEST_PID, COUNTER_PID,
+    MAX_MSG_HEADER_LEN, META_PID, SEG_HEADER_LEN,
 };
 
 use crate::*;
@@ -70,18 +70,16 @@ impl Log {
             // here because it might not still
             // exist in the inline log.
             let heap_id = ptr.heap_id().unwrap();
-            self.config.heap.read(heap_id, self.config.use_compression).map(
-                |(kind, buf)| {
-                    let header = MessageHeader {
-                        kind,
-                        pid,
-                        segment_number: expected_segment_number,
-                        crc32: 0,
-                        len: 0,
-                    };
-                    LogRead::Heap(header, buf, heap_id, 0)
-                },
-            )
+            self.config.heap.read(heap_id).map(|(kind, buf)| {
+                let header = MessageHeader {
+                    kind,
+                    pid,
+                    segment_number: expected_segment_number,
+                    crc32: 0,
+                    len: 0,
+                };
+                LogRead::Heap(header, buf, heap_id, 0)
+            })
         }
     }
 
@@ -129,43 +127,13 @@ impl Log {
     /// completed or aborted later. Useful for maintaining
     /// linearizability across CAS operations that may need to
     /// persist part of their operation.
-    #[allow(unused)]
     pub fn reserve<T: Serialize + Debug>(
         &self,
         log_kind: LogKind,
         pid: PageId,
         item: &T,
         guard: &Guard,
     ) -> Result<Reservation<'_>> {
-        #[cfg(feature = "compression")]
-        {
-            if self.config.use_compression && pid != BATCH_MANIFEST_PID {
-                use zstd::bulk::compress;
-
-                let buf = item.serialize();
-
-                #[cfg(feature = "metrics")]
-                let _measure = Measure::new(&M.compress);
-
-                let compressed_buf =
-                    compress(&buf, self.config.compression_factor).unwrap();
-
-                let ret = self.reserve_inner(
-                    log_kind,
-                    pid,
-                    &IVec::from(compressed_buf),
-                    None,
-                    guard,
-                );
-
-                if let Err(e) = &ret {
-                    self.iobufs.set_global_error(*e);
-                }
-
-                return ret;
-            }
-        }
-
         let ret = self.reserve_inner(log_kind, pid, item, None, guard);
 
         if let Err(e) = &ret {
@@ -860,7 +828,7 @@ pub(crate) fn read_message<R: ReadAt>(
             assert_eq!(buf.len(), 16);
             let heap_id = HeapId::deserialize(&mut &buf[..]).unwrap();
 
-            match config.heap.read(heap_id, config.use_compression) {
+            match config.heap.read(heap_id) {
                 Ok((kind, buf2)) => {
                     assert_eq!(header.kind, kind);
                     trace!(
@@ -883,10 +851,7 @@ pub(crate) fn read_message<R: ReadAt>(
         | MessageKind::Free
         | MessageKind::Counter => {
             trace!("read a successful inline message");
-            let buf2 =
-                if config.use_compression { decompress(buf) } else { buf };
-
-            Ok(LogRead::Inline(header, buf2, inline_len))
+            Ok(LogRead::Inline(header, buf, inline_len))
         }
         MessageKind::BatchManifest => {
             assert_eq!(buf.len(), std::mem::size_of::<Lsn>());

src/pagecache/mod.rs

@@ -230,32 +230,6 @@ pub(crate) const fn u32_to_arr(number: u32) -> [u8; 4] {
     number.to_le_bytes()
 }
 
-#[allow(clippy::needless_pass_by_value)]
-#[allow(clippy::needless_return)]
-pub(in crate::pagecache) fn decompress(in_buf: Vec<u8>) -> Vec<u8> {
-    #[cfg(feature = "compression")]
-    {
-        use zstd::stream::decode_all;
-
-        let scootable_in_buf = &mut &*in_buf;
-        let raw: IVec = IVec::deserialize(scootable_in_buf)
-            .expect("this had to be serialized with an extra length frame");
-        #[cfg(feature = "metrics")]
-        let _measure = Measure::new(&M.decompress);
-        let out_buf = decode_all(&raw[..]).expect(
-            "failed to decompress data. \
-             This is not expected, please open an issue on \
-             https://github.com/spacejam/sled so we can \
-             fix this critical issue ASAP. Thank you :)",
-        );
-
-        return out_buf;
-    }
-
-    #[cfg(not(feature = "compression"))]
-    in_buf
-}
-
 #[derive(Debug, Clone, Copy)]
 pub struct NodeView<'g>(pub(crate) PageView<'g>);
 

src/pagecache/reservation.rs

@@ -56,8 +56,7 @@ impl<'a> Reservation<'a> {
 
     /// Refills the reservation buffer with new data.
     /// Must supply a buffer of an identical length
-    /// as the one initially provided. Don't use this
-    /// on messages subject to compression etc...
+    /// as the one initially provided.
     ///
     /// # Panics
     ///

src/pagecache/snapshot.rs

@@ -1,6 +1,3 @@
-#[cfg(feature = "zstd")]
-use zstd::bulk::{compress, decompress};
-
 use crate::*;
 
 use super::{
@@ -530,22 +527,7 @@ fn read_snapshot(config: &RunningConfig) -> Result<Option<Snapshot>> {
         return Err(Error::corruption(None));
     }
 
-    #[cfg(feature = "zstd")]
-    let bytes = if config.use_compression {
-        use std::convert::TryInto;
-
-        let len_expected: u64 =
-            u64::from_le_bytes(len_expected_bytes.as_ref().try_into().unwrap());
-
-        decompress(&*buf, usize::try_from(len_expected).unwrap()).unwrap()
-    } else {
-        buf
-    };
-
-    #[cfg(not(feature = "zstd"))]
-    let bytes = buf;
-
-    Snapshot::deserialize(&mut bytes.as_slice()).map(Some)
+    Snapshot::deserialize(&mut buf.as_slice()).map(Some)
 }
 
 pub(in crate::pagecache) fn write_snapshot(
@@ -554,21 +536,10 @@ pub(in crate::pagecache) fn write_snapshot(
 ) -> Result<()> {
     trace!("writing snapshot {:?}", snapshot);
 
-    let raw_bytes = snapshot.serialize();
-    let decompressed_len = raw_bytes.len();
-
-    #[cfg(feature = "zstd")]
-    let bytes = if config.use_compression {
-        compress(&*raw_bytes, config.compression_factor).unwrap()
-    } else {
-        raw_bytes
-    };
-
-    #[cfg(not(feature = "zstd"))]
-    let bytes = raw_bytes;
+    let bytes = snapshot.serialize();
 
     let crc32: [u8; 4] = u32_to_arr(crc32(&bytes));
-    let len_bytes: [u8; 8] = u64_to_arr(decompressed_len as u64);
+    let len_bytes: [u8; 8] = u64_to_arr(bytes.len() as u64);
 
     let path_1_suffix =
         format!("snap.{:016X}.generating", snapshot.stable_lsn.unwrap_or(0));