fix wal close not waiting for threads, better test harness

Signed-off-by: Alex Chi <iskyzh@gmail.com>

mini-lsm-book/src/week2-06-wal.md

@@ -55,7 +55,7 @@ If WAL is enabled, you will need to recover the memtables based on WALs when loa
 cargo run --bin mini-lsm-cli -- --enable-wal
 ```
 
-Remember to recover the correct `next_sst_id` from the state, which should be `max{memtable id, sst id}` + 1. In your `close` function, you should not flush memtables to SSTs if `enable_wal` is set to true, as WAL itself provides persistency.
+Remember to recover the correct `next_sst_id` from the state, which should be `max{memtable id, sst id}` + 1. In your `close` function, you should not flush memtables to SSTs if `enable_wal` is set to true, as WAL itself provides persistency. You should wait until all compaction and flush threads to exit before closing the database.
 
 ## Test Your Understanding
 

mini-lsm-mvcc/src/lsm_storage.rs

@@ -188,12 +188,6 @@ impl MiniLsm {
         self.compaction_notifier.send(()).ok();
         self.flush_notifier.send(()).ok();
 
-        if self.inner.options.enable_wal {
-            self.inner.sync()?;
-            self.inner.sync_dir()?;
-            return Ok(());
-        }
-
         let mut compaction_thread = self.compaction_thread.lock();
         if let Some(compaction_thread) = compaction_thread.take() {
             compaction_thread
@@ -207,6 +201,12 @@ impl MiniLsm {
                 .map_err(|e| anyhow::anyhow!("{:?}", e))?;
         }
 
+        if self.inner.options.enable_wal {
+            self.inner.sync()?;
+            self.inner.sync_dir()?;
+            return Ok(());
+        }
+
         // create memtable and skip updating manifest
         if !self.inner.state.read().memtable.is_empty() {
             self.inner

mini-lsm/src/lsm_storage.rs

@@ -188,12 +188,6 @@ impl MiniLsm {
         self.compaction_notifier.send(()).ok();
         self.flush_notifier.send(()).ok();
 
-        if self.inner.options.enable_wal {
-            self.inner.sync()?;
-            self.inner.sync_dir()?;
-            return Ok(());
-        }
-
         let mut compaction_thread = self.compaction_thread.lock();
         if let Some(compaction_thread) = compaction_thread.take() {
             compaction_thread
@@ -207,6 +201,12 @@ impl MiniLsm {
                 .map_err(|e| anyhow::anyhow!("{:?}", e))?;
         }
 
+        if self.inner.options.enable_wal {
+            self.inner.sync()?;
+            self.inner.sync_dir()?;
+            return Ok(());
+        }
+
         // create memtable and skip updating manifest
         if !self.inner.state.read().memtable.is_empty() {
             self.inner

mini-lsm/src/tests/harness.rs

@@ -401,12 +401,13 @@ pub fn check_compaction_ratio(storage: Arc<MiniLsm>) {
 }
 
 pub fn dump_files_in_dir(path: impl AsRef<Path>) {
+    println!("--- DIR DUMP ---");
     for f in path.as_ref().read_dir().unwrap() {
         let f = f.unwrap();
+        print!("{}", f.path().display());
         println!(
-            "{}, size={:.3}KB",
+            ", size={:.3}KB",
-            f.path().display(),
             f.metadata().unwrap().size() as f64 / 1024.0
-        )
+        );
     }
 }