From c36c145b347c1f9e519f75e816338269e5ab3361 Mon Sep 17 00:00:00 2001 From: "shiro.lee" Date: Mon, 5 Aug 2024 22:43:45 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E8=B0=83=E6=95=B4disk=20trait=E4=BD=8D?= =?UTF-8?q?=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ecstore/src/disk/local.rs | 7 +- ecstore/src/disk/mod.rs | 176 +++++++++++++++++++++++++++++++++++++ ecstore/src/disk_api.rs | 177 -------------------------------------- ecstore/src/erasure.rs | 2 +- ecstore/src/lib.rs | 1 - ecstore/src/peer.rs | 2 +- 6 files changed, 181 insertions(+), 184 deletions(-) delete mode 100644 ecstore/src/disk_api.rs diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 4b0b77f27..78becf1a3 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -1,4 +1,7 @@ use super::{error::DiskError, format::FormatV3}; +use super::{ + DeleteOptions, DiskAPI, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, VolumeInfo, +}; use bytes::Bytes; use futures::future::join_all; use path_absolutize::Absolutize; @@ -14,10 +17,6 @@ use tracing::{debug, warn}; use uuid::Uuid; use crate::{ - disk_api::{ - DeleteOptions, DiskAPI, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, - VolumeInfo, - }, endpoint::{Endpoint, Endpoints}, error::{Error, Result}, file_meta::FileMeta, diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 0106baf58..513ee8241 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -11,3 +11,179 @@ pub const RUSTFS_META_TMP_DELETED_BUCKET: &str = ".rustfs.sys/tmp/.trash"; pub const BUCKET_META_PREFIX: &str = "buckets"; pub const FORMAT_CONFIG_FILE: &str = "format.json"; const STORAGE_FORMAT_FILE: &str = "xl.meta"; + +use crate::{ + erasure::ReadAt, + error::Result, + store_api::{FileInfo, RawFileInfo}, +}; +use bytes::Bytes; +use std::{fmt::Debug, io::SeekFrom, pin::Pin}; +use time::OffsetDateTime; +use tokio::{ + fs::File, + io::{AsyncReadExt, AsyncSeekExt, AsyncWrite}, +}; +use uuid::Uuid; + +#[async_trait::async_trait] +pub trait DiskAPI: Debug + Send + Sync + 'static { + fn is_local(&self) -> bool; + fn id(&self) -> Uuid; + + async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>; + async fn read_all(&self, volume: &str, path: &str) -> Result; + async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()>; + async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>; + async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result; + async fn append_file(&self, volume: &str, path: &str) -> Result; + async fn read_file(&self, volume: &str, path: &str) -> Result; + // 读目录下的所有文件、目录 + async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: usize) -> Result>; + // 读目录下的所有xl.meta + async fn walk_dir(&self) -> Result>; + async fn rename_data( + &self, + src_volume: &str, + src_path: &str, + file_info: FileInfo, + dst_volume: &str, + dst_path: &str, + ) -> Result; + + async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>; + async fn delete_volume(&self, volume: &str) -> Result<()>; + async fn list_volumes(&self) -> Result>; + async fn make_volume(&self, volume: &str) -> Result<()>; + async fn stat_volume(&self, volume: &str) -> Result; + + async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()>; + async fn read_version( + &self, + org_volume: &str, + volume: &str, + path: &str, + version_id: &str, + opts: &ReadOptions, + ) -> Result; + async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result; + async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; +} + +pub struct RenameDataResp { + pub old_data_dir: String, +} + +#[derive(Debug, Clone, Default)] +pub struct DeleteOptions { + pub recursive: bool, + pub immediate: bool, +} + +#[derive(Debug, Clone)] +pub struct ReadMultipleReq { + pub bucket: String, + pub prefix: String, + pub files: Vec, + pub max_size: usize, + pub metadata_only: bool, + pub abort404: bool, + pub max_results: usize, +} + +#[derive(Debug, Clone)] +pub struct ReadMultipleResp { + pub bucket: String, + pub prefix: String, + pub file: String, + pub exists: bool, + pub error: String, + pub data: Vec, + pub mod_time: OffsetDateTime, +} + +impl Default for ReadMultipleResp { + fn default() -> Self { + Self { + bucket: String::new(), + prefix: String::new(), + file: String::new(), + exists: false, + error: String::new(), + data: Vec::new(), + mod_time: OffsetDateTime::UNIX_EPOCH, + } + } +} + +pub struct VolumeInfo { + pub name: String, + pub created: OffsetDateTime, +} + +pub struct ReadOptions { + pub read_data: bool, + pub healing: bool, +} + +pub struct FileWriter { + pub inner: Pin>, +} + +impl AsyncWrite for FileWriter { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + Pin::new(&mut self.inner).poll_write(cx, buf) + } + + fn poll_flush( + mut self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown( + mut self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } +} + +impl FileWriter { + pub fn new(inner: W) -> Self + where + W: AsyncWrite + Send + Sync + 'static, + { + Self { inner: Box::pin(inner) } + } +} + +#[derive(Debug)] +pub struct FileReader { + pub inner: File, +} + +impl FileReader { + pub fn new(inner: File) -> Self { + Self { inner } + } +} + +impl ReadAt for FileReader { + async fn read_at(&mut self, offset: usize, length: usize) -> Result<(Vec, usize)> { + self.inner.seek(SeekFrom::Start(offset as u64)).await?; + + let mut buffer = vec![0; length]; + + let bytes_read = self.inner.read(&mut buffer).await?; + + buffer.truncate(bytes_read); + + Ok((buffer, bytes_read)) + } +} diff --git a/ecstore/src/disk_api.rs b/ecstore/src/disk_api.rs deleted file mode 100644 index fcc0cb1ae..000000000 --- a/ecstore/src/disk_api.rs +++ /dev/null @@ -1,177 +0,0 @@ -use std::{fmt::Debug, io::SeekFrom, pin::Pin}; - -use bytes::Bytes; -use time::OffsetDateTime; -use tokio::{ - fs::File, - io::{AsyncReadExt, AsyncSeekExt, AsyncWrite}, -}; -use uuid::Uuid; - -use crate::{ - erasure::ReadAt, - error::Result, - store_api::{FileInfo, RawFileInfo}, -}; - -#[async_trait::async_trait] -pub trait DiskAPI: Debug + Send + Sync + 'static { - fn is_local(&self) -> bool; - fn id(&self) -> Uuid; - - async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>; - async fn read_all(&self, volume: &str, path: &str) -> Result; - async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()>; - async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>; - async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result; - async fn append_file(&self, volume: &str, path: &str) -> Result; - async fn read_file(&self, volume: &str, path: &str) -> Result; - // 读目录下的所有文件、目录 - async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: usize) -> Result>; - // 读目录下的所有xl.meta - async fn walk_dir(&self) -> Result>; - async fn rename_data( - &self, - src_volume: &str, - src_path: &str, - file_info: FileInfo, - dst_volume: &str, - dst_path: &str, - ) -> Result; - - async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>; - async fn delete_volume(&self, volume: &str) -> Result<()>; - async fn list_volumes(&self) -> Result>; - async fn make_volume(&self, volume: &str) -> Result<()>; - async fn stat_volume(&self, volume: &str) -> Result; - - async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()>; - async fn read_version( - &self, - org_volume: &str, - volume: &str, - path: &str, - version_id: &str, - opts: &ReadOptions, - ) -> Result; - async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result; - async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; -} - -pub struct RenameDataResp { - pub old_data_dir: String, -} - -#[derive(Debug, Clone, Default)] -pub struct DeleteOptions { - pub recursive: bool, - pub immediate: bool, -} - -#[derive(Debug, Clone)] -pub struct ReadMultipleReq { - pub bucket: String, - pub prefix: String, - pub files: Vec, - pub max_size: usize, - pub metadata_only: bool, - pub abort404: bool, - pub max_results: usize, -} - -#[derive(Debug, Clone)] -pub struct ReadMultipleResp { - pub bucket: String, - pub prefix: String, - pub file: String, - pub exists: bool, - pub error: String, - pub data: Vec, - pub mod_time: OffsetDateTime, -} - -impl Default for ReadMultipleResp { - fn default() -> Self { - Self { - bucket: Default::default(), - prefix: Default::default(), - file: Default::default(), - exists: Default::default(), - error: Default::default(), - data: Default::default(), - mod_time: OffsetDateTime::UNIX_EPOCH, - } - } -} - -pub struct VolumeInfo { - pub name: String, - pub created: OffsetDateTime, -} - -pub struct ReadOptions { - pub read_data: bool, - pub healing: bool, -} - -pub struct FileWriter { - pub inner: Pin>, -} - -impl AsyncWrite for FileWriter { - fn poll_write( - mut self: Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: &[u8], - ) -> std::task::Poll> { - Pin::new(&mut self.inner).poll_write(cx, buf) - } - - fn poll_flush( - mut self: Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - Pin::new(&mut self.inner).poll_flush(cx) - } - - fn poll_shutdown( - mut self: Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - Pin::new(&mut self.inner).poll_shutdown(cx) - } -} - -impl FileWriter { - pub fn new(inner: W) -> Self - where - W: AsyncWrite + Send + Sync + 'static, - { - Self { inner: Box::pin(inner) } - } -} - -#[derive(Debug)] -pub struct FileReader { - pub inner: File, -} - -impl FileReader { - pub fn new(inner: File) -> Self { - Self { inner } - } -} - -impl ReadAt for FileReader { - async fn read_at(&mut self, offset: usize, length: usize) -> Result<(Vec, usize)> { - self.inner.seek(SeekFrom::Start(offset as u64)).await?; - - let mut buffer = vec![0; length]; - - let bytes_read = self.inner.read(&mut buffer).await?; - - buffer.truncate(bytes_read); - - Ok((buffer, bytes_read)) - } -} diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index 0a18e50c1..bf4ab739b 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -13,7 +13,7 @@ use uuid::Uuid; use crate::chunk_stream::ChunkedStream; use crate::disk::error::DiskError; -use crate::disk_api::FileReader; +use crate::disk::FileReader; pub struct Erasure { data_shards: usize, diff --git a/ecstore/src/lib.rs b/ecstore/src/lib.rs index 57ffa1f75..13a05a475 100644 --- a/ecstore/src/lib.rs +++ b/ecstore/src/lib.rs @@ -1,7 +1,6 @@ mod bucket_meta; mod chunk_stream; pub mod disk; -pub mod disk_api; mod disks_layout; mod endpoint; mod erasure; diff --git a/ecstore/src/peer.rs b/ecstore/src/peer.rs index c56b4128e..65439c8c6 100644 --- a/ecstore/src/peer.rs +++ b/ecstore/src/peer.rs @@ -4,8 +4,8 @@ use std::{collections::HashMap, fmt::Debug, sync::Arc}; use tracing::warn; use crate::{ + disk::VolumeInfo, disk::{error::DiskError, DiskStore}, - disk_api::VolumeInfo, endpoint::{EndpointServerPools, Node}, error::{Error, Result}, store_api::{BucketInfo, BucketOptions, MakeBucketOptions},