fix(services/monoiofs): handle async cancel during file open (#5094)

fix: handle async cancel during file open
apache · Sep 4, 2024 · 1237d40 · 1237d40
1 parent 4bc4349
commit 1237d40
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 16 deletions.
diff --git a/core/src/services/monoiofs/core.rs b/core/src/services/monoiofs/core.rs
@@ -145,15 +145,15 @@ impl MonoiofsCore {
     pub async fn spawn<F, Fut, T>(&self, f: F)
     where
         F: FnOnce() -> Fut + 'static + Send,
-        Fut: Future<Output = T>,
+        Fut: Future<Output = T> + 'static,
         T: 'static,
     {
         let result = self
             .tx
             .send_async(Box::new(move || {
                 // task will be spawned on current thread, task panic
                 // will cause current worker thread panic
-                monoio::spawn(async move { f().await });
+                monoio::spawn(f());
             }))
             .await;
         self.unwrap(result);

diff --git a/core/src/services/monoiofs/reader.rs b/core/src/services/monoiofs/reader.rs
@@ -68,15 +68,16 @@ impl MonoiofsReader {
         // worker thread
         let file = match result {
             Ok(file) => {
-                open_result_tx
-                    .send(Ok(()))
-                    .expect("send result from worker thread should success");
+                let Ok(()) = open_result_tx.send(Ok(())) else {
+                    // MonoiofsReader::new is cancelled, exit worker task
+                    return;
+                };
                 file
             }
             Err(e) => {
-                open_result_tx
-                    .send(Err(new_std_io_error(e)))
-                    .expect("send result from worker thread should success");
+                // discard the result if send failed due to MonoiofsReader::new
+                // cancelled since we are going to exit anyway
+                let _ = open_result_tx.send(Err(new_std_io_error(e)));
                 return;
             }
         };
@@ -89,8 +90,8 @@ impl MonoiofsReader {
             match req {
                 ReaderRequest::Read { pos, buf, tx } => {
                     let (result, buf) = file.read_at(buf, pos).await;
-                    // buf.len() will be set to n by monoio if read successfully,
-                    // so n is dropped
+                    // buf.len() will be set to n by monoio if read
+                    // successfully, so n is dropped
                     let result = result.map(move |_| buf).map_err(new_std_io_error);
                     // discard the result if send failed due to
                     // MonoiofsReader::read cancelled

diff --git a/core/src/services/monoiofs/writer.rs b/core/src/services/monoiofs/writer.rs
@@ -72,15 +72,16 @@ impl MonoiofsWriter {
         // worker thread
         let file = match result {
             Ok(file) => {
-                open_result_tx
-                    .send(Ok(()))
-                    .expect("send result from worker thread should success");
+                let Ok(()) = open_result_tx.send(Ok(())) else {
+                    // MonoiofsWriter::new is cancelled, exit worker task
+                    return;
+                };
                 file
             }
             Err(e) => {
-                open_result_tx
-                    .send(Err(new_std_io_error(e)))
-                    .expect("send result from worker thread should success");
+                // discard the result if send failed due to MonoiofsWriter::new
+                // cancelled since we are going to exit anyway
+                let _ = open_result_tx.send(Err(new_std_io_error(e)));
                 return;
             }
         };