review localdisk

ecstore/src/disk/local.rs

@@ -17,6 +17,7 @@ use crate::{
 };
 use bytes::Bytes;
 use path_absolutize::Absolutize;
+use std::io::Read;
 use std::{
     fs::Metadata,
     path::{Path, PathBuf},
@@ -405,6 +406,65 @@ impl LocalDisk {
 
         Ok(())
     }
+
+    async fn write_all_meta(&self, volume: &str, path: &str, buf: &[u8], sync: bool) -> Result<()> {
+        let volume_dir = self.get_bucket_path(&volume)?;
+        let file_path = volume_dir.join(Path::new(&path));
+        check_path_length(&file_path.to_string_lossy().to_string())?;
+
+        let tmp_volume_dir = self.get_bucket_path(super::RUSTFS_META_TMP_BUCKET)?;
+        let tmp_file_path = tmp_volume_dir.join(Path::new(Uuid::new_v4().to_string().as_str()));
+
+        self.write_all_internal(&tmp_file_path, buf, sync, tmp_volume_dir).await?;
+
+        os::rename_all(tmp_file_path, file_path, volume_dir).await
+    }
+
+    // write_all_private
+    pub async fn write_all_private(
+        &self,
+        volume: &str,
+        path: &str,
+        buf: &[u8],
+        sync: bool,
+        skip_parent: impl AsRef<Path>,
+    ) -> Result<()> {
+        let volume_dir = self.get_bucket_path(&volume)?;
+        let file_path = volume_dir.join(Path::new(&path));
+        check_path_length(&file_path.to_string_lossy().to_string())?;
+
+        self.write_all_internal(file_path, buf, sync, skip_parent).await
+    }
+
+    pub async fn write_all_internal(
+        &self,
+        p: impl AsRef<Path>,
+        data: impl AsRef<[u8]>,
+        sync: bool,
+        base_dir: impl AsRef<Path>,
+    ) -> Result<()> {
+        if sync {
+        } else {
+        }
+        // create top dir if not exists
+        fs::create_dir_all(&p.as_ref().parent().unwrap_or_else(|| Path::new("."))).await?;
+
+        fs::write(&p, data).await?;
+        Ok(())
+    }
+
+    async fn open_file(&self, path: impl AsRef<Path>, mode: usize, skip_parent: impl AsRef<Path>) -> Result<File> {
+        let mut skip_parent = skip_parent.as_ref();
+        if skip_parent.as_os_str().is_empty() {
+            skip_parent = self.root.as_path();
+        }
+
+        if let Some(parent) = path.as_ref().parent() {
+            os::make_dir_all(parent, skip_parent).await?;
+        }
+
+        unimplemented!()
+    }
 }
 
 fn is_root_path(path: impl AsRef<Path>) -> bool {
@@ -433,14 +493,6 @@ pub async fn read_file_exists(path: impl AsRef<Path>) -> Result<(Vec<u8>, Option
     Ok((data, meta))
 }
 
-pub async fn write_all_internal(p: impl AsRef<Path>, data: impl AsRef<[u8]>) -> Result<()> {
-    // create top dir if not exists
-    fs::create_dir_all(&p.as_ref().parent().unwrap_or_else(|| Path::new("."))).await?;
-
-    fs::write(&p, data).await?;
-    Ok(())
-}
-
 pub async fn read_file_all(path: impl AsRef<Path>) -> Result<(Vec<u8>, Metadata)> {
     let p = path.as_ref();
     let meta = read_file_metadata(&path).await?;
@@ -584,18 +636,25 @@ impl DiskAPI for LocalDisk {
     }
 
     #[must_use]
-    async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
+    async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>> {
         // TOFIX:
         let p = self.get_object_path(volume, path)?;
         let (data, _) = read_file_all(&p).await?;
 
-        Ok(Bytes::from(data))
+        Ok(data)
     }
 
     async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
-        let p = self.get_object_path(volume, path)?;
+        if volume == super::RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE {
+            let mut format_info = self.format_info.lock().await;
+            format_info.data = data.clone();
+        }
 
-        write_all_internal(p, data).await?;
+        let volume_dir = self.get_bucket_path(&volume)?;
+        let file_path = volume_dir.join(Path::new(&path));
+        check_path_length(&file_path.to_string_lossy().to_string())?;
+
+        self.write_all_internal(file_path, data, true, volume_dir).await?;
 
         Ok(())
     }
@@ -978,7 +1037,8 @@ impl DiskAPI for LocalDisk {
         let fm_data = meta.marshal_msg()?;
 
         // 写入xl.meta
-        write_all_internal(&src_file_path, fm_data).await?;
+        self.write_all(src_volume, format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str(), fm_data)
+            .await?;
 
         let no_inline = src_data_path.has_root() && fi.data.is_none() && fi.size > 0;
         if no_inline {
@@ -1035,7 +1095,7 @@ impl DiskAPI for LocalDisk {
 
         if let Err(e) = utils::fs::access(&p).await {
             if os_is_not_exist(&e) {
-                os::make_dir_all(&p).await?;
+                os::make_dir_all(&p, self.root.as_path()).await?;
             }
         }
 
@@ -1102,14 +1162,15 @@ impl DiskAPI for LocalDisk {
 
         Ok(())
     }
-    async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, _opts: UpdateMetadataOpts) -> Result<()> {
-        if let Some(metadata) = fi.metadata {
+    async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: UpdateMetadataOpts) -> Result<()> {
+        if let Some(metadata) = &fi.metadata {
             let volume_dir = self.get_bucket_path(&volume)?;
             let file_path = volume_dir.join(Path::new(&path));
 
             check_path_length(&file_path.to_string_lossy().to_string())?;
 
-            self.read_all(&volume, format!("{}/{}", &path, super::STORAGE_FORMAT_FILE).as_str())
+            let buf = self
+                .read_all(&volume, format!("{}/{}", &path, super::STORAGE_FORMAT_FILE).as_str())
                 .await
                 .map_err(|e| {
                     if DiskError::FileNotFound.is(&e) && fi.version_id.is_some() {
@@ -1119,6 +1180,25 @@ impl DiskAPI for LocalDisk {
                     }
                 })?;
 
+            if !FileMeta::is_xl_format(buf.as_slice()) {
+                return Err(Error::new(DiskError::FileVersionNotFound));
+            }
+
+            let xl_meta = FileMeta::load(buf.as_slice())?;
+
+            xl_meta.update_object_version(fi)?;
+
+            let wbuf = xl_meta.marshal_msg()?;
+
+            return self
+                .write_all_meta(
+                    &volume,
+                    format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str(),
+                    &wbuf,
+                    !opts.no_persistence,
+                )
+                .await;
+
             // FIXME:
         }
 
@@ -1144,7 +1224,8 @@ impl DiskAPI for LocalDisk {
 
         let fm_data = meta.marshal_msg()?;
 
-        write_all_internal(p, fm_data).await?;
+        self.write_all(volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str(), fm_data)
+            .await?;
 
         return Ok(());
     }

ecstore/src/disk/mod.rs

@@ -126,7 +126,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>>;
     // CleanAbandonedData
     async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()>;
-    async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes>;
+    async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>>;
 }
 
 pub struct UpdateMetadataOpts {

ecstore/src/disk/os.rs

@@ -52,10 +52,18 @@ pub fn check_path_length(path_name: &str) -> Result<()> {
     Ok(())
 }
 
-pub async fn make_dir_all(path: impl AsRef<Path>) -> Result<()> {
+pub async fn make_dir_all(path: impl AsRef<Path>, base_dir: impl AsRef<Path>) -> Result<()> {
     check_path_length(path.as_ref().to_string_lossy().to_string().as_str())?;
 
-    utils::fs::make_dir_all(path.as_ref()).map_err(os_err_to_file_err).await?;
+    if let Err(e) = reliable_mkdir_all(path.as_ref(), base_dir.as_ref()).await {
+        if is_sys_err_not_dir(&e) {
+            return Err(Error::new(DiskError::FileAccessDenied));
+        } else if is_sys_err_path_not_found(&e) {
+            return Err(Error::new(DiskError::FileAccessDenied));
+        }
+
+        return Err(os_err_to_file_err(e));
+    }
 
     Ok(())
 }
@@ -181,7 +189,7 @@ pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>
 
     if let Some(parent) = dir_path.as_ref().parent() {
         // 不支持递归，直接create_dir_all了
-        if let Err(e) = fs::create_dir_all(&parent).await {
+        if let Err(e) = utils::fs::make_dir_all(&parent).await {
             if os_is_exist(&e) {
                 return Ok(());
             }

ecstore/src/disk/remote.rs

@@ -143,7 +143,7 @@ impl DiskAPI for RemoteDisk {
         Ok(())
     }
 
-    async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
+    async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>> {
         info!("read_all");
         let mut client = self.get_client_v2().await?;
         let request = Request::new(ReadAllRequest {
@@ -160,7 +160,7 @@ impl DiskAPI for RemoteDisk {
             return Err(DiskError::FileNotFound.into());
         }
 
-        Ok(Bytes::from(response.data))
+        Ok(response.data)
     }
 
     async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {

ecstore/src/store_init.rs

@@ -232,7 +232,7 @@ pub async fn load_format_erasure(disk: &DiskStore, _heal: bool) -> Result<Format
             None => e,
         })?;
 
-    let fm = FormatV3::try_from(data.as_ref())?;
+    let fm = FormatV3::try_from(data.as_slice())?;
 
     // TODO: heal
 

ecstore/src/utils/fs.rs

@@ -1,6 +1,9 @@
 use std::{fs::Metadata, path::Path};
 
-use tokio::{fs, io};
+use tokio::{
+    fs::{self, File},
+    io,
+};
 
 #[cfg(not(windows))]
 pub fn same_file(f1: &Metadata, f2: &Metadata) -> bool {
@@ -44,6 +47,51 @@ pub fn same_file(f1: &Metadata, f2: &Metadata) -> bool {
     true
 }
 
+type FileMode = usize;
+
+const O_RDONLY: FileMode = 0x00000;
+const O_WRONLY: FileMode = 0x00001;
+const O_RDWR: FileMode = 0x00002;
+const O_CREAT: FileMode = 0x00040;
+const O_EXCL: FileMode = 0x00080;
+const O_NOCTTY: FileMode = 0x00100;
+const O_TRUNC: FileMode = 0x00200;
+const O_NONBLOCK: FileMode = 0x00800;
+const O_APPEND: FileMode = 0x00400;
+const O_SYNC: FileMode = 0x01000;
+const O_ASYNC: FileMode = 0x02000;
+const O_CLOEXEC: FileMode = 0x80000;
+
+pub async fn open_file(path: impl AsRef<Path>, mode: FileMode) -> io::Result<File> {
+    let mut opts = fs::OpenOptions::new();
+
+    match mode & (O_RDONLY | O_WRONLY | O_RDWR) {
+        O_RDONLY => {
+            opts.read(true);
+        }
+        O_WRONLY => {
+            opts.write(true);
+        }
+        O_RDWR => {
+            opts.read(true);
+            opts.write(true);
+        }
+        _ => (),
+    };
+
+    if mode & O_CREAT != 0 {
+        opts.write(true);
+    }
+
+    if mode & O_APPEND != 0 {
+        opts.write(true);
+    }
+
+    // FIXME: TODO
+
+    opts.open(path.as_ref()).await
+}
+
 pub async fn access(path: impl AsRef<Path>) -> io::Result<()> {
     fs::metadata(path).await?;
     Ok(())