JerAguilon opened a new pull request, #37839:
URL: https://github.com/apache/arrow/pull/37839
### Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in
the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your
changes and offer better suggestions for fixes.
-->
### What changes are included in this PR?
While asofjoining some large parquet datasets with many row groups, I ran
into a deadlock that I described here:
https://github.com/apache/arrow/issues/37796. Copy pasting below for
convenience:
1. The left hand side of the asofjoin completes, so `InputFinished` proceeds
as
[expected](https://github.com/apache/arrow/blob/2455bc07e09cd5341d1fabdb293afbd07682f0b2/cpp/src/arrow/acero/asof_join_node.cc#L1323).
So far so good
2. The right hand table(s) of the join are a huge dataset scan. They're
still streaming and can legally still call `AsofJoinNode::InputReceived` all
they want ([doc
ref](https://arrow.apache.org/docs/cpp/api/acero.html#_CPPv4N5arrow5acero8ExecNode13InputReceivedEP8ExecNode9ExecBatch))
3. Each input batch is blindly pushed to the `InputState`s, which in turn
defer to `BackpressureHandler`s to decide whether to pause inputs. ([code
pointer](https://github.com/apache/arrow/blob/2455bc07e09cd5341d1fabdb293afbd07682f0b2/cpp/src/arrow/acero/asof_join_node.cc#L1689))
4. If enough batches come in right after `EndFromProcessThread` is called,
then we might exceed the
[high_threshold](https://github.com/apache/arrow/blob/2455bc07e09cd5341d1fabdb293afbd07682f0b2/cpp/src/arrow/acero/asof_join_node.cc#L575)
and tell the input node to pause via the
[BackpressureController](https://github.com/apache/arrow/blob/2455bc07e09cd5341d1fabdb293afbd07682f0b2/cpp/src/arrow/acero/asof_join_node.cc#L540)
5. At this point, the process thread has stopped for the asofjoiner, so the
right hand table(s) won't be dequeue'd, meaning
`BackpressureController::Resume()` will never be called. This causes a
[deadlock](https://arrow.apache.org/docs/cpp/api/acero.html#_CPPv4N5arrow5acero19BackpressureControl5PauseEv)
TLDR this is caused by a straggling input node being paused due to
backpressure _after_ the process thread has ended. And since every `PauseInput`
needs a corresponding `ResumeInput` to be exit gracefully, we deadlock.
Turns out this is fairly easy to reproduce with small tables, if you make a
slow input node composed of 1-row record batches with a synthetic delay. My
solution is to:
1. Create a `ForceShutdown` hook that puts the input nodes in a resumed
state, and for good measure we call `StopProducing`
2. For good measure, if nodes come after the process thread exits, we short
circuit and return OK. This is because `InputReceived` can be called an
arbitrary number of times after `StopProducing`, so it makes sense to not
enqueue useless batches.
<!--
There is no need to duplicate the description in the issue here but it is
sometimes worth providing a summary of the individual changes in this PR.
-->
### Are these changes tested?
Yes, I added a delay to the batches of one of the already-existing asofjoin
backpressure tests. Checkout out `main`, we get a timeout failure. With my
changes, it passes.
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
3. Serve as another way to document the expected behavior of the code
### Are there any user-facing changes?
No
--
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]
