Add rocksdb as new kernel

Cargo.toml

@@ -61,6 +61,7 @@ tonic = "0.10.2"
 prost = "0.12"
 # 其他数据库内核
 sled = "0.34.7"
+rocksdb = "0.21.0"
 
 [dev-dependencies]
 assert_cmd = "0.11.0"

src/bench/kernel_bench.rs

@@ -6,6 +6,7 @@ use std::sync::atomic::AtomicU32;
 use std::sync::atomic::Ordering::Relaxed;
 
 use kip_db::kernel::lsm::storage::KipStorage;
+use kip_db::kernel::rocksdb_storage::RocksdbStorage;
 use kip_db::kernel::sled_storage::SledStorage;
 use kip_db::kernel::Storage;
 
@@ -109,7 +110,7 @@ fn monotonic_crud<T: Storage>(c: &mut Criterion) {
  let count = AtomicU32::new(0_u32);
  b.iter(|| async {
  db.set(
- Bytes::from(count.fetch_add(1, Relaxed).to_be_bytes()),
+ Bytes::from(count.fetch_add(1, Relaxed).to_be_bytes().to_vec()),
  Bytes::new(),
  )
  .await
@@ -150,9 +151,12 @@ fn random_crud<T: Storage>(c: &mut Criterion) {
 
  c.bench_function(&format!("Store: {}, random inserts", T::name()), |b| {
  b.iter(|| async {
- db.set(Bytes::from(random(SIZE).to_be_bytes()), Bytes::new())
- .await
- .unwrap();
+ db.set(
+ Bytes::from(random(SIZE).to_be_bytes().to_vec()),
+ Bytes::new(),
+ )
+ .await
+ .unwrap();
  })
  });
 
@@ -191,21 +195,25 @@ fn empty_opens<T: Storage>(c: &mut Criterion) {
 fn kv_bulk_load(c: &mut Criterion) {
  bulk_load::<KipStorage>(c);
  bulk_load::<SledStorage>(c);
+ bulk_load::<RocksdbStorage>(c);
 }
 
 fn kv_monotonic_crud(c: &mut Criterion) {
  monotonic_crud::<KipStorage>(c);
  monotonic_crud::<SledStorage>(c);
+ monotonic_crud::<RocksdbStorage>(c);
 }
 
 fn kv_random_crud(c: &mut Criterion) {
  random_crud::<KipStorage>(c);
  random_crud::<SledStorage>(c);
+ random_crud::<RocksdbStorage>(c);
 }
 
 fn kv_empty_opens(c: &mut Criterion) {
  empty_opens::<KipStorage>(c);
  empty_opens::<SledStorage>(c);
+ empty_opens::<RocksdbStorage>(c);
 }
 
 criterion_group!(

src/error.rs

@@ -45,6 +45,9 @@ pub enum KernelError {
  #[error(transparent)]
  SledErr(#[from] sled::Error),
 
+ #[error(transparent)]
+ RocksdbErr(#[from] rocksdb::Error),
+
  #[error("Cache size overflow")]
  CacheSizeOverFlow,
 

src/kernel/lsm/compactor.rs

@@ -26,11 +26,11 @@ pub(crate) type MergeShardingVec = Vec<(i64, Vec<KeyValue>)>;
 pub(crate) type DelNode = (Vec<i64>, TableMeta);
 /// Major压缩时的待删除Gen封装(N为此次Major所压缩的Level)，第一个为Level N级，第二个为Level N+1级
 pub(crate) type DelNodeTuple = (DelNode, DelNode);
-pub(crate) type SeekScope = (Scope, usize);
+pub type SeekScope = (Scope, usize);
 
 /// Store与Compactor的交互信息
 #[derive(Debug)]
-pub(crate) enum CompactTask {
+pub enum CompactTask {
  Seek(SeekScope),
  Flush(Option<oneshot::Sender<()>>),
 }

src/kernel/lsm/table/scope.rs

@@ -14,7 +14,7 @@ const SEEK_COMPACTION_COUNT: u32 = 100;
 /// 用于缓存SSTable中所有数据的第一个和最后一个数据的Key
 /// 标明数据的范围以做到快速区域定位
 #[derive(Serialize, Deserialize, Debug, Clone)]
-pub(crate) struct Scope {
+pub struct Scope {
  pub(crate) start: Bytes,
  pub(crate) end: Bytes,
  gen: i64,

src/kernel/lsm/version/cleaner.rs

@@ -5,7 +5,7 @@ use tokio::sync::mpsc::UnboundedReceiver;
 use tracing::error;
 
 #[derive(Debug)]
-pub(crate) enum CleanTag {
+pub enum CleanTag {
  Clean(u64),
  Add { version: u64, gens: Vec<i64> },
 }

src/kernel/mod.rs

@@ -13,6 +13,7 @@ use crate::KernelError;
 
 pub mod io;
 pub mod lsm;
+pub mod rocksdb_storage;
 pub mod sled_storage;
 pub mod utils;
 

src/kernel/rocksdb_storage.rs

@@ -0,0 +1,72 @@
+use crate::kernel::Storage;
+use crate::KernelError;
+use async_trait::async_trait;
+use bytes::Bytes;
+use core::slice::SlicePattern;
+use std::path::PathBuf;
+
+#[derive(Debug)]
+pub struct RocksdbStorage {
+ data_base: rocksdb::DB,
+}
+
+#[async_trait]
+impl Storage for RocksdbStorage {
+ #[inline]
+ fn name() -> &'static str
+ where
+ Self: Sized,
+ {
+ "Rocksdb"
+ }
+
+ #[inline]
+ async fn open(path: impl Into<PathBuf> + Send) -> crate::kernel::KernelResult<Self> {
+ let db = rocksdb::DB::open_default(path.into())?;
+
+ Ok(RocksdbStorage { data_base: db })
+ }
+
+ #[inline]
+ async fn flush(&self) -> crate::kernel::KernelResult<()> {
+ let _ignore = self.data_base.flush()?;
+ Ok(())
+ }
+
+ #[inline]
+ async fn set(&self, key: Bytes, value: Bytes) -> crate::kernel::KernelResult<()> {
+ let _ignore = self.data_base.put(key.as_slice(), value.to_vec())?;
+ Ok(())
+ }
+
+ #[inline]
+ async fn get(&self, key: &[u8]) -> crate::kernel::KernelResult<Option<Bytes>> {
+ match self.data_base.get(key)? {
+ None => Ok(None),
+ Some(i_vec) => Ok(Some(Bytes::from(i_vec.to_vec()))),
+ }
+ }
+
+ #[inline]
+ async fn remove(&self, key: &[u8]) -> crate::kernel::KernelResult<()> {
+ match self.data_base.delete(key) {
+ Ok(_) => Ok(()),
+ Err(e) => Err(KernelError::RocksdbErr(e)),
+ }
+ }
+
+ #[inline]
+ async fn size_of_disk(&self) -> crate::kernel::KernelResult<u64> {
+ unimplemented!("Rocksdb does not support size_of_disk()")
+ }
+
+ #[inline]
+ async fn len(&self) -> crate::kernel::KernelResult<usize> {
+ unimplemented!("Rocksdb does not support len()")
+ }
+
+ #[inline]
+ async fn is_empty(&self) -> bool {
+ unimplemented!("Rocksdb does not support is_empty()")
+ }
+}