Support recycling log files

README.md

@@ -85,7 +85,7 @@ cargo +nightly test --test failpoints --all-features -- --test-threads 1
 
 ```
 cargo +nightly bench --all-features <bench-case-name>
-cargo run --release --package stress --help
+cargo run --release --package stress -- --help
 ```
 
 ## License

src/config.rs

@@ -78,6 +78,12 @@ pub struct Config {
     /// Default: "0.6"
     pub purge_rewrite_garbage_ratio: f64,
 
+    /// Reycle the garbage, that is, stale append / rewrite logs,
+    /// if the size of stale logs lowered than this value.
+    ///
+    /// Default: None
+    pub recycle_garbage_ratio: Option<f64>,
+
     /// Maximum memory bytes allowed for the in-memory index.
     /// Effective under the `swap` feature only.
     ///
@@ -100,6 +106,7 @@ impl Default for Config {
             purge_threshold: ReadableSize::gb(10),
             purge_rewrite_threshold: None,
             purge_rewrite_garbage_ratio: 0.6,
+            recycle_garbage_ratio: None,
             memory_limit: None,
         };
         // Test-specific configurations.
@@ -215,6 +222,7 @@ mod tests {
             soft_sanitized.target_file_size
         );
         assert_eq!(soft_sanitized.format_version, Version::default() as u64);
+        assert!(soft_sanitized.recycle_garbage_ratio.is_none());
     }
 
     #[test]

src/engine.rs

@@ -15,17 +15,18 @@ use crate::consistency::ConsistencyChecker;
 use crate::env::{DefaultFileSystem, FileSystem};
 use crate::event_listener::EventListener;
 use crate::file_pipe_log::debug::LogItemReader;
-use crate::file_pipe_log::{
-    DefaultMachineFactory, FilePipeLog, FilePipeLogBuilder, RecoveryConfig,
-};
+use crate::file_pipe_log::{DefaultMachineFactory, FilePipeLog, FilePipeLogBuilder};
 use crate::log_batch::{Command, LogBatch, MessageExt};
 use crate::memtable::{EntryIndex, MemTableRecoverContextFactory, MemTables};
 use crate::metrics::*;
-use crate::pipe_log::{FileBlockHandle, FileId, LogQueue, PipeLog};
+use crate::pipe_log::{FileBlockHandle, FileId, LogQueue, PipeLog, Signature};
 use crate::purge::{PurgeHook, PurgeManager};
 use crate::write_barrier::{WriteBarrier, Writer};
 use crate::{Error, GlobalStats, Result};
 
+#[cfg(feature = "scripting")]
+use crate::file_pipe_log::RecoveryConfig;
+
 const METRICS_FLUSH_INTERVAL: Duration = Duration::from_secs(30);
 
 pub struct Engine<F = DefaultFileSystem, P = FilePipeLog<F>>
@@ -142,6 +143,10 @@ where
         let len = log_batch.finish_populate(self.cfg.batch_compression_threshold.0 as usize)?;
         let block_handle = {
             let mut writer = Writer::new(log_batch, sync, start);
+            // Here, we introduce the `GroupCommit` by the self-implemented
+            // `Write Barrier` to improve the data syncing of multi-thread writing.
+            // If the `write_barrier.enter(...)` returned the leader of the
+            // writer group, it would responsible for dumping data into the log.
             if let Some(mut group) = self.write_barrier.enter(&mut writer) {
                 let now = Instant::now();
                 let _t = StopWatch::new_with(&ENGINE_WRITE_LEADER_DURATION_HISTOGRAM, now);
@@ -151,9 +156,12 @@ where
                             .as_secs_f64(),
                     );
                     sync |= writer.sync;
-                    let log_batch = writer.get_payload();
+                    let log_batch = writer.get_mut_payload();
                     let res = if !log_batch.is_empty() {
-                        // @TODO(lucasliang): bind `Version` to each `LogBatch`
+                        // Signs a checksum, so-called `signature`, into the LogBatch.
+                        let (format_version, file_id) =
+                            self.pipe_log.fetch_active_file(LogQueue::Append);
+                        log_batch.sign_checksum(format_version, file_id);
                         self.pipe_log
                             .append(LogQueue::Append, log_batch.encoded_bytes())
                     } else {
@@ -480,12 +488,15 @@ where
 {
     BLOCK_CACHE.with(|cache| {
         if cache.key.get() != idx.entries.unwrap() {
+            let file_id = idx.entries.unwrap().id;
+            let version = pipe_log.fetch_format_version(file_id)?;
             cache.insert(
                 idx.entries.unwrap(),
                 LogBatch::decode_entries_block(
                     &pipe_log.read_bytes(idx.entries.unwrap())?,
                     idx.entries.unwrap(),
                     idx.compression_type,
+                    (version, Signature::new(file_id)),
                 )?,
             );
         }
@@ -504,12 +515,15 @@ where
 {
     BLOCK_CACHE.with(|cache| {
         if cache.key.get() != idx.entries.unwrap() {
+            let file_id = idx.entries.unwrap().id;
+            let version = pipe_log.fetch_format_version(file_id)?;
             cache.insert(
                 idx.entries.unwrap(),
                 LogBatch::decode_entries_block(
                     &pipe_log.read_bytes(idx.entries.unwrap())?,
                     idx.entries.unwrap(),
                     idx.compression_type,
+                    (version, Signature::new(file_id)),
                 )?,
             );
         }

src/env/default.rs

@@ -74,6 +74,13 @@ impl LogFd {
         close(self.0).map_err(|e| from_nix_error(e, "close"))
     }
 
+    pub fn rename<P: AsRef<Path>>(src: P, dst: P) -> IoResult<()> {
+        fail_point!("log_fd::rename::err", |_| {
+            Err(from_nix_error(nix::Error::EINVAL, "fp"))
+        });
+        std::fs::rename(src, dst)
+    }
+
     /// Synchronizes all in-memory data of the file except metadata to the
     /// filesystem.
     pub fn sync(&self) -> IoResult<()> {
@@ -165,6 +172,33 @@ impl LogFd {
             Ok(())
         }
     }
+
+    #[allow(unused_variables)]
+    pub fn allocate_with_hole(&self, offset: usize, size: usize) -> IoResult<()> {
+        fail_point!("log_fd::falloc_with_hole::err", |_| {
+            Err(from_nix_error(nix::Error::EINVAL, "fp"))
+        });
+        #[cfg(target_os = "linux")]
+        {
+            fcntl::fallocate(
+                self.0,
+                // It'a special flag for users who want to `fallcate` a file with
+                // fill a `hole`(that is, zero) into the specified `offset + size`
+                // position in the file, which is equivalent to the manual `fallocate`
+                // and `fill zeros` into the file.
+                // Refer to `[link]https://man7.org/linux/man-pages/man2/fallocate.2.html`.
+                fcntl::FallocateFlags::FALLOC_FL_KEEP_SIZE
+                    | fcntl::FallocateFlags::FALLOC_FL_PUNCH_HOLE,
+                offset as i64,
+                size as i64,
+            )
+            .map_err(|e| from_nix_error(e, "fallocate"))
+        }
+        #[cfg(not(target_os = "linux"))]
+        {
+            Ok(())
+        }
+    }
 }
 
 impl Handle for LogFd {
@@ -210,6 +244,15 @@ impl LogFile {
     }
 }
 
+impl Clone for LogFile {
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+            offset: self.offset,
+        }
+    }
+}
+
 impl Write for LogFile {
     fn write(&mut self, buf: &[u8]) -> IoResult<usize> {
         let len = self.inner.write(self.offset, buf)?;
@@ -255,6 +298,10 @@ impl WriteExt for LogFile {
     fn allocate(&mut self, offset: usize, size: usize) -> IoResult<()> {
         self.inner.allocate(offset, size)
     }
+
+    fn allocate_with_hole(&mut self, offset: usize, size: usize) -> IoResult<()> {
+        self.inner.allocate_with_hole(offset, size)
+    }
 }
 
 pub struct DefaultFileSystem;
@@ -272,6 +319,10 @@ impl FileSystem for DefaultFileSystem {
         LogFd::open(path.as_ref())
     }
 
+    fn rename<P: AsRef<Path>>(&self, src: P, dst: P) -> IoResult<()> {
+        LogFd::rename(src, dst)
+    }
+
     fn new_reader(&self, handle: Arc<Self::Handle>) -> IoResult<Self::Reader> {
         Ok(LogFile::new(handle))
     }

src/env/mod.rs

@@ -18,6 +18,7 @@ pub trait FileSystem: Send + Sync {
 
     fn create<P: AsRef<Path>>(&self, path: P) -> Result<Self::Handle>;
     fn open<P: AsRef<Path>>(&self, path: P) -> Result<Self::Handle>;
+    fn rename<P: AsRef<Path>>(&self, src: P, dst: P) -> Result<()>;
     fn new_reader(&self, handle: Arc<Self::Handle>) -> Result<Self::Reader>;
     fn new_writer(&self, handle: Arc<Self::Handle>) -> Result<Self::Writer>;
 }
@@ -34,4 +35,5 @@ pub trait WriteExt {
     fn truncate(&mut self, offset: usize) -> Result<()>;
     fn sync(&mut self) -> Result<()>;
     fn allocate(&mut self, offset: usize, size: usize) -> Result<()>;
+    fn allocate_with_hole(&mut self, offset: usize, size: usize) -> Result<()>;
 }

src/env/obfuscated.rs

@@ -63,13 +63,16 @@ impl WriteExt for ObfuscatedWriter {
     fn allocate(&mut self, offset: usize, size: usize) -> IoResult<()> {
         self.0.allocate(offset, size)
     }
+
+    fn allocate_with_hole(&mut self, offset: usize, size: usize) -> IoResult<()> {
+        self.0.allocate_with_hole(offset, size)
+    }
 }
 
 /// `[ObfuscatedFileSystem]` is a special implementation of `[FileSystem]`,
 /// which is used for constructing and simulating an abnormal file system for
 /// `[Read]` and `[Write]`.
 pub struct ObfuscatedFileSystem(DefaultFileSystem);
-
 impl Default for ObfuscatedFileSystem {
     fn default() -> Self {
         ObfuscatedFileSystem(DefaultFileSystem)
@@ -89,6 +92,10 @@ impl FileSystem for ObfuscatedFileSystem {
         self.0.open(path)
     }
 
+    fn rename<P: AsRef<Path>>(&self, src: P, dst: P) -> IoResult<()> {
+        self.0.rename(src, dst)
+    }
+
     fn new_reader(&self, inner: Arc<Self::Handle>) -> IoResult<Self::Reader> {
         Ok(ObfuscatedReader(self.0.new_reader(inner)?))
     }

src/file_pipe_log/format.rs

@@ -84,6 +84,7 @@ pub(super) fn lock_file_path<P: AsRef<Path>>(dir: P) -> PathBuf {
 #[repr(u64)]
 pub enum Version {
     V1 = 1,
+    V2 = 2,
 }
 
 impl Version {
@@ -190,7 +191,7 @@ mod tests {
         let version2 = Version::from_u64(1).unwrap();
         assert_eq!(version, version2);
         assert!(Version::is_valid(1));
-        assert!(!Version::is_valid(2));
+        assert!(!Version::is_valid(4));
     }
 
     #[test]