You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/09/04 05:07:54 UTC
(opendal) branch main updated: fix(services/monoiofs): handle async cancel during file open (#5094)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 1237d408a9 fix(services/monoiofs): handle async cancel during file open (#5094)
1237d408a9 is described below
commit 1237d408a92ea778c7ec459bbac843581f92423e
Author: NKID00 <th...@nkid00.name>
AuthorDate: Wed Sep 4 13:07:48 2024 +0800
fix(services/monoiofs): handle async cancel during file open (#5094)
fix: handle async cancel during file open
---
core/src/services/monoiofs/core.rs | 4 ++--
core/src/services/monoiofs/reader.rs | 17 +++++++++--------
core/src/services/monoiofs/writer.rs | 13 +++++++------
3 files changed, 18 insertions(+), 16 deletions(-)
diff --git a/core/src/services/monoiofs/core.rs b/core/src/services/monoiofs/core.rs
index 30d14adc98..0a3d18e8a2 100644
--- a/core/src/services/monoiofs/core.rs
+++ b/core/src/services/monoiofs/core.rs
@@ -145,7 +145,7 @@ 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
@@ -153,7 +153,7 @@ impl MonoiofsCore {
.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
index 6e4e307906..d86df4d575 100644
--- 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
index 6af3d485cf..eb85ebc395 100644
--- 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;
}
};