r/w io as async

ecstore/src/disk/io.rs

@@ -1,169 +1,27 @@
 use crate::error::Result;
 use futures::TryStreamExt;
-use std::io::Cursor;
 use std::pin::Pin;
 use std::task::Poll;
-use tokio::fs::File;
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::AsyncWrite;
+use tokio::sync::oneshot;
 use tokio_util::io::ReaderStream;
 use tokio_util::io::StreamReader;
 use tracing::error;
 use tracing::warn;
 
-#[derive(Debug)]
-pub enum FileReader {
-    Local(File),
-    // Remote(RemoteFileReader),
-    Buffer(Cursor<Vec<u8>>),
-    Http(HttpFileReader),
-}
-
-impl AsyncRead for FileReader {
-    #[tracing::instrument(level = "debug", skip(self, buf))]
-    fn poll_read(
-        mut self: Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-        buf: &mut tokio::io::ReadBuf<'_>,
-    ) -> std::task::Poll<std::result::Result<(), std::io::Error>> {
-        match &mut *self {
-            Self::Local(reader) => Pin::new(reader).poll_read(cx, buf),
-            Self::Buffer(reader) => Pin::new(reader).poll_read(cx, buf),
-            Self::Http(reader) => Pin::new(reader).poll_read(cx, buf),
-        }
-    }
-}
-
-#[derive(Debug)]
-pub struct HttpFileReader {
-    // client: reqwest::Client,
-    // url: String,
-    // disk: String,
-    // volume: String,
-    // path: String,
-    // offset: usize,
-    // length: usize,
-    inner: tokio::io::DuplexStream,
-    // buf: Vec<u8>,
-    // pos: usize,
-}
-
-impl HttpFileReader {
-    pub fn new(url: &str, disk: &str, volume: &str, path: &str, offset: usize, length: usize) -> Result<Self> {
-        warn!("http read start {}", path);
-        let url = url.to_owned();
-        let disk = disk.to_owned();
-        let volume = volume.to_owned();
-        let path = path.to_owned();
-
-        // let (reader, mut writer) = tokio::io::simplex(1024);
-        let (reader, mut writer) = tokio::io::duplex(1024 * 1024 * 10);
-
-        tokio::spawn(async move {
-            let client = reqwest::Client::new();
-            let resp = match client
-                .get(format!(
-                    "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}",
-                    url,
-                    urlencoding::encode(&disk),
-                    urlencoding::encode(&volume),
-                    urlencoding::encode(&path),
-                    offset,
-                    length
-                ))
-                .send()
-                .await
-                .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
-            {
-                Ok(resp) => resp,
-                Err(err) => {
-                    warn!("http file reader error: {}", err);
-                    return;
-                }
-            };
-
-            let mut rd = StreamReader::new(
-                resp.bytes_stream()
-                    .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)),
-            );
-
-            if let Err(err) = tokio::io::copy(&mut rd, &mut writer).await {
-                error!("http file reader copy error: {}", err);
-            };
-        });
-        Ok(Self {
-            // client: reqwest::Client::new(),
-            // url: url.to_string(),
-            // disk: disk.to_string(),
-            // volume: volume.to_string(),
-            // path: path.to_string(),
-            // offset,
-            // length,
-            inner: reader,
-            // buf: Vec::new(),
-            // pos: 0,
-        })
-    }
-}
-
-impl AsyncRead for HttpFileReader {
-    #[tracing::instrument(level = "debug", skip(self, buf))]
-    fn poll_read(
-        mut self: Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-        buf: &mut tokio::io::ReadBuf<'_>,
-    ) -> std::task::Poll<std::result::Result<(), std::io::Error>> {
-        Pin::new(&mut self.inner).poll_read(cx, buf)
-    }
-}
-
-#[derive(Debug)]
-pub enum FileWriter {
-    Local(File),
-    Http(HttpFileWriter),
-    Buffer(Cursor<Vec<u8>>),
-}
-
-impl AsyncWrite for FileWriter {
-    #[tracing::instrument(level = "debug", skip(self, buf))]
-    fn poll_write(
-        mut self: Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-        buf: &[u8],
-    ) -> Poll<std::result::Result<usize, std::io::Error>> {
-        match &mut *self {
-            Self::Local(writer) => Pin::new(writer).poll_write(cx, buf),
-            Self::Buffer(writer) => Pin::new(writer).poll_write(cx, buf),
-            Self::Http(writer) => Pin::new(writer).poll_write(cx, buf),
-        }
-    }
-
-    #[tracing::instrument(level = "debug", skip(self))]
-    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<std::result::Result<(), std::io::Error>> {
-        match &mut *self {
-            Self::Local(writer) => Pin::new(writer).poll_flush(cx),
-            Self::Buffer(writer) => Pin::new(writer).poll_flush(cx),
-            Self::Http(writer) => Pin::new(writer).poll_flush(cx),
-        }
-    }
-
-    #[tracing::instrument(level = "debug", skip(self))]
-    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<std::result::Result<(), std::io::Error>> {
-        match &mut *self {
-            Self::Local(writer) => Pin::new(writer).poll_shutdown(cx),
-            Self::Buffer(writer) => Pin::new(writer).poll_shutdown(cx),
-            Self::Http(writer) => Pin::new(writer).poll_shutdown(cx),
-        }
-    }
-}
+use super::FileReader;
 
 #[derive(Debug)]
 pub struct HttpFileWriter {
     wd: tokio::io::WriteHalf<tokio::io::SimplexStream>,
+    err_rx: oneshot::Receiver<std::io::Error>,
 }
 
 impl HttpFileWriter {
     pub fn new(url: &str, disk: &str, volume: &str, path: &str, size: usize, append: bool) -> Result<Self> {
-        let (rd, wd) = tokio::io::simplex(1024 * 1024 * 10);
+        let (rd, wd) = tokio::io::simplex(4096);
+
+        let (err_tx, err_rx) = oneshot::channel::<std::io::Error>();
 
         let body = reqwest::Body::wrap_stream(ReaderStream::new(rd));
 
@@ -187,18 +45,22 @@ impl HttpFileWriter {
                 .body(body)
                 .send()
                 .await
+                .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
             {
                 error!("HttpFileWriter put file err: {:?}", err);
+
+                if let Err(er) = err_tx.send(err) {
+                    error!("HttpFileWriter tx.send err: {:?}", er);
+                }
                 // return;
             }
 
-            // TODO: handle response
-
-            // debug!("http write done {}", path);
+            // error!("http write done {}", path);
         });
 
         Ok(Self {
             wd,
+            err_rx,
             // client: reqwest::Client::new(),
             // url: url.to_string(),
             // disk: disk.to_string(),
@@ -214,6 +76,10 @@ impl AsyncWrite for HttpFileWriter {
         cx: &mut std::task::Context<'_>,
         buf: &[u8],
     ) -> Poll<std::result::Result<usize, std::io::Error>> {
+        if let Ok(err) = self.as_mut().err_rx.try_recv() {
+            return Poll::Ready(Err(err));
+        }
+
         Pin::new(&mut self.wd).poll_write(cx, buf)
     }
 
@@ -227,3 +93,29 @@ impl AsyncWrite for HttpFileWriter {
         Pin::new(&mut self.wd).poll_shutdown(cx)
     }
 }
+
+pub async fn new_http_reader(
+    url: &str,
+    disk: &str,
+    volume: &str,
+    path: &str,
+    offset: usize,
+    length: usize,
+) -> Result<FileReader> {
+    let resp = reqwest::Client::new()
+        .get(format!(
+            "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}",
+            url,
+            urlencoding::encode(disk),
+            urlencoding::encode(volume),
+            urlencoding::encode(path),
+            offset,
+            length
+        ))
+        .send()
+        .await?;
+
+    let inner = StreamReader::new(resp.bytes_stream().map_err(std::io::Error::other));
+
+    Ok(Box::new(inner))
+}

ecstore/src/disk/local.rs

@@ -745,7 +745,7 @@ impl LocalDisk {
 
         let meta = file.metadata().await?;
 
-        bitrot_verify(FileReader::Local(file), meta.size() as usize, part_size, algo, sum.to_vec(), shard_size).await
+        bitrot_verify(Box::new(file), meta.size() as usize, part_size, algo, sum.to_vec(), shard_size).await
     }
 
     async fn scan_dir<W: AsyncWrite + Unpin>(
@@ -1314,7 +1314,7 @@ impl DiskAPI for LocalDisk {
         let src_file_path = src_volume_dir.join(Path::new(src_path));
         let dst_file_path = dst_volume_dir.join(Path::new(dst_path));
 
-        warn!("rename_part src_file_path:{:?}, dst_file_path:{:?}", &src_file_path, &dst_file_path);
+        // warn!("rename_part src_file_path:{:?}, dst_file_path:{:?}", &src_file_path, &dst_file_path);
 
         check_path_length(src_file_path.to_string_lossy().as_ref())?;
         check_path_length(dst_file_path.to_string_lossy().as_ref())?;
@@ -1471,7 +1471,7 @@ impl DiskAPI for LocalDisk {
 
     #[tracing::instrument(level = "debug", skip(self))]
     async fn create_file(&self, origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result<FileWriter> {
-        warn!("disk create_file: origvolume: {}, volume: {}, path: {}", origvolume, volume, path);
+        // warn!("disk create_file: origvolume: {}, volume: {}, path: {}", origvolume, volume, path);
 
         if !origvolume.is_empty() {
             let origvolume_dir = self.get_bucket_path(origvolume)?;
@@ -1495,7 +1495,7 @@ impl DiskAPI for LocalDisk {
             .await
             .map_err(os_err_to_file_err)?;
 
-        Ok(FileWriter::Local(f))
+        Ok(Box::new(f))
 
         // Ok(())
     }
@@ -1517,7 +1517,7 @@ impl DiskAPI for LocalDisk {
 
         let f = self.open_file(file_path, O_CREATE | O_APPEND | O_WRONLY, volume_dir).await?;
 
-        Ok(FileWriter::Local(f))
+        Ok(Box::new(f))
     }
 
     // TODO: io verifier
@@ -1552,7 +1552,7 @@ impl DiskAPI for LocalDisk {
             }
         })?;
 
-        Ok(FileReader::Local(f))
+        Ok(Box::new(f))
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
@@ -1603,7 +1603,7 @@ impl DiskAPI for LocalDisk {
 
         f.seek(SeekFrom::Start(offset as u64)).await?;
 
-        Ok(FileReader::Local(f))
+        Ok(Box::new(f))
     }
     #[tracing::instrument(level = "debug", skip(self))]
     async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result<Vec<String>> {
@@ -2291,6 +2291,9 @@ impl DiskAPI for LocalDisk {
         self.scanning.fetch_add(1, Ordering::SeqCst);
         defer!(|| { self.scanning.fetch_sub(1, Ordering::SeqCst) });
 
+        // must befor metadata_sys
+        let Some(store) = new_object_layer_fn() else { return Err(Error::msg("errServerNotInitialized")) };
+
         // Check if the current bucket has replication configuration
         if let Ok((rcfg, _)) = metadata_sys::get_replication_config(&cache.info.name).await {
             if has_active_rules(&rcfg, "", true) {
@@ -2298,7 +2301,6 @@ impl DiskAPI for LocalDisk {
             }
         }
 
-        let Some(store) = new_object_layer_fn() else { return Err(Error::msg("errServerNotInitialized")) };
         let loc = self.get_disk_location();
         let disks = store.get_disks(loc.pool_idx.unwrap(), loc.disk_idx.unwrap()).await?;
         let disk = Arc::new(LocalDisk::new(&self.endpoint(), false).await?);

ecstore/src/disk/mod.rs

@@ -29,14 +29,16 @@ use crate::{
 };
 use endpoint::Endpoint;
 use error::DiskError;
-use io::{FileReader, FileWriter};
 use local::LocalDisk;
 use madmin::info_commands::DiskMetrics;
 use remote::RemoteDisk;
 use serde::{Deserialize, Serialize};
 use std::{cmp::Ordering, fmt::Debug, path::PathBuf, sync::Arc};
 use time::OffsetDateTime;
-use tokio::{io::AsyncWrite, sync::mpsc::Sender};
+use tokio::{
+    io::{AsyncRead, AsyncWrite},
+    sync::mpsc::Sender,
+};
 use tracing::info;
 use tracing::warn;
 use uuid::Uuid;
@@ -372,6 +374,9 @@ pub async fn new_disk(ep: &endpoint::Endpoint, opt: &DiskOption) -> Result<DiskS
     }
 }
 
+pub type FileReader = Box<dyn AsyncRead + Send + Sync + Unpin>;
+pub type FileWriter = Box<dyn AsyncWrite + Send + Sync + Unpin>;
+
 #[async_trait::async_trait]
 pub trait DiskAPI: Debug + Send + Sync + 'static {
     fn to_string(&self) -> String;

ecstore/src/disk/remote.rs

@@ -22,9 +22,9 @@ use tracing::info;
 use uuid::Uuid;
 
 use super::{
-    endpoint::Endpoint, io::HttpFileReader, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation,
-    DiskOption, FileInfoVersions, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
-    UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
+    endpoint::Endpoint, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption,
+    FileInfoVersions, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts,
+    VolumeInfo, WalkDirOptions,
 };
 use crate::{
     disk::error::DiskError,
@@ -36,7 +36,10 @@ use crate::{
     },
     store_api::{FileInfo, RawFileInfo},
 };
-use crate::{disk::io::HttpFileWriter, utils::proto_err_to_err};
+use crate::{
+    disk::io::{new_http_reader, HttpFileWriter},
+    utils::proto_err_to_err,
+};
 use crate::{disk::MetaCacheEntry, metacache::writer::MetacacheWriter};
 use protos::proto_gen::node_service::RenamePartRequst;
 
@@ -316,7 +319,7 @@ impl DiskAPI for RemoteDisk {
     #[tracing::instrument(level = "debug", skip(self))]
     async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result<FileWriter> {
         info!("create_file");
-        Ok(FileWriter::Http(HttpFileWriter::new(
+        Ok(Box::new(HttpFileWriter::new(
             self.endpoint.grid_host().as_str(),
             self.endpoint.to_string().as_str(),
             volume,
@@ -329,7 +332,7 @@ impl DiskAPI for RemoteDisk {
     #[tracing::instrument(level = "debug", skip(self))]
     async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
         info!("append_file");
-        Ok(FileWriter::Http(HttpFileWriter::new(
+        Ok(Box::new(HttpFileWriter::new(
             self.endpoint.grid_host().as_str(),
             self.endpoint.to_string().as_str(),
             volume,
@@ -342,26 +345,20 @@ impl DiskAPI for RemoteDisk {
     #[tracing::instrument(level = "debug", skip(self))]
     async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
         info!("read_file");
-        Ok(FileReader::Http(HttpFileReader::new(
-            self.endpoint.grid_host().as_str(),
-            self.endpoint.to_string().as_str(),
-            volume,
-            path,
-            0,
-            0,
-        )?))
+        Ok(new_http_reader(self.endpoint.grid_host().as_str(), self.endpoint.to_string().as_str(), volume, path, 0, 0).await?)
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
     async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader> {
-        Ok(FileReader::Http(HttpFileReader::new(
+        Ok(new_http_reader(
             self.endpoint.grid_host().as_str(),
             self.endpoint.to_string().as_str(),
             volume,
             path,
             offset,
             length,
-        )?))
+        )
+        .await?)
     }
 
     async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result<Vec<String>> {