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/03 15:07:26 UTC
(opendal) 01/02: Push task while meeting retryable error
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch retry-layer-polish
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit d75f286fd58416c7a946238e2d7cb61d1e9a8e60
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Jun 3 19:13:09 2024 +0800
Push task while meeting retryable error
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/futures_util.rs | 29 +++++++++++++++++++----------
1 file changed, 19 insertions(+), 10 deletions(-)
diff --git a/core/src/raw/futures_util.rs b/core/src/raw/futures_util.rs
index 1ebca8ab76..e9324ed602 100644
--- a/core/src/raw/futures_util.rs
+++ b/core/src/raw/futures_util.rs
@@ -160,8 +160,11 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
match o {
Ok(o) => self.results.push_back(o),
Err(err) => {
- self.tasks
- .push_front(self.executor.execute((self.factory)(i)));
+ // Retry this task if the error is temporary
+ if err.is_temporary() {
+ self.tasks
+ .push_front(self.executor.execute((self.factory)(i)));
+ }
return Err(err);
}
}
@@ -190,8 +193,11 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
continue;
}
Err(err) => {
- self.tasks
- .push_front(self.executor.execute((self.factory)(i)));
+ // Retry this task if the error is temporary
+ if err.is_temporary() {
+ self.tasks
+ .push_front(self.executor.execute((self.factory)(i)));
+ }
return Err(err);
}
}
@@ -206,14 +212,17 @@ impl<I: Send + 'static, O: Send + 'static> ConcurrentTasks<I, O> {
if let Some(task) = self.tasks.pop_front() {
let (i, o) = task.await;
- match o {
- Ok(o) => return Some(Ok(o)),
+ return match o {
+ Ok(o) => Some(Ok(o)),
Err(err) => {
- self.tasks
- .push_front(self.executor.execute((self.factory)(i)));
- return Some(Err(err));
+ // Retry this task if the error is temporary
+ if err.is_temporary() {
+ self.tasks
+ .push_front(self.executor.execute((self.factory)(i)));
+ }
+ Some(Err(err))
}
- }
+ };
}
None