rluvaton commented on code in PR #23560:
URL: https://github.com/apache/datafusion/pull/23560#discussion_r3578852629
##########
datafusion/physical-plan/src/common.rs:
##########
@@ -196,11 +197,12 @@ pub fn spawn_buffered(
let sender = builder.tx();
builder.spawn(async move {
- while let Some(item) = input.next().await {
- if sender.send(item).await.is_err() {
- // Receiver dropped when query is shutdown early
(e.g., limit) or error,
- // no need to return propagate the send error.
- return Ok(());
+ while let Ok(permit) = sender.reserve().await {
+ // Receiver dropped when query is shutdown early (e.g.,
limit) or error,
+ // no need to return propagate the send error.
+ match input.next().await {
+ Some(item) => permit.send(item),
+ None => break,
}
Review Comment:
Please add the comment for the `reserve` here rather than in the test code.
something like
```
/// We call `reserve` (which wait to have capacity of at least 1 message in
the channel buffer) **before** polling from input
/// since if we first polled from input and then wait for the channel to
have capacity for the message we will essentially hold `buffer` (in the
channel) + 1 (waiting for the channel to have capacity)
```
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
