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/06/04 02:26:53 UTC
(opendal) branch main updated: chore(core): Fix unit tests (#4684)
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 759ffbafbc chore(core): Fix unit tests (#4684)
759ffbafbc is described below
commit 759ffbafbc2cbd4f22b4bb1fc58d40294cf6bc52
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Jun 4 10:26:47 2024 +0800
chore(core): Fix unit tests (#4684)
* chore(core): Fix unit tests
Signed-off-by: Xuanwo <gi...@xuanwo.io>
* Stop all tasks while not recover error happened
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---------
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/futures_util.rs | 30 ++++++++++++++++++++++++++++++
core/src/raw/oio/write/multipart_write.rs | 4 +++-
2 files changed, 33 insertions(+), 1 deletion(-)
diff --git a/core/src/raw/futures_util.rs b/core/src/raw/futures_util.rs
index e9324ed602..31ef65be77 100644
--- a/core/src/raw/futures_util.rs
+++ b/core/src/raw/futures_util.rs
@@ -88,6 +88,12 @@ pub struct ConcurrentTasks<I, O> {
tasks: VecDeque<Task<(I, Result<O>)>>,
/// `results` stores the successful results.
results: VecDeque<O>,
+
+ /// hitting the last unrecoverable error.
+ ///
+ /// If concurrent tasks hit an unrecoverable error, it will stop executing new tasks and return
+ /// an unrecoverable error to users.
+ errored: bool,
}
impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
@@ -105,6 +111,7 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
tasks: VecDeque::with_capacity(concurrent),
results: VecDeque::with_capacity(concurrent),
+ errored: false,
}
}
@@ -140,6 +147,13 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
/// - Execute the task in the background if there are available slots.
/// - Await the first task in the queue if there is no available slots.
pub async fn execute(&mut self, input: I) -> Result<()> {
+ if self.errored {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "concurrent tasks met an unrecoverable error",
+ ));
+ }
+
// Short path for non-concurrent case.
if !self.is_concurrent() {
let (_, o) = (self.factory)(input).await;
@@ -164,6 +178,9 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
if err.is_temporary() {
self.tasks
.push_front(self.executor.execute((self.factory)(i)));
+ } else {
+ self.clear();
+ self.errored = true;
}
return Err(err);
}
@@ -197,6 +214,9 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
if err.is_temporary() {
self.tasks
.push_front(self.executor.execute((self.factory)(i)));
+ } else {
+ self.clear();
+ self.errored = true;
}
return Err(err);
}
@@ -206,6 +226,13 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
/// Fetch the successful result from the result queue.
pub async fn next(&mut self) -> Option<Result<O>> {
+ if self.errored {
+ return Some(Err(Error::new(
+ ErrorKind::Unexpected,
+ "concurrent tasks met an unrecoverable error",
+ )));
+ }
+
if let Some(result) = self.results.pop_front() {
return Some(Ok(result));
}
@@ -219,6 +246,9 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
if err.is_temporary() {
self.tasks
.push_front(self.executor.execute((self.factory)(i)));
+ } else {
+ self.clear();
+ self.errored = true;
}
Some(Err(err))
}
diff --git a/core/src/raw/oio/write/multipart_write.rs b/core/src/raw/oio/write/multipart_write.rs
index 77179a20ce..08b9598607 100644
--- a/core/src/raw/oio/write/multipart_write.rs
+++ b/core/src/raw/oio/write/multipart_write.rs
@@ -326,7 +326,9 @@ mod tests {
// We will have 10% percent rate for write part to fail.
if thread_rng().gen_bool(1.0 / 10.0) {
- return Err(Error::new(ErrorKind::Unexpected, "I'm a crazy monkey!"));
+ return Err(
+ Error::new(ErrorKind::Unexpected, "I'm a crazy monkey!").set_temporary()
+ );
}
{