refactor: optimize write performance for multiple replicas (#244)

* fix: opendal is not available (#241)

* feat: improve s3 api error message

* fix: fix data loading bug

* refactor: optimize write performance for multiple replicas

* feat: load command adds watch option

Author: bigbigxu

Cargo.lock

@@ -121,3 +121,4 @@ bigdecimal = "0.4.8"
 bitvec = "1.0.1"
 config = "0.13.4"
 tempfile = "3.21.0"
+md-5 = "0.10.6"

Cargo.toml

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::cmds::LoadStatusCommand;
 use crate::util::*;
 use clap::Parser;
 use curvine_client::rpc::JobMasterClient;
@@ -24,6 +25,10 @@ pub struct LoadCommand {
 
     #[arg(long, default_value = "${CURVINE_CONF_FILE}")]
     conf: String,
+
+    /// Watch load job status after submission
+    #[arg(long, short = 'w')]
+    watch: bool,
 }
 
 impl LoadCommand {
@@ -39,6 +44,18 @@ impl LoadCommand {
         let command = LoadJobCommand::builder(&self.path).build();
         let rep = handle_rpc_result(client.submit_load_job(command)).await;
         println!("{}", rep);
+
+        if self.watch {
+            let status_command = LoadStatusCommand::new(
+                rep.job_id.clone(),
+                false,
+                "1s".to_string(),
+                self.conf.clone(),
+            );
+
+            status_command.execute(client).await?;
+        }
+
         Ok(())
     }
 }

curvine-cli/src/cmds/load.rs

@@ -34,6 +34,14 @@ pub struct LoadStatusCommand {
 }
 
 impl LoadStatusCommand {
+    pub fn new(job_id: String, verbose: bool, watch: String, conf: String) -> Self {
+        Self {
+            job_id,
+            verbose,
+            watch: Some(watch),
+            conf,
+        }
+    }
     pub async fn execute(&self, client: JobMasterClient) -> CommonResult<()> {
         println!("\n Checking status for {}", self.job_id);
 

curvine-cli/src/cmds/load_status.rs

@@ -17,6 +17,7 @@ use crate::block::{BlockWriterLocal, BlockWriterRemote};
 use crate::file::FsContext;
 use curvine_common::state::{BlockLocation, CommitBlock, LocatedBlock, WorkerAddress};
 use curvine_common::FsResult;
+use futures::future::try_join_all;
 use orpc::err_box;
 use orpc::runtime::{RpcRuntime, Runtime};
 use orpc::sys::DataSlice;
@@ -125,51 +126,93 @@ impl BlockWriter {
 
     pub async fn write(&mut self, chunk: DataSlice) -> FsResult<()> {
         let chunk = chunk.freeze();
-        for writer in &mut self.inners {
-            if let Err(e) = writer.write(chunk.clone()).await {
-                self.fs_context.add_failed_worker(writer.worker_address());
-                return Err(e);
-            }
+        let mut futures = Vec::with_capacity(self.inners.len());
+        for writer in self.inners.iter_mut() {
+            let chunk_clone = chunk.clone();
+            let task = async move {
+                writer
+                    .write(chunk_clone)
+                    .await
+                    .map_err(|e| (writer.worker_address().clone(), e))
+            };
+            futures.push(task);
+        }
+
+        if let Err((worker_addr, e)) = try_join_all(futures).await {
+            self.fs_context.add_failed_worker(&worker_addr);
+            return Err(e);
         }
         Ok(())
     }
 
-    pub fn blocking_write(&mut self, rt: &Runtime, buf: DataSlice) -> FsResult<()> {
-        for writer in &mut self.inners {
-            if let Err(e) = writer.blocking_write(rt, buf.clone()) {
-                self.fs_context.add_failed_worker(writer.worker_address());
-                return Err(e);
+    pub fn blocking_write(&mut self, rt: &Runtime, chunk: DataSlice) -> FsResult<()> {
+        if self.inners.len() == 1 {
+            if let Err(e) = self.inners[0].blocking_write(rt, chunk) {
+                self.fs_context
+                    .add_failed_worker(self.inners[0].worker_address());
+                Err(e)
+            } else {
+                Ok(())
             }
+        } else {
+            rt.block_on(self.write(chunk))?;
+            Ok(())
         }
-        Ok(())
     }
 
     pub async fn flush(&mut self) -> FsResult<()> {
-        for writer in &mut self.inners {
-            if let Err(e) = writer.flush().await {
-                self.fs_context.add_failed_worker(writer.worker_address());
-                return Err(e);
-            }
+        let mut futures = Vec::with_capacity(self.inners.len());
+        for writer in self.inners.iter_mut() {
+            let task = async move {
+                writer
+                    .flush()
+                    .await
+                    .map_err(|e| (writer.worker_address().clone(), e))
+            };
+            futures.push(task);
+        }
+
+        if let Err((worker_addr, e)) = try_join_all(futures).await {
+            self.fs_context.add_failed_worker(&worker_addr);
+            return Err(e);
         }
         Ok(())
     }
 
     pub async fn complete(&mut self) -> FsResult<CommitBlock> {
-        for writer in &mut self.inners {
-            if let Err(e) = writer.complete().await {
-                self.fs_context.add_failed_worker(writer.worker_address());
-                return Err(e);
-            }
+        let mut futures = Vec::with_capacity(self.inners.len());
+        for writer in self.inners.iter_mut() {
+            let task = async move {
+                writer
+                    .complete()
+                    .await
+                    .map_err(|e| (writer.worker_address().clone(), e))
+            };
+            futures.push(task);
+        }
+
+        if let Err((worker_addr, e)) = try_join_all(futures).await {
+            self.fs_context.add_failed_worker(&worker_addr);
+            return Err(e);
         }
         Ok(self.to_commit_block())
     }
 
     pub async fn cancel(&mut self) -> FsResult<()> {
-        for writer in &mut self.inners {
-            if let Err(e) = writer.cancel().await {
-                self.fs_context.add_failed_worker(writer.worker_address());
-                return Err(e);
-            }
+        let mut futures = Vec::with_capacity(self.inners.len());
+        for writer in self.inners.iter_mut() {
+            let task = async move {
+                writer
+                    .cancel()
+                    .await
+                    .map_err(|e| (writer.worker_address().clone(), e))
+            };
+            futures.push(task);
+        }
+
+        if let Err((worker_addr, e)) = try_join_all(futures).await {
+            self.fs_context.add_failed_worker(&worker_addr);
+            return Err(e);
         }
         Ok(())
     }

curvine-client/src/block/block_writer.rs

@@ -194,13 +194,8 @@ macro_rules! impl_filesystem_for_enum {
             ::curvine_common::fs::FileSystem<
                 $crate::unified::UnifiedWriter,
                 $crate::unified::UnifiedReader,
-                ::curvine_common::conf::UfsConf,
             > for $enum_name
         {
-            fn conf(&self) -> &::curvine_common::conf::UfsConf {
-                match_fs_variants!(self, conf)
-            }
-
             async fn mkdir(
                 &self,
                 path: &::curvine_common::fs::Path,

curvine-client/src/unified/macros.rs

@@ -16,7 +16,6 @@ use crate::file::{FsReader, FsWriter};
 use crate::impl_filesystem_for_enum;
 use crate::*;
 use crate::{impl_reader_for_enum, impl_writer_for_enum};
-use curvine_common::conf::UfsConf;
 use curvine_common::fs::Path;
 use curvine_common::state::MountInfo;
 use curvine_common::FsResult;
@@ -78,15 +77,15 @@ impl UfsFileSystem {
         match path.scheme() {
             #[cfg(feature = "s3")]
             Some(S3_SCHEME) => {
-                let fs = S3FileSystem::new(UfsConf::with_map(conf))?;
+                let fs = S3FileSystem::new(conf)?;
                 Ok(UfsFileSystem::S3(fs))
             }
 
             #[cfg(feature = "opendal")]
             Some(scheme)
                 if ["s3", "oss", "cos", "gcs", "azure", "azblob", "hdfs"].contains(&scheme) =>
             {
-                let fs = OpendalFileSystem::new(path, UfsConf::with_map(conf))?;
+                let fs = OpendalFileSystem::new(path, conf)?;
                 Ok(UfsFileSystem::OpenDAL(fs))
             }
 

curvine-client/src/unified/mod.rs

@@ -69,6 +69,10 @@ impl UnifiedFileSystem {
         Ok(fs)
     }
 
+    pub fn conf(&self) -> &ClusterConf {
+        self.cv.conf()
+    }
+
     pub fn cv(&self) -> CurvineFileSystem {
         self.cv.clone()
     }
@@ -206,11 +210,7 @@ impl UnifiedFileSystem {
     }
 }
 
-impl FileSystem<UnifiedWriter, UnifiedReader, ClusterConf> for UnifiedFileSystem {
-    fn conf(&self) -> &ClusterConf {
-        self.cv.conf()
-    }
-
+impl FileSystem<UnifiedWriter, UnifiedReader> for UnifiedFileSystem {
     async fn mkdir(&self, path: &Path, create_parent: bool) -> FsResult<bool> {
         match self.get_mount(path).await? {
             None => self.cv.mkdir(path, create_parent).await,

curvine-client/src/unified/unified_filesystem.rs

@@ -53,11 +53,11 @@ pub struct JobConf {
 }
 
 impl JobConf {
-    pub const DEFAULT_JOB_LIFE_TTL: &'static str = "6h";
+    pub const DEFAULT_JOB_LIFE_TTL: &'static str = "24h";
     pub const DEFAULT_JOB_CLEANUP_TTL_STR: &'static str = "10m";
     pub const DEFAULT_JOB_MAX_FILES: usize = 100000;
     pub const DEFAULT_TASK_TIMEOUT: &'static str = "1h";
-    pub const DEFAULT_TASK_REPORT_INTERVAL: &'static str = "5s";
+    pub const DEFAULT_TASK_REPORT_INTERVAL: &'static str = "10s";
     pub const DEFAULT_WORKER_MAX_CONCURRENT_TASKS: usize = 1000;
 
     pub fn init(&mut self) -> FsResult<()> {

curvine-common/src/conf/job_conf.rs

@@ -20,9 +20,7 @@ use crate::FsResult;
 use prost::bytes::BytesMut;
 use std::future::Future;
 
-pub trait FileSystem<Writer, Reader, Conf> {
-    fn conf(&self) -> &Conf;
-
+pub trait FileSystem<Writer, Reader> {
     fn mkdir(&self, path: &Path, create_parent: bool) -> impl Future<Output = FsResult<bool>>;
 
     fn create(&self, path: &Path, overwrite: bool) -> impl Future<Output = FsResult<Writer>>;