prepare for 1.0

Author: marvin-j97

src/key_range.rs

@@ -0,0 +1,60 @@
+use crate::{
+    serde::{Deserializable, DeserializeError, Serializable, SerializeError},
+    value::UserKey,
+};
+use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
+use std::{
+    io::{Read, Write},
+    ops::Deref,
+    sync::Arc,
+};
+
+/// A key range in the format of [min, max] (inclusive on both sides)
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct KeyRange((UserKey, UserKey));
+
+impl std::ops::Deref for KeyRange {
+    type Target = (UserKey, UserKey);
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl KeyRange {
+    pub fn new(range: (UserKey, UserKey)) -> Self {
+        Self(range)
+    }
+}
+
+impl Serializable for KeyRange {
+    fn serialize<W: Write>(&self, writer: &mut W) -> Result<(), SerializeError> {
+        // NOTE: Max key size = u16
+        #[allow(clippy::cast_possible_truncation)]
+        writer.write_u16::<BigEndian>(self.deref().0.len() as u16)?;
+        writer.write_all(&self.deref().0)?;
+
+        // NOTE: Max key size = u16
+        #[allow(clippy::cast_possible_truncation)]
+        writer.write_u16::<BigEndian>(self.deref().1.len() as u16)?;
+        writer.write_all(&self.deref().1)?;
+
+        Ok(())
+    }
+}
+
+impl Deserializable for KeyRange {
+    fn deserialize<R: Read>(reader: &mut R) -> Result<Self, DeserializeError> {
+        let key_min_len = reader.read_u16::<BigEndian>()?;
+        let mut key_min = vec![0; key_min_len.into()];
+        reader.read_exact(&mut key_min)?;
+        let key_min: UserKey = Arc::from(key_min);
+
+        let key_max_len = reader.read_u16::<BigEndian>()?;
+        let mut key_max = vec![0; key_max_len.into()];
+        reader.read_exact(&mut key_max)?;
+        let key_max: UserKey = Arc::from(key_max);
+
+        Ok(Self::new((key_min, key_max)))
+    }
+}

src/lib.rs

@@ -93,6 +93,7 @@ mod error;
 mod handle;
 mod id;
 mod index;
+mod key_range;
 mod manifest;
 mod mock;
 mod path;

src/manifest.rs

@@ -1,14 +1,15 @@
 use crate::{
     id::SegmentId,
-    segment::stats::{SegmentFileTrailer, Stats},
+    key_range::KeyRange,
+    segment::{gc_stats::GcStats, meta::Metadata, trailer::SegmentFileTrailer},
     Segment, SegmentWriter as MultiWriter,
 };
 use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
 use std::{
     collections::HashMap,
     io::{Cursor, Write},
     path::{Path, PathBuf},
-    sync::{atomic::AtomicU64, Arc, RwLock},
+    sync::{Arc, RwLock},
 };
 
 pub const VLOG_MARKER: &str = ".vlog";
@@ -125,11 +126,8 @@ impl SegmentManifest {
                     Arc::new(Segment {
                         id,
                         path,
-                        stats: Stats {
-                            persisted: trailer,
-                            stale_bytes: AtomicU64::default(),
-                            stale_items: AtomicU64::default(),
-                        },
+                        meta: trailer.metadata,
+                        gc_stats: GcStats::default(),
                     }),
                 );
             }
@@ -191,15 +189,22 @@ impl SegmentManifest {
                 Arc::new(Segment {
                     id: segment_id,
                     path: writer.path,
-                    stats: Stats {
-                        persisted: SegmentFileTrailer {
-                            item_count: writer.item_count,
-                            total_bytes: writer.written_blob_bytes,
-                            total_uncompressed_bytes: writer.uncompressed_bytes,
-                        },
-                        stale_items: AtomicU64::default(),
-                        stale_bytes: AtomicU64::default(),
+                    meta: Metadata {
+                        item_count: writer.item_count,
+                        total_bytes: writer.written_blob_bytes,
+                        total_uncompressed_bytes: writer.uncompressed_bytes,
+                        key_range: KeyRange::new((
+                            writer
+                                .first_key
+                                .clone()
+                                .expect("should have written at least 1 item"),
+                            writer
+                                .last_key
+                                .clone()
+                                .expect("should have written at least 1 item"),
+                        )),
                     },
+                    gc_stats: GcStats::default(),
                 }),
             );
 
@@ -276,7 +281,7 @@ impl SegmentManifest {
             .read()
             .expect("lock is poisoned")
             .values()
-            .map(|x| x.stats.total_bytes)
+            .map(|x| x.meta.total_bytes)
             .sum::<u64>()
     }
 
@@ -287,7 +292,7 @@ impl SegmentManifest {
             .read()
             .expect("lock is poisoned")
             .values()
-            .map(|x| x.stats.stale_items())
+            .map(|x| x.meta.stale_items())
             .sum::<u64>()
     } */
 
@@ -299,7 +304,7 @@ impl SegmentManifest {
             .read()
             .expect("lock is poisoned")
             .values()
-            .map(|x| x.stats.total_uncompressed_bytes)
+            .map(|x| x.meta.total_uncompressed_bytes)
             .sum::<u64>();
         if used_bytes == 0 {
             return 0.0;
@@ -310,7 +315,7 @@ impl SegmentManifest {
             .read()
             .expect("lock is poisoned")
             .values()
-            .map(|x| x.stats.stale_bytes())
+            .map(|x| x.gc_stats.stale_bytes())
             .sum::<u64>();
 
         if stale_bytes == 0 {
@@ -330,7 +335,7 @@ impl SegmentManifest {
             .read()
             .expect("lock is poisoned")
             .values()
-            .map(|x| x.stats.total_uncompressed_bytes)
+            .map(|x| x.meta.total_uncompressed_bytes)
             .sum::<u64>();
 
         if used_bytes == 0 {
@@ -342,7 +347,7 @@ impl SegmentManifest {
             .read()
             .expect("lock is poisoned")
             .values()
-            .map(|x| x.stats.stale_bytes())
+            .map(|x| x.gc_stats.stale_bytes())
             .sum::<u64>();
 
         let alive_bytes = used_bytes - stale_bytes;

src/segment/gc_stats.rs

@@ -0,0 +1,30 @@
+use std::sync::atomic::AtomicU64;
+
+#[derive(Debug, Default)]
+#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
+pub struct GcStats {
+    pub(crate) stale_items: AtomicU64,
+    pub(crate) stale_bytes: AtomicU64,
+}
+
+impl GcStats {
+    pub fn set_stale_items(&self, x: u64) {
+        self.stale_items
+            .store(x, std::sync::atomic::Ordering::Release)
+    }
+
+    pub fn set_stale_bytes(&self, x: u64) {
+        self.stale_bytes
+            .store(x, std::sync::atomic::Ordering::Release)
+    }
+
+    /// Returns the amount of dead items in the segment
+    pub fn stale_items(&self) -> u64 {
+        self.stale_items.load(std::sync::atomic::Ordering::Acquire)
+    }
+
+    /// Returns the amount of dead bytes in the segment
+    pub fn stale_bytes(&self) -> u64 {
+        self.stale_bytes.load(std::sync::atomic::Ordering::Acquire)
+    }
+}

src/segment/meta.rs

@@ -0,0 +1,67 @@
+use crate::{
+    key_range::KeyRange,
+    serde::{Deserializable, DeserializeError, Serializable, SerializeError},
+};
+use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
+use std::io::{Read, Write};
+
+pub const METADATA_HEADER_MAGIC: &[u8] = &[b'F', b'J', b'L', b'L', b'S', b'M', b'D', b'1'];
+
+#[derive(Debug)]
+#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
+pub struct Metadata {
+    /// Number of KV-pairs in the segment
+    pub item_count: u64,
+
+    /// compressed size in bytes (on disk) (without the fixed size trailer)
+    pub total_bytes: u64,
+
+    /// true size in bytes (if no compression were used)
+    pub total_uncompressed_bytes: u64,
+
+    // TODO:
+    ///// What type of compression is used
+    // pub compression: CompressionType,
+    /// Key range
+    pub key_range: KeyRange,
+}
+
+impl Serializable for Metadata {
+    fn serialize<W: Write>(&self, writer: &mut W) -> Result<(), SerializeError> {
+        // Write header
+        writer.write_all(METADATA_HEADER_MAGIC)?;
+
+        writer.write_u64::<BigEndian>(self.item_count)?;
+        writer.write_u64::<BigEndian>(self.total_bytes)?;
+        writer.write_u64::<BigEndian>(self.total_uncompressed_bytes)?;
+
+        self.key_range.serialize(writer)?;
+
+        Ok(())
+    }
+}
+
+impl Deserializable for Metadata {
+    fn deserialize<R: Read>(reader: &mut R) -> Result<Self, DeserializeError> {
+        // Check header
+        let mut magic = [0u8; METADATA_HEADER_MAGIC.len()];
+        reader.read_exact(&mut magic)?;
+
+        if magic != METADATA_HEADER_MAGIC {
+            return Err(DeserializeError::InvalidHeader("SegmentMetadata"));
+        }
+
+        let item_count = reader.read_u64::<BigEndian>()?;
+        let total_bytes = reader.read_u64::<BigEndian>()?;
+        let total_uncompressed_bytes = reader.read_u64::<BigEndian>()?;
+
+        let key_range = KeyRange::deserialize(reader)?;
+
+        Ok(Self {
+            item_count,
+            total_bytes,
+            total_uncompressed_bytes,
+            key_range,
+        })
+    }
+}

src/segment/mod.rs

@@ -1,11 +1,14 @@
+pub mod gc_stats;
 pub mod merge;
+pub mod meta;
 pub mod multi_writer;
 pub mod reader;
-pub mod stats;
+pub mod trailer;
 pub mod writer;
 
-use self::stats::Stats;
 use crate::id::SegmentId;
+use gc_stats::GcStats;
+use meta::Metadata;
 use std::path::PathBuf;
 
 /// A disk segment is an immutable, sorted, contiguous file
@@ -19,26 +22,52 @@ pub struct Segment {
     pub path: PathBuf,
 
     /// Segment statistics
-    pub stats: Stats,
+    pub meta: Metadata,
+
+    /// Runtime stats for garbage collection
+    pub gc_stats: GcStats,
 }
 
 impl Segment {
-    /// Returns a scanner that can iterate through the segment
+    /// Returns a scanner that can iterate through the segment.
     ///
     /// # Errors
     ///
     /// Will return `Err` if an IO error occurs.
     pub fn scan(&self) -> std::io::Result<reader::Reader> {
-        reader::Reader::new(&self.path, self.id, self.stats.item_count)
+        reader::Reader::new(&self.path, self.id, self.meta.item_count)
     }
 
-    /// Always returns `false`
+    /// Always returns `false` because a segment is never empty.
     pub fn is_empty(&self) -> bool {
         false
     }
 
-    /// Returns the amount of items in the segment
+    /// Returns the amount of items in the segment.
     pub fn len(&self) -> u64 {
-        self.stats.item_count
+        self.meta.item_count
+    }
+
+    /// Marks the segment as fully stale.
+    pub(crate) fn mark_as_stale(&self) {
+        self.gc_stats.set_stale_items(self.meta.item_count);
+
+        self.gc_stats
+            .set_stale_bytes(self.meta.total_uncompressed_bytes);
+    }
+
+    /// Returns `true` if the segment is fully stale.
+    pub fn is_stale(&self) -> bool {
+        self.gc_stats.stale_items() == self.meta.item_count
+    }
+
+    /// Returns the percent of dead items in the segment.
+    pub fn stale_ratio(&self) -> f32 {
+        let dead = self.gc_stats.stale_items() as f32;
+        if dead == 0.0 {
+            return 0.0;
+        }
+
+        dead / self.meta.item_count as f32
     }
 }