meteorgan commented on code in PR #5941:
URL: https://github.com/apache/opendal/pull/5941#discussion_r2024781610
##########
core/src/raw/futures_util.rs:
##########
@@ -163,66 +183,31 @@ impl<I: Send + 'static, O: Send + 'static>
ConcurrentTasks<I, O> {
};
}
- loop {
- // Try poll once to see if there is any ready task.
- if let Some(task) = self.tasks.front_mut() {
- if let Poll::Ready((i, o)) = poll!(task) {
- match o {
- Ok(o) => {
- let _ = self.tasks.pop_front();
- self.results.push_back(o)
- }
- Err(err) => {
- // Retry this task if the error is temporary
- if err.is_temporary() {
- self.tasks
- .front_mut()
- .expect("tasks must have at least one
task")
-
.replace(self.executor.execute((self.factory)(i)));
- } else {
- self.clear();
- self.errored = true;
- }
- return Err(err);
- }
- }
- }
- }
-
- // Try to push new task if there are available space.
- if self.tasks.len() < self.tasks.capacity() {
- self.tasks
- .push_back(self.executor.execute((self.factory)(input)));
- return Ok(());
- }
-
- // Wait for the next task to be ready.
- let task = self
+ // Try to push new task if there are available space.
Review Comment:
this comment is not correct.
##########
core/src/raw/futures_util.rs:
##########
@@ -163,66 +183,31 @@ impl<I: Send + 'static, O: Send + 'static>
ConcurrentTasks<I, O> {
};
}
- loop {
- // Try poll once to see if there is any ready task.
- if let Some(task) = self.tasks.front_mut() {
- if let Poll::Ready((i, o)) = poll!(task) {
- match o {
- Ok(o) => {
- let _ = self.tasks.pop_front();
- self.results.push_back(o)
- }
- Err(err) => {
- // Retry this task if the error is temporary
- if err.is_temporary() {
- self.tasks
- .front_mut()
- .expect("tasks must have at least one
task")
-
.replace(self.executor.execute((self.factory)(i)));
- } else {
- self.clear();
- self.errored = true;
- }
- return Err(err);
- }
- }
- }
- }
-
- // Try to push new task if there are available space.
- if self.tasks.len() < self.tasks.capacity() {
- self.tasks
- .push_back(self.executor.execute((self.factory)(input)));
- return Ok(());
- }
-
- // Wait for the next task to be ready.
- let task = self
+ // Try to push new task if there are available space.
+ if !self.has_remaining() {
+ let (i, o) = self
.tasks
- .front_mut()
- .expect("tasks must have at least one task");
- let (i, o) = task.await;
+ .pop_front()
+ .expect("tasks must be available")
+ .await;
+ self.completed.fetch_sub(1, Ordering::Relaxed);
Review Comment:
I don't fully understand the usage of `completed`, why does it subtract one
here if it represents "the counter of completed tasks".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
