Optimize the cold startup of RaftEngine when enable-log-recycle == true

CHANGELOG.md

@@ -11,6 +11,10 @@
 
 * Fix data loss caused by aborted rewrite operation. Downgrading to an earlier version without the fix may produce phantom Raft Groups or keys, i.e. never written but appear in queries.
 
+### New Features
+
+* Support preparing prefilled logs to enable log recycling when start-up.
+
 ## [0.3.0] - 2022-09-14
 
 ### Bug Fixes

Makefile

@@ -38,7 +38,7 @@ clean:
 
 ## Format code in-place using rustfmt.
 format:
-	cargo fmt --all
+	cargo ${TOOLCHAIN_ARGS} fmt --all
 
 ## Run clippy.
 clippy:

src/config.rs

@@ -91,6 +91,14 @@ pub struct Config {
     ///
     /// Default: false
     pub enable_log_recycle: bool,
+
+    /// Whether to prepare log files for recycling when start.
+    /// If `true`, batch empty log files will be prepared for recycling when
+    /// starting engine.
+    /// Only available for `enable-log-reycle` is true.
+    ///
+    /// Default: false
+    pub prefill_for_recycle: bool,
 }
 
 impl Default for Config {
@@ -110,6 +118,7 @@ impl Default for Config {
             purge_rewrite_garbage_ratio: 0.6,
             memory_limit: None,
             enable_log_recycle: false,
+            prefill_for_recycle: false,
         };
         // Test-specific configurations.
         #[cfg(test)]
@@ -156,6 +165,11 @@ impl Config {
                 self.format_version
             ));
         }
+        if !self.enable_log_recycle && self.prefill_for_recycle {
+            return Err(box_err!(
+                "prefill is not allowed when log recycle is disabled"
+            ));
+        }
         #[cfg(not(feature = "swap"))]
         if self.memory_limit.is_some() {
             warn!("memory-limit will be ignored because swap feature is disabled");
@@ -207,6 +221,7 @@ mod tests {
             purge-threshold = "3MB"
             format-version = 1
             enable-log-recycle = false
+            prefill-for-recycle = false
         "#;
         let mut load: Config = toml::from_str(custom).unwrap();
         assert_eq!(load.dir, "custom_dir");
@@ -233,6 +248,7 @@ mod tests {
             target-file-size = "5000MB"
             format-version = 2
             enable-log-recycle = true
+            prefill-for-recycle = true
         "#;
         let soft_load: Config = toml::from_str(soft_error).unwrap();
         let mut soft_sanitized = soft_load;
@@ -252,6 +268,14 @@ mod tests {
         "#;
         let mut cfg_load: Config = toml::from_str(recycle_error).unwrap();
         assert!(cfg_load.sanitize().is_err());
+
+        let prefill_error = r#"
+            enable-log-recycle = false
+            prefill-for-recycle = true
+            format-version = 2
+        "#;
+        let mut cfg_load: Config = toml::from_str(prefill_error).unwrap();
+        assert!(cfg_load.sanitize().is_err());
     }
 
     #[test]

src/engine.rs

@@ -595,7 +595,7 @@ where
 mod tests {
     use super::*;
     use crate::env::ObfuscatedFileSystem;
-    use crate::file_pipe_log::FileNameExt;
+    use crate::file_pipe_log::{parse_recycled_file_name, FileNameExt};
     use crate::log_batch::AtomicGroupBuilder;
     use crate::pipe_log::Version;
     use crate::test_util::{generate_entries, PanicGuard};
@@ -1551,6 +1551,8 @@ mod tests {
                 dir: dir.path().to_str().unwrap().to_owned(),
                 target_file_size: ReadableSize(1),
                 purge_threshold: ReadableSize(1),
+                format_version: Version::V1,
+                enable_log_recycle: false,
                 ..Default::default()
             };
             // config with v2
@@ -1559,6 +1561,7 @@ mod tests {
                 target_file_size: ReadableSize(1),
                 purge_threshold: ReadableSize(1),
                 format_version: Version::V2,
+                enable_log_recycle: false,
                 ..Default::default()
             };
             test_engine_ops(&cfg_v1, &cfg_v2);
@@ -1574,6 +1577,8 @@ mod tests {
                 dir: dir.path().to_str().unwrap().to_owned(),
                 target_file_size: ReadableSize(1),
                 purge_threshold: ReadableSize(1),
+                format_version: Version::V1,
+                enable_log_recycle: false,
                 ..Default::default()
             };
             // config with v2
@@ -1583,6 +1588,7 @@ mod tests {
                 purge_threshold: ReadableSize(1),
                 format_version: Version::V2,
                 enable_log_recycle: true,
+                prefill_for_recycle: true,
                 ..Default::default()
             };
             test_engine_ops(&cfg_v1, &cfg_v2);
@@ -1948,26 +1954,38 @@ mod tests {
     pub struct DeleteMonitoredFileSystem {
         inner: ObfuscatedFileSystem,
         append_metadata: Mutex<BTreeSet<u64>>,
+        recycled_metadata: Mutex<BTreeSet<u64>>,
     }
 
     impl DeleteMonitoredFileSystem {
         fn new() -> Self {
             Self {
                 inner: ObfuscatedFileSystem::default(),
                 append_metadata: Mutex::new(BTreeSet::new()),
+                recycled_metadata: Mutex::new(BTreeSet::new()),
             }
         }
 
         fn update_metadata(&self, path: &Path, delete: bool) -> bool {
-            let id = FileId::parse_file_name(path.file_name().unwrap().to_str().unwrap()).unwrap();
-            if id.queue == LogQueue::Append {
-                if delete {
-                    self.append_metadata.lock().unwrap().remove(&id.seq)
-                } else {
-                    self.append_metadata.lock().unwrap().insert(id.seq)
+            let path = path.file_name().unwrap().to_str().unwrap();
+            let parse_append = FileId::parse_file_name(path);
+            let parse_recycled = parse_recycled_file_name(path);
+            match (parse_append, parse_recycled) {
+                (Some(id), None) if id.queue == LogQueue::Append => {
+                    if delete {
+                        self.append_metadata.lock().unwrap().remove(&id.seq)
+                    } else {
+                        self.append_metadata.lock().unwrap().insert(id.seq)
+                    }
                 }
-            } else {
-                false
+                (None, Some(seq)) => {
+                    if delete {
+                        self.recycled_metadata.lock().unwrap().remove(&seq)
+                    } else {
+                        self.recycled_metadata.lock().unwrap().insert(seq)
+                    }
+                }
+                _ => false,
             }
         }
     }
@@ -2019,12 +2037,15 @@ mod tests {
             if self.inner.exists_metadata(&path) {
                 return true;
             }
-            let id = FileId::parse_file_name(path.as_ref().file_name().unwrap().to_str().unwrap())
-                .unwrap();
-            if id.queue == LogQueue::Append {
-                self.append_metadata.lock().unwrap().contains(&id.seq)
-            } else {
-                false
+            let path = path.as_ref().file_name().unwrap().to_str().unwrap();
+            let parse_append = FileId::parse_file_name(path);
+            let parse_recycled = parse_recycled_file_name(path);
+            match (parse_append, parse_recycled) {
+                (Some(id), None) if id.queue == LogQueue::Append => {
+                    self.append_metadata.lock().unwrap().contains(&id.seq)
+                }
+                (None, Some(seq)) => self.recycled_metadata.lock().unwrap().contains(&seq),
+                _ => false,
             }
         }
 
@@ -2080,7 +2101,7 @@ mod tests {
             &start
         );
 
-        // Simulate stale metadata.
+        // Simulate recycled metadata.
         for i in start / 2..start {
             fs.append_metadata.lock().unwrap().insert(i);
         }
@@ -2105,10 +2126,12 @@ mod tests {
             purge_threshold: ReadableSize(100),
             format_version: Version::V2,
             enable_log_recycle: true,
+            prefill_for_recycle: true,
             ..Default::default()
         };
         let fs = Arc::new(DeleteMonitoredFileSystem::new());
         let engine = RaftLogEngine::open_with_file_system(cfg, fs.clone()).unwrap();
+        let recycled_start = *fs.recycled_metadata.lock().unwrap().iter().next().unwrap();
         for rid in 1..=10 {
             engine.append(rid, 1, 11, Some(&entry_data));
         }
@@ -2123,23 +2146,31 @@ mod tests {
             .must_rewrite_append_queue(Some(end - 1), None);
         assert!(start < engine.file_span(LogQueue::Append).0);
         assert_eq!(engine.file_count(Some(LogQueue::Append)), 1);
-        // no file have been physically deleted.
+        // Recycled files have been reused.
         assert_eq!(
             fs.append_metadata.lock().unwrap().iter().next().unwrap(),
-            &start
+            &(start + 20)
         );
-        // reusing these files.
+        let recycled_start_1 = *fs.recycled_metadata.lock().unwrap().iter().next().unwrap();
+        assert!(recycled_start < recycled_start_1);
+        // Reuse these files.
         for rid in 1..=5 {
             engine.append(rid, 1, 11, Some(&entry_data));
         }
         let start_1 = *fs.append_metadata.lock().unwrap().iter().next().unwrap();
-        assert!(start < start_1);
+        assert!(start <= start_1);
+        let recycled_start_2 = *fs.recycled_metadata.lock().unwrap().iter().next().unwrap();
+        assert!(recycled_start_1 < recycled_start_2);
 
-        // reopen the engine and validate the stale files are removed
+        // Reopen the engine and validate the recycled files are reserved
+        let file_count = fs.inner.file_count();
         let engine = engine.reopen();
-        assert_eq!(fs.inner.file_count(), engine.file_count(None));
+        assert_eq!(file_count, fs.inner.file_count());
+        assert!(file_count > engine.file_count(None));
         let start_2 = *fs.append_metadata.lock().unwrap().iter().next().unwrap();
-        assert!(start_1 < start_2);
+        assert_eq!(start_1, start_2);
+        let recycled_start_3 = *fs.recycled_metadata.lock().unwrap().iter().next().unwrap();
+        assert_eq!(recycled_start_2, recycled_start_3);
     }
 
     #[test]
@@ -2250,6 +2281,97 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_start_engine_with_resize_recycle_capacity() {
+        let dir = tempfile::Builder::new()
+            .prefix("test_start_engine_with_resize_recycle_capacity")
+            .tempdir()
+            .unwrap();
+        let path = dir.path().to_str().unwrap();
+        let file_system = Arc::new(DeleteMonitoredFileSystem::new());
+        let entry_data = vec![b'x'; 512];
+
+        // Case 1: start an engine with no-recycle.
+        let cfg = Config {
+            dir: path.to_owned(),
+            enable_log_recycle: false,
+            ..Default::default()
+        };
+        let engine = RaftLogEngine::open_with_file_system(cfg, file_system.clone()).unwrap();
+        let (start, _) = engine.file_span(LogQueue::Append);
+        // Only one valid file left, the last one => active_file.
+        assert_eq!(engine.file_count(Some(LogQueue::Append)), 1);
+        assert_eq!(file_system.inner.file_count(), engine.file_count(None));
+        // Append data.
+        for rid in 1..=5 {
+            engine.append(rid, 1, 10, Some(&entry_data));
+        }
+        assert_eq!(engine.file_span(LogQueue::Append).0, start);
+        assert_eq!(file_system.inner.file_count(), engine.file_count(None));
+        drop(engine);
+
+        // Case 2: restart the engine with a common size of recycling capacity.
+        let cfg = Config {
+            dir: path.to_owned(),
+            target_file_size: ReadableSize(1),
+            purge_threshold: ReadableSize(80), // common size of capacity
+            enable_log_recycle: true,
+            prefill_for_recycle: true,
+            ..Default::default()
+        };
+        let engine =
+            RaftLogEngine::open_with_file_system(cfg.clone(), file_system.clone()).unwrap();
+        let (start, end) = engine.file_span(LogQueue::Append);
+        // Only one valid file left, the last one => active_file.
+        assert_eq!(start, end);
+        let recycled_count = file_system.inner.file_count() - engine.file_count(None);
+        assert!(recycled_count > 0);
+        // Append data. Several recycled files have been reused.
+        for rid in 1..=5 {
+            engine.append(rid, 10, 20, Some(&entry_data));
+        }
+        assert_eq!(engine.file_span(LogQueue::Append).0, start);
+        assert!(recycled_count > file_system.inner.file_count() - engine.file_count(None));
+        let (start, end) = engine.file_span(LogQueue::Append);
+        let recycled_count = file_system.inner.file_count() - engine.file_count(None);
+        drop(engine);
+
+        // Case 3: restart the engine with a smaller capacity. Redundant recycled files
+        // will be cleared.
+        let cfg_v2 = Config {
+            target_file_size: ReadableSize(1),
+            purge_threshold: ReadableSize(50),
+            ..cfg
+        };
+        let engine =
+            RaftLogEngine::open_with_file_system(cfg_v2.clone(), file_system.clone()).unwrap();
+        assert_eq!(engine.file_span(LogQueue::Append), (start, end));
+        assert!(recycled_count > file_system.inner.file_count() - engine.file_count(None));
+        // Recycled files have filled the LogQueue::Append, purge_expired_files won't
+        // truely remove files from it.
+        engine.purge_expired_files().unwrap();
+        assert_eq!(engine.file_span(LogQueue::Append), (start, end));
+        for rid in 1..=10 {
+            engine.append(rid, 20, 31, Some(&entry_data));
+        }
+        assert!(engine.file_span(LogQueue::Append).1 > end);
+        let engine = engine.reopen();
+        assert!(recycled_count > file_system.inner.file_count() - engine.file_count(None));
+        drop(engine);
+
+        // Case 4: restart the engine without log recycling. Recycled logs should be
+        // cleared.
+        let cfg_v3 = Config {
+            target_file_size: ReadableSize::kb(2),
+            purge_threshold: ReadableSize::kb(100),
+            enable_log_recycle: false,
+            prefill_for_recycle: false,
+            ..cfg_v2
+        };
+        let engine = RaftLogEngine::open_with_file_system(cfg_v3, file_system.clone()).unwrap();
+        assert_eq!(file_system.inner.file_count(), engine.file_count(None));
+    }
+
     #[test]
     fn test_rewrite_atomic_group() {
         let dir = tempfile::Builder::new()

src/env/default.rs

@@ -273,6 +273,7 @@ impl FileSystem for DefaultFileSystem {
     }
 
     fn delete<P: AsRef<Path>>(&self, path: P) -> IoResult<()> {
+        fail_point!("default_fs::delete_skipped", |_| { Ok(()) });
         std::fs::remove_file(path)
     }
 

src/env/mod.rs

@@ -30,6 +30,12 @@ pub trait FileSystem: Send + Sync {
         self.rename(src_path, dst_path)
     }
 
+    #[inline]
+    fn reuse_and_open<P: AsRef<Path>>(&self, src_path: P, dst_path: P) -> Result<Self::Handle> {
+        self.reuse(src_path.as_ref(), dst_path.as_ref())?;
+        self.open(dst_path)
+    }
+
     /// Deletes user implemented metadata associated with `path`. Returns
     /// `true` if any metadata is deleted.
     ///