refactor(storage): extract local worker storage adapter (#1243)

Cargo.toml

@@ -27,6 +27,7 @@ members = [
     "crates/infra/curvine-io",
     "crates/adapters/curvine-rocksdb",
     "crates/metadata/curvine-raft",
+    "crates/adapters/curvine-storage-local",
     "crates/adapters/curvine-storage-spdk",
     "crates/adapters/curvine-ufs-opendal",
     "crates/adapters/curvine-ufs-oss-hdfs",
@@ -70,6 +71,7 @@ default-members = [
     "crates/infra/curvine-sys",
     "crates/infra/curvine-io",
     "crates/metadata/curvine-raft",
+    "crates/adapters/curvine-storage-local",
     "curvine-common",
     "curvine-server",
     "crates/client/curvine-client-core",
@@ -101,6 +103,7 @@ curvine-sys = { path = "crates/infra/curvine-sys" }
 curvine-io = { path = "crates/infra/curvine-io" }
 curvine-rocksdb = { path = "crates/adapters/curvine-rocksdb" }
 curvine-raft = { path = "crates/metadata/curvine-raft" }
+curvine-storage-local = { path = "crates/adapters/curvine-storage-local" }
 curvine-storage-spdk = { path = "crates/adapters/curvine-storage-spdk" }
 curvine-ufs-opendal = { path = "crates/adapters/curvine-ufs-opendal" }
 curvine-ufs-oss-hdfs = { path = "crates/adapters/curvine-ufs-oss-hdfs" }

crates/adapters/curvine-storage-local/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "curvine-storage-local"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+description = "Curvine local worker storage adapter."
+
+[features]
+default = []
+spdk = ["dep:curvine-storage-spdk", "curvine-storage-spdk/spdk"]
+spdk-rdma = ["spdk", "curvine-storage-spdk/spdk-rdma"]
+
+[dependencies]
+orpc = { workspace = true }
+curvine-common = { workspace = true }
+curvine-rocksdb = { workspace = true }
+curvine-storage-spdk = { workspace = true, optional = true }
+bytes = { workspace = true }
+serde = { workspace = true }
+log = { workspace = true }
+indexmap = { workspace = true }
+byteorder = { workspace = true }

curvine-server/src/worker/block/block_meta.rs → crates/adapters/curvine-storage-local/src/block_meta.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::worker::storage::VfsDir;
+use crate::VfsDir;
 use curvine_common::state::{ExtendedBlock, StorageType};
 use orpc::common::FileUtils;
 use orpc::{err_box, sys, CommonResult};
@@ -51,14 +51,14 @@ impl BlockState {
 /// 3. Worker restarts the loading block, len is the file length.
 #[derive(Debug, Clone)]
 pub struct BlockMeta {
-    pub(crate) id: i64,
-    pub(crate) len: i64,
-    pub(crate) state: BlockState,
-    pub(crate) dir_id: u32,
-    pub(crate) storage_type: StorageType,
-    pub(crate) actual_len: i64,
+    pub id: i64,
+    pub len: i64,
+    pub state: BlockState,
+    pub dir_id: u32,
+    pub storage_type: StorageType,
+    pub actual_len: i64,
     /// SPDK bdev byte offset
-    pub(crate) bdev_offset: i64,
+    pub bdev_offset: i64,
 }
 
 impl BlockMeta {

curvine-server/src/worker/storage/dataset.rs → crates/adapters/curvine-storage-local/src/dataset.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::worker::block::BlockMeta;
+use crate::BlockMeta;
 use curvine_common::state::ExtendedBlock;
 use orpc::common::ByteUnit;
 use orpc::{err_box, CommonResult};

curvine-server/src/worker/storage/dir_list.rs → crates/adapters/curvine-storage-local/src/dir_list.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::worker::storage::{ChoosingPolicy, RobinChoosingPolicy, VfsDir};
+use crate::{ChoosingPolicy, RobinChoosingPolicy, VfsDir};
 use curvine_common::state::{StorageInfo, StorageType};
 use indexmap::map::Values;
 use indexmap::IndexMap;
@@ -266,7 +266,7 @@ impl Index<usize> for DirList {
 
 #[cfg(test)]
 mod tests {
-    use crate::worker::storage::{DirList, StorageRequest, VfsDir};
+    use crate::{DirList, StorageRequest, VfsDir};
     use curvine_common::conf::WorkerDataDir;
     use curvine_common::state::StorageType;
     use orpc::common::{ByteUnit, FileUtils};

curvine-server/src/worker/storage/layout/bdev_layout.rs → crates/adapters/curvine-storage-local/src/layout/bdev_layout.rs

@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::worker::block::{BlockMeta, BlockState};
-use crate::worker::storage::layout::{validate_open_offset, BlockLayout};
-use crate::worker::storage::{BlockReadContext, BlockWriteContext, SpdkMetaStore, VfsDir};
+use crate::layout::{validate_open_offset, BlockLayout};
+use crate::{BlockMeta, BlockState};
+use crate::{BlockReadContext, BlockWriteContext, SpdkMetaStore, VfsDir};
 use curvine_common::state::ExtendedBlock;
 use log::{info, warn};
 use orpc::io::IOResult;

curvine-server/src/worker/storage/layout/file_layout.rs → crates/adapters/curvine-storage-local/src/layout/file_layout.rs

@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::worker::block::{BlockMeta, BlockState};
-use crate::worker::storage::layout::{validate_open_offset, BlockLayout};
-use crate::worker::storage::{BlockReadContext, BlockWriteContext, VfsDir};
+use crate::layout::{validate_open_offset, BlockLayout};
+use crate::{BlockMeta, BlockState};
+use crate::{BlockReadContext, BlockWriteContext, VfsDir};
 use curvine_common::state::ExtendedBlock;
 #[cfg(test)]
 use orpc::common::ByteUnit;
@@ -31,7 +31,7 @@ pub struct FileLayout;
 const ACTIVE_DIR: &str = "active";
 const STAGING_DIR: &str = "staging";
 
-pub(crate) struct FileFinalizePlan {
+pub struct FileFinalizePlan {
     final_meta: BlockMeta,
     staging_path: PathBuf,
     active_path: PathBuf,
@@ -84,7 +84,7 @@ impl FileLayout {
         Ok(path)
     }
 
-    pub(crate) fn block_path(dir: &VfsDir, meta: &BlockMeta) -> CommonResult<PathBuf> {
+    pub fn block_path(dir: &VfsDir, meta: &BlockMeta) -> CommonResult<PathBuf> {
         Ok(Self::block_dir(dir, meta)?.join(meta.state().get_name(meta.id())))
     }
 

curvine-server/src/worker/storage/layout/mod.rs → crates/adapters/curvine-storage-local/src/layout/mod.rs

@@ -19,8 +19,8 @@ pub use self::bdev_layout::BdevLayout;
 pub(crate) use self::file_layout::FileFinalizePlan;
 pub use self::file_layout::FileLayout;
 
-use crate::worker::block::BlockMeta;
-use crate::worker::storage::{BlockReadContext, BlockWriteContext, SpdkMetaStore, VfsDir};
+use crate::BlockMeta;
+use crate::{BlockReadContext, BlockWriteContext, SpdkMetaStore, VfsDir};
 use curvine_common::state::{ExtendedBlock, StorageType};
 use orpc::io::IOResult;
 use orpc::{err_box, CommonResult};

curvine-server/src/worker/storage/mod.rs → crates/adapters/curvine-storage-local/src/lib.rs

@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod block_meta;
+pub use self::block_meta::{BlockMeta, BlockState};
+
 mod vfs_dir;
 pub use self::vfs_dir::*;
 

curvine-server/src/worker/storage/meta_store/mem_meta_store.rs → crates/adapters/curvine-storage-local/src/meta_store/mem_meta_store.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::worker::block::BlockMeta;
+use crate::BlockMeta;
 use std::collections::HashMap;
 
 #[derive(Default)]

curvine-server/src/worker/storage/meta_store/mod.rs → crates/adapters/curvine-storage-local/src/meta_store/mod.rs

@@ -18,7 +18,7 @@ mod vfs_meta_store;
 pub use self::mem_meta_store::MemMetaStore;
 pub use self::vfs_meta_store::VfsMetaStore;
 
-use crate::worker::block::BlockMeta;
+use crate::BlockMeta;
 use orpc::CommonResult;
 
 /// Persistent side-car store for block metadata records.

curvine-server/src/worker/storage/meta_store/vfs_meta_store.rs → crates/adapters/curvine-storage-local/src/meta_store/vfs_meta_store.rs

@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::worker::block::BlockMeta;
-use crate::worker::storage::meta_store::{BlockMetaStore, MemMetaStore};
-use crate::worker::storage::SpdkMetaStore;
+use crate::meta_store::{BlockMetaStore, MemMetaStore};
+use crate::BlockMeta;
+use crate::SpdkMetaStore;
 use curvine_common::state::StorageType;
 use log::warn;
 use std::sync::Arc;

curvine-server/src/worker/storage/policy.rs → crates/adapters/curvine-storage-local/src/policy.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::worker::storage::VfsDir;
+use crate::VfsDir;
 use curvine_common::state::StorageType;
 use indexmap::IndexMap;
 use orpc::{err_box, CommonResult};

curvine-server/src/worker/storage/spdk_meta_store.rs → crates/adapters/curvine-storage-local/src/spdk_meta_store.rs

@@ -1,6 +1,6 @@
 //! RocksDB-backed SPDK block metadata.
-use crate::worker::block::BlockMeta;
-use crate::worker::storage::meta_store::BlockMetaStore;
+use crate::meta_store::BlockMetaStore;
+use crate::BlockMeta;
 /// Key: block_id (8B). Value: dir_id(4B) | offset(8B) | size(8B) | len(8B) | finalized(1B) = 29B.
 /// O(1) per block
 use byteorder::{BigEndian, ByteOrder};