Add is_empty to Engine API (#228)

* add  method to engine

Signed-off-by: tabokie <xy.tao@outlook.com>

* update changelog

Signed-off-by: tabokie <xy.tao@outlook.com>

CHANGELOG.md

@@ -6,6 +6,10 @@
 
 * Unconditionally tolerate `fallocate` failures as a fix to its portability issue. Errors other than `EOPNOTSUPP` will still emit a warning.
 
+### Public API Changes
+
+* Add `is_empty` to `Engine` API.
+
 ## [0.2.0] - 2022-05-25
 
 ### Bug Fixes

src/engine.rs

@@ -296,6 +296,12 @@ where
         })
     }
 
+    /// Returns `true` if the engine contains no Raft Group. Empty Raft Group
+    /// that isn't cleaned is counted as well.
+    pub fn is_empty(&self) -> bool {
+        self.memtables.is_empty()
+    }
+
     // For testing.
     pub fn file_span(&self, queue: LogQueue) -> (u64, u64) {
         self.pipe_log.file_span(queue)
@@ -1598,4 +1604,33 @@ mod tests {
             vec.clear();
         });
     }
+
+    #[test]
+    fn test_engine_is_empty() {
+        let dir = tempfile::Builder::new()
+            .prefix("test_engine_is_empty")
+            .tempdir()
+            .unwrap();
+        let entry_data = vec![b'x'; 128];
+        let cfg = Config {
+            dir: dir.path().to_str().unwrap().to_owned(),
+            ..Default::default()
+        };
+        let fs = Arc::new(ObfuscatedFileSystem::default());
+        let rid = 1;
+
+        let engine = RaftLogEngine::open_with_file_system(cfg, fs).unwrap();
+        assert!(engine.is_empty());
+        engine.append(rid, 1, 11, Some(&entry_data));
+        assert!(!engine.is_empty());
+
+        let mut log_batch = LogBatch::default();
+        log_batch.add_command(rid, Command::Compact { index: 11 });
+        log_batch.delete(rid, b"last_index".to_vec());
+        engine.write(&mut log_batch, true).unwrap();
+        assert!(!engine.is_empty());
+
+        engine.clean(rid);
+        assert!(engine.is_empty());
+    }
 }

src/memtable.rs

@@ -912,6 +912,16 @@ impl<A: AllocatorTrait> MemTableAccessor<A> {
         ids
     }
 
+    /// Returns `true` if it does not contains any memtable.
+    pub fn is_empty(&self) -> bool {
+        for i in 0..MEMTABLE_SLOT_COUNT {
+            if !self.slots[i].read().is_empty() {
+                return false;
+            }
+        }
+        true
+    }
+
     /// Merges with a newer neighbor [`MemTableAccessor`].
     ///
     /// This method is only used for recovery.