Optimizations: set_len to max file size to save up on metadata updates; sync_file_range instead of all file

Author: tomerfiliba

Cargo.toml

@@ -10,6 +10,7 @@ repository = "https://github.com/sweet-security/candystore"
 [dependencies]
 crc16-ibm3740-fast = "0.5.0"
 fslock = "0.2.1"
+libc = "0.2.183"
 memmap2 = "0.9.10"
 num_cpus = "1.17.0"
 parking_lot = "0.12.5"
@@ -26,7 +27,6 @@ zerocopy = { version = "0.8.47", features = ["derive"] }
 proptest = "1.10.0"
 tempfile = "3"
 rand = "0.10.0"
-libc = "0.2.183"
 
 [features]
 whitebox-testing = []

src/data_file.rs

@@ -17,7 +17,7 @@ use crate::internal::{
     DATA_ENTRY_OFFSET_MAGIC, DATA_ENTRY_OFFSET_MASK, DATA_FILE_SIGNATURE, DATA_FILE_VERSION,
     EntryType, FILE_OFFSET_ALIGNMENT, KEY_NAMESPACE_BITS, KVBuf, KVRef, KeyNamespace,
     MAX_KEY_NAMESPACE, PAGE_SIZE, READ_BUFFER_SIZE, SIZE_HINT_UNIT, data_file_path,
-    invalid_data_error, read_available_at, read_into_at, sync_dir, write_all_at,
+    invalid_data_error, read_available_at, read_into_at, sync_dir, sync_file_range, write_all_at,
 };
 use crate::types::{Config, Error, MAX_USER_KEY_SIZE, MAX_USER_VALUE_SIZE, Result};
 
@@ -197,24 +197,124 @@ impl Drop for InflightGuard<'_> {
 pub(crate) struct DataFile {
     pub(crate) file: File,
     file_offset: AtomicU64,
+    last_synced_offset: AtomicU64,
     sealed_for_rotation: AtomicBool,
     config: Arc<Config>,
     pub(crate) file_idx: u16,
     pub(crate) file_ordinal: u64,
+    preallocated: bool,
+    recovery_tail_upper_bound: u64,
 }
 
 impl DataFile {
     pub(crate) fn used_bytes(&self) -> u64 {
         self.file_offset.load(Ordering::Acquire)
     }
 
+    pub(crate) fn recovery_tail_upper_bound(&self) -> u64 {
+        self.recovery_tail_upper_bound
+    }
+
+    pub(crate) fn sync_data(&self, start_offset: u64, end_offset: u64) -> Result<()> {
+        let used_bytes = self.used_bytes();
+        let start_offset = start_offset.min(used_bytes);
+        let end_offset = end_offset.min(used_bytes);
+        if end_offset <= start_offset {
+            return Ok(());
+        }
+
+        if !self.preallocated {
+            self.file.sync_all().map_err(Error::IOError)?;
+        } else {
+            sync_file_range(
+                &self.file,
+                size_of::<DataFileHeader>() as u64 + start_offset,
+                end_offset - start_offset,
+            )?;
+        }
+        self.last_synced_offset
+            .fetch_max(end_offset, Ordering::Release);
+        Ok(())
+    }
+
+    pub(crate) fn sync_to_current(&self) -> Result<()> {
+        let start = self.last_synced_offset.load(Ordering::Acquire);
+        self.sync_data(start, self.used_bytes())
+    }
+
     pub(crate) fn truncate_to_offset(&self, file_offset: u64) -> Result<()> {
         debug_assert_eq!(file_offset % FILE_OFFSET_ALIGNMENT, 0);
-        self.file
-            .set_len(size_of::<DataFileHeader>() as u64 + file_offset)
-            .map_err(Error::IOError)?;
+        if self.preallocated {
+            // A crash between the two set_len calls would leave the file
+            // non-preallocated.  That is harmless: the next open will
+            // detect it as non-preallocated and fall back to sync_all
+            // until rotation creates a fresh preallocated file.
+            self.file
+                .set_len(size_of::<DataFileHeader>() as u64 + file_offset)
+                .map_err(Error::IOError)?;
+            self.file
+                .set_len(size_of::<DataFileHeader>() as u64 + self.config.max_data_file_size as u64)
+                .map_err(Error::IOError)?;
+        } else {
+            self.file
+                .set_len(size_of::<DataFileHeader>() as u64 + file_offset)
+                .map_err(Error::IOError)?;
+        }
         self.file_offset.store(file_offset, Ordering::Release);
-        self.file.sync_all().map_err(Error::IOError)
+        self.file.sync_all().map_err(Error::IOError)?;
+        self.last_synced_offset
+            .store(file_offset, Ordering::Release);
+        Ok(())
+    }
+
+    fn used_data_upper_bound(file: &File, physical_data_len: u64) -> Result<u64> {
+        if physical_data_len == 0 {
+            return Ok(0);
+        }
+
+        let mut end = physical_data_len;
+        while end > 0 {
+            let start = end.saturating_sub(READ_BUFFER_SIZE as u64);
+            let chunk = read_available_at(
+                file,
+                (end - start) as usize,
+                size_of::<DataFileHeader>() as u64 + start,
+            )
+            .map_err(Error::IOError)?;
+            if let Some(rel) = chunk.iter().rposition(|byte| *byte != 0) {
+                let aligned = (start + rel as u64 + 1).next_multiple_of(FILE_OFFSET_ALIGNMENT);
+                return Ok(aligned.min(physical_data_len));
+            }
+            end = start;
+        }
+
+        Ok(0)
+    }
+
+    /// Scans forward from offset 0, parsing each entry, and returns the
+    /// aligned end of the last valid entry.  We temporarily set `file_offset`
+    /// to `tail_upper_bound` so that `read_next_entry_ref` won't short-circuit
+    /// before reaching it.  This is safe because `open` is single-threaded;
+    /// the real value is overwritten by the caller immediately after.
+    fn detect_used_bytes(&self, tail_upper_bound: u64) -> Result<u64> {
+        if tail_upper_bound == 0 {
+            return Ok(0);
+        }
+
+        self.file_offset.store(tail_upper_bound, Ordering::Release);
+
+        let mut offset = 0u64;
+        let mut read_buf = Vec::new();
+        let mut buf_file_offset = 0u64;
+        let mut last_durable_offset = 0u64;
+        while let Some((_, _, next_offset)) =
+            self.read_next_entry_ref(offset, &mut read_buf, &mut buf_file_offset)?
+        {
+            offset = next_offset;
+            last_durable_offset = next_offset.next_multiple_of(FILE_OFFSET_ALIGNMENT);
+        }
+
+        Ok(last_durable_offset)
     }
 
     fn parse_data_entry(buf: &[u8], offset: u64) -> Result<ParsedDataEntry> {
@@ -296,23 +396,30 @@ impl DataFile {
                 "invalid data file header",
             )));
         }
-        let mut file_offset = file
+        let physical_data_len = file
             .metadata()
             .map_err(Error::IOError)?
             .len()
             .saturating_sub(size_of::<DataFileHeader>() as u64);
-        file_offset -= file_offset % FILE_OFFSET_ALIGNMENT;
-        file.set_len(size_of::<DataFileHeader>() as u64 + file_offset)
-            .map_err(Error::IOError)?;
+        let preallocated = physical_data_len == config.max_data_file_size as u64;
+        let recovery_tail_upper_bound = Self::used_data_upper_bound(&file, physical_data_len)?;
 
-        Ok(Self {
+        let inst = Self {
             file,
-            file_offset: AtomicU64::new(file_offset),
+            file_offset: AtomicU64::new(physical_data_len),
+            last_synced_offset: AtomicU64::new(0),
             sealed_for_rotation: AtomicBool::new(false),
             config,
             file_idx,
             file_ordinal: header.ordinal,
-        })
+            preallocated,
+            recovery_tail_upper_bound,
+        };
+        let used_bytes = inst.detect_used_bytes(recovery_tail_upper_bound)?;
+        inst.file_offset.store(used_bytes, Ordering::Release);
+        inst.last_synced_offset.store(used_bytes, Ordering::Release);
+
+        Ok(inst)
     }
 
     pub(crate) fn create(
@@ -328,7 +435,7 @@ impl DataFile {
             .write(true)
             .open(data_file_path(base_path, file_idx))
             .map_err(Error::IOError)?;
-        file.set_len(size_of::<DataFileHeader>() as u64)
+        file.set_len(size_of::<DataFileHeader>() as u64 + config.max_data_file_size as u64)
             .map_err(Error::IOError)?;
         let header = DataFileHeader {
             magic: *DATA_FILE_SIGNATURE,
@@ -343,10 +450,13 @@ impl DataFile {
         Ok(Self {
             file,
             file_offset: AtomicU64::new(0),
+            last_synced_offset: AtomicU64::new(0),
             sealed_for_rotation: AtomicBool::new(false),
             config,
             file_idx,
             file_ordinal: ordinal,
+            preallocated: true,
+            recovery_tail_upper_bound: 0,
         })
     }
 
@@ -554,9 +664,16 @@ impl DataFile {
         read_buf: &'a mut Vec<u8>,
         buf_file_offset: &mut u64,
     ) -> Result<Option<(KVRef<'a>, u64, u64)>> {
+        let used_bytes = self.used_bytes();
+        if offset >= used_bytes {
+            return Ok(None);
+        }
         offset = offset.next_multiple_of(FILE_OFFSET_ALIGNMENT);
 
         loop {
+            if offset >= used_bytes {
+                return Ok(None);
+            }
             let buf_start = if offset >= *buf_file_offset {
                 (offset - *buf_file_offset) as usize
             } else {

src/internal.rs

@@ -70,6 +70,50 @@ pub(crate) fn sync_dir(_path: &Path) -> Result<()> {
     Ok(())
 }
 
+#[cfg(target_os = "linux")]
+pub(crate) fn sync_file_range(file: &File, offset: u64, len: u64) -> Result<()> {
+    use std::os::fd::AsRawFd;
+
+    if len == 0 {
+        return Ok(());
+    }
+
+    let sync_offset = i64::try_from(offset)
+        .map_err(|_| Error::IOError(std::io::Error::other("sync offset overflow")))?;
+    let sync_len = i64::try_from(len)
+        .map_err(|_| Error::IOError(std::io::Error::other("sync length overflow")))?;
+
+    let rc = unsafe {
+        libc::sync_file_range(
+            file.as_raw_fd(),
+            sync_offset,
+            sync_len,
+            libc::SYNC_FILE_RANGE_WAIT_BEFORE
+                | libc::SYNC_FILE_RANGE_WRITE
+                | libc::SYNC_FILE_RANGE_WAIT_AFTER,
+        )
+    };
+    if rc == 0 {
+        return Ok(());
+    }
+
+    let err = std::io::Error::last_os_error();
+    match err.raw_os_error() {
+        Some(libc::EINVAL | libc::ENOSYS | libc::EOPNOTSUPP) => {
+            file.sync_all().map_err(Error::IOError)
+        }
+        _ => Err(Error::IOError(err)),
+    }
+}
+
+#[cfg(not(target_os = "linux"))]
+pub(crate) fn sync_file_range(file: &File, _offset: u64, len: u64) -> Result<()> {
+    if len == 0 {
+        return Ok(());
+    }
+    file.sync_all().map_err(Error::IOError)
+}
+
 pub(crate) fn parse_data_file_idx(path: &Path) -> Option<u16> {
     let name = path.file_name()?.to_str()?;
     let suffix = name.strip_prefix("data_")?;

src/store.rs

@@ -80,6 +80,7 @@ struct CheckpointSnapshot {
     checkpoint_offset: u64,
     checkpointed_delta: i64,
     last_commit_ordinal: u64,
+    last_commit_offset: u64,
 }
 
 #[derive(Default)]
@@ -465,20 +466,23 @@ impl StoreInner {
             .ok_or(Error::MissingDataFile(active_idx))?;
         let (checkpoint_ordinal, checkpoint_offset, checkpointed_delta) =
             self.inflight_tracker.checkpoint_progress(&active_file);
-        let last_commit_ordinal = self.index_file.checkpoint_cursor().0;
+        let (last_commit_ordinal, last_commit_offset) = self.index_file.checkpoint_cursor();
         Ok(CheckpointSnapshot {
             checkpoint_ordinal,
             checkpoint_offset,
             checkpointed_delta,
             last_commit_ordinal,
+            last_commit_offset,
         })
     }
 
     fn sync_checkpoint(&self, snap: CheckpointSnapshot) -> Result<()> {
         let files = self.data_files.read();
         for data_file in files.values() {
-            if data_file.file_ordinal >= snap.last_commit_ordinal {
-                data_file.file.sync_all().map_err(Error::IOError)?;
+            if data_file.file_ordinal > snap.last_commit_ordinal {
+                data_file.sync_to_current()?;
+            } else if data_file.file_ordinal == snap.last_commit_ordinal {
+                data_file.sync_data(snap.last_commit_offset, data_file.used_bytes())?;
             }
         }
         drop(files);
@@ -1247,7 +1251,7 @@ impl CandyStore {
         self.inner.index_file.sync_all()?;
         let files = self.inner.data_files.read();
         for data_file in files.values() {
-            data_file.file.sync_all().map_err(Error::IOError)?;
+            data_file.sync_to_current()?;
         }
         sync_dir(&self.inner.base_path)
     }

src/store/checkpoint.rs

@@ -230,10 +230,12 @@ impl StoreInner {
     ) {
         let snapshot_for_follow_up = snapshot.as_ref().ok().copied();
         let result = snapshot.and_then(|snap| self.sync_checkpoint(snap));
+        let mut should_signal_compaction = false;
 
         let mut state = self.checkpoint_state.lock();
         match result {
             Ok(()) => {
+                should_signal_compaction = true;
                 state.last_checkpoint_dur_ms =
                     u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
                 if let Some(target_epoch) = target_epoch {
@@ -265,6 +267,11 @@ impl StoreInner {
             }
         }
         self.checkpoint_condvar.notify_all();
+        drop(state);
+
+        if should_signal_compaction {
+            self.signal_compaction_scan();
+        }
     }
 
     fn run_checkpoint_worker(self: &Arc<Self>) {

src/store/compaction.rs

@@ -238,7 +238,7 @@ impl StoreInner {
         if !removed.is_empty() {
             let active_idx = self.active_file_idx.load(Ordering::Acquire);
             if let Some(active_file) = self.data_files.read().get(&active_idx).cloned() {
-                let _ = active_file.file.sync_all();
+                let _ = active_file.sync_to_current();
             }
             let _ = self.index_file.sync_all();
         }

src/store/recovery.rs

@@ -93,8 +93,8 @@ impl CandyStore {
         // extent. This handles the case where the data file was truncated
         // (e.g. disk-full or corruption) and ensures the replay loop won't
         // encounter stale pointers when comparing existing entries.
-        let pre_rebuild_used_bytes = data_file.used_bytes();
-        let pre_purge_extent = pre_rebuild_used_bytes.next_multiple_of(FILE_OFFSET_ALIGNMENT);
+        let pre_rebuild_tail_upper_bound = data_file.recovery_tail_upper_bound();
+        let pre_purge_extent = pre_rebuild_tail_upper_bound.next_multiple_of(FILE_OFFSET_ALIGNMENT);
         self.apply_recovery_delta(
             self.purge_uncommitted_file_entries(data_file.file_idx, pre_purge_extent)?,
             pending_committed_delta,
@@ -151,11 +151,11 @@ impl CandyStore {
 
         let durable_extent = last_durable_offset.next_multiple_of(FILE_OFFSET_ALIGNMENT);
 
-        if durable_extent < pre_rebuild_used_bytes {
+        if durable_extent < pre_rebuild_tail_upper_bound {
             self.inner
                 .stats
                 .num_rebuild_purged_bytes
-                .fetch_add(pre_rebuild_used_bytes - durable_extent, Ordering::Relaxed);
+                .fetch_add(pre_rebuild_tail_upper_bound - durable_extent, Ordering::Relaxed);
             data_file.truncate_to_offset(durable_extent)?;
         }