Flush trigger

storage/Cargo.toml

@@ -8,9 +8,11 @@ description = "LSMTree based storage engine"
 [dependencies]
 anyhow = "1.0.99"
 bytes = "1.10.1"
+crossbeam = "0.8.4"
 crossbeam-skiplist = "0.1.3"
 moka = { version = "0.12.11", features = ["sync"] }
 ouroboros = "0.18.5"
+serde = "1.0.228"
 
 [dev-dependencies]
 tempfile = "3.23.0"

storage/src/block/builder.rs

@@ -5,9 +5,9 @@ use crate::block::{Block, SIZEOF_U16};
 #[derive(Debug)]
 pub struct BlockBuilder {
     pub(crate) data: Vec<u8>,
+    pub(crate) first_key: Vec<u8>,
     offsets: Vec<u16>,
     block_size: usize,
-    pub(crate) first_key: Vec<u8>,
 }
 
 impl BlockBuilder {

storage/src/compaction/flush.rs

@@ -0,0 +1,70 @@
+use anyhow::Result;
+use std::{
+    sync::Arc,
+    thread::{self, JoinHandle},
+    time::Duration,
+};
+
+use crate::{SSTableBuilder, Storage, lsm_storage::StorageState};
+
+impl Storage {
+    pub(crate) fn flush_frozen_memtable(&self) -> Result<()> {
+        let mut sst_builder = SSTableBuilder::new(self.config.block_size);
+        let mut guard = self.state.write().unwrap();
+
+        let mut memtables = guard.frozen_memtables.clone();
+        let mut l0_sstables = guard.l0_sstables.clone();
+        let mut sstables = guard.sstables.clone();
+
+        let Some(memtable) = memtables.pop() else {
+            return Ok(());
+        };
+        memtable.flush(&mut sst_builder)?;
+
+        let sst = sst_builder.build(
+            memtable.id,
+            self.block_cache.clone(),
+            self.sst_path(memtable.id),
+        )?;
+        l0_sstables.push(memtable.id);
+        sstables.insert(memtable.id, Arc::new(sst));
+
+        *guard = Arc::new(StorageState {
+            memtable: guard.memtable.clone(),
+            frozen_memtables: memtables,
+            l0_sstables,
+            sstables,
+        });
+
+        Ok(())
+    }
+
+    fn trigger_flush(&self) -> Result<()> {
+        let memtable_count = {
+            let guard = self.state.read().unwrap();
+            guard.frozen_memtables.len()
+        };
+
+        if self.config.num_memtable_limit > memtable_count {
+            self.flush_frozen_memtable()?;
+        }
+
+        Ok(())
+    }
+
+    fn sst_path(&self, id: usize) -> String {
+        format!("{}/sst/{}.sst", self.config.db_dir, id)
+    }
+}
+
+// TODO: suppot msg passing
+pub fn spawn_flusher(storage: Arc<Storage>) -> JoinHandle<()> {
+    let this = storage.clone();
+
+    thread::spawn(move || {
+        loop {
+            this.trigger_flush().expect("memtable to have been flushed");
+            thread::sleep(Duration::from_millis(50));
+        }
+    })
+}

storage/src/compaction/mod.rs

@@ -0,0 +1 @@
+pub mod flush;

storage/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod common;
+pub mod compaction;
 pub mod iterators;
 pub mod lsm_storage;
 mod lsm_util;