ARROW-?????: [C++] as-of-join backpressure for large sources #21
Conversation
|
Thanks for opening a pull request! If this is not a minor PR. Could you open an issue for this pull request on JIRA? https://issues.apache.org/jira/browse/ARROW Opening JIRAs ahead of time contributes to the Openness of the Apache Arrow project. Then could you also rename pull request title in the following format? or See also: |
|
@icexelloss, @westonpace: I've added a experimentation tool for backpressure with large sources. Here's an example invocation: Trying various parameterizations, I've observed runs with:
I intend to investigate further. |
|
This CI job failure suggests that the |
Is this an issue because of using the serial executor? I don't recall having this issue before |
No, the issue is due to a fix, the |
|
Ok. As far as I recall this is a new issue that got introduced on this branch. Perhaps it is because of the serial execution but I am not sure. Do you know why would the process thread try to join itself? As far as I remember the destructor should be called on the thread that owns the |
|
Perhaps @westonpace has some insight what might be going on |
|
Ah, we probably need to move line 952 up to line 950 in asof_join_node.cc so that the node gets mark finished on a plan thread and not on the process thread. I'll reproduce / fix that this evening (I'm in meetings until about 2PM PST so I won't be able to look at it until then). I'll do a quick review and merge then too. |
|
Aha - that makes sense
…On Thu, Oct 27, 2022 at 10:58 AM Weston Pace ***@***.***> wrote:
Ah, we probably need to move line 952 up to line 950 in asof_join_node.cc
so that the node gets mark finished on a plan thread and not on the process
thread. I'll reproduce / fix that this evening (I'm in meetings until about
2PM PST so I won't be able to look at it until then). I'll do a quick
review and merge then too.
—
Reply to this email directly, view it on GitHub
<#21 (comment)>, or
unsubscribe
<https://github.com/notifications/unsubscribe-auth/AAGBXLDJIZGWTRJ67EYBUO3WFKKBXANCNFSM6AAAAAARPN4O44>
.
You are receiving this because you were mentioned.Message ID:
***@***.***>
|
|
This sounds in the right direction, however:
|
|
Just moving MarkFinished into the plan task caused a small problem because then the process thread might run again before the plan task had a chance to launch and it might try to mark the node finished a second time. I changed the process thread to return a bool so that it returns false if it is finished (and then this exits the thread). I don't think RAII is needed anymore since, as soon as the |
|
Also, I was able to reproduce the bug by repeatedly running the new backpressure test under stress and confirmed that this change prevented this bug. |
| if (state_.at(0)->Finished()) { | ||
| ErrorIfNotOk(plan_->ScheduleTask([this] { | ||
| outputs_[0]->InputFinished(this, batches_produced_); | ||
| finished_.MarkFinished(); |
There was a problem hiding this comment.
What if the previous line throws an exception? This line potentially calls into a node not provided by Arrow, so it could throw. The call to MarkFinished should happen in any case.
There was a problem hiding this comment.
I could be mistaken but I don't think input finished is allowed to throw exception. I think there is a ticket/discussion to allow InputReceived to return a Status rather than void but not sure about InputFinished.
There was a problem hiding this comment.
I take a safety point of view here. Even if the method is not allowed to throw an exception, it is not something Arrow QA can validate, because the node called into is potentially outside of Arrow code. I'm proposing a small fix that prevents a developer oversight, due to causing an exception where one shouldn't occur, from escalating into a deadlock. A deadlock is strictly worse at least when a long-running execution is planned because the user may notice the deadlock much later than when it occurred; an early-failure is better.
There was a problem hiding this comment.
I agree an early-failure is better - an intended exception should not cause the system to deadlock ideally. @rtpsw I think I am missing sth - do you have a proposed fix?
There was a problem hiding this comment.
I'd propose fixing by wrapping finished_.MarkFinished() in a destructor for its scope - something like:
ErrorIfNotOk(plan_->ScheduleTask([this] {
Defer cleanup([]() { finished_.MarkFinished(); });
outputs_[0]->InputFinished(this, batches_produced_);
return Status::OK();
}));
where Defer is something like:
template <typename Callable>
struct Defer {
Callable callable;
Defer(Callable callable) : callable(std::move(callable)) {}
~Defer() noexcept { callable(); }
};
There was a problem hiding this comment.
Sorry for posting code here. I'll only free up to preparing a commit a bit later.
|
@westonpace, your recent commit appears to cause an as-of-join test to hang on my machine. The issue doesn't seem to be deterministic; I see a hang in a "random" |
I'm not sure if I did. I was mostly testing the backpressure test. I'll take a look. |
|
Below is a gdb session of a hanging test. I waited maybe a minute before manually interrupting. |
|
Another issue seems to be that TSAN is reporting a race condition because the backpressure concurrent queue's DoHandle is accessing UnsyncSize which doesn't use any lock. I'll work on the deadlock first and then look at the race condition later. |
|
I'll commit shortly with a possible fix. @westonpace, please review. |
|
Ah, I also have a possible fix 😆 |
|
This was mine: apache@7ef1e24 |
|
Your fix is ok but that condition will never evaluate to false so if your fix works then we should just get rid of that if check entirely. |
|
*never evaluate to true |
|
Btw, here is the race condition: |
|
@westonpace, if the as-of-join tests work with your fix, then let's go with it. I'd just suggest adding this patch to it. |
|
To be sure, is the race condition related to the hang and fixed by your hang-fixing code? |
Never mind, now I see this. |
Ah, the |
| : handler_(std::move(handler)) {} | ||
|
|
||
| T Pop() { | ||
| std::unique_lock<std::mutex> lock(ConcurrentQueue<T>::GetMutex()); |
There was a problem hiding this comment.
Is this related to the deadlock issue?
|
@rtpsw It looks like this PR is a mixture of Can you better title the PR and summarize the change so I can understand how urgent/important this issue is? I assume we want to pick this change for the MVP release as well? |
|
This PR started with refactoring and adding testing of as-of-join backpressure for large sources. During discussions, a couple of side-issues were handled:
Given the above, I think this PR is important to finish soon. If we had ample time, we could have had a separate PR for each side-issue. |
This reverts commit ddd72d9.
@rtpsw I put my fix in and added your defer suggestion. If custom code throws exceptions it will probably still break things. I don't think we are going to solve that today. But at least we can be a little bit safer :)
I ran my test repeatedly and did not see the issue. I was actually running TSAN, just with stress (an ubuntu program which stresses the CPU). To do this I run If you want to run ASAN you can (there is a CI job which runs it as well) by setting ARROW_USE_ASAN in the cmake options. For good measure I went ahead and ran the asof join tests through ASAN a few times and didn't see anything. |
…o make it more obvious what the function is doing. Fix a substrait test that was relying on an order that isn't really deterministic.
|
I went ahead and merged this since we seemed to be in a stable state. Let me know if there is anything else I need to look at. |
|
Thanks, @westonpace. I reviewed - LGTM. |
…n timezone (apache#45051) ### Rationale for this change If the timezone database is present on the system, but does not contain a timezone referenced in a ORC file, the ORC reader will crash with an uncaught C++ exception. This can happen for example on Ubuntu 24.04 where some timezone aliases have been removed from the main `tzdata` package to a `tzdata-legacy` package. If `tzdata-legacy` is not installed, trying to read a ORC file that references e.g. the "US/Pacific" timezone would crash. Here is a backtrace excerpt: ``` #12 0x00007f1a3ce23a55 in std::terminate() () from /lib/x86_64-linux-gnu/libstdc++.so.6 #13 0x00007f1a3ce39391 in __cxa_throw () from /lib/x86_64-linux-gnu/libstdc++.so.6 #14 0x00007f1a3f4accc4 in orc::loadTZDB(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) () from /tmp/arrow-HEAD.ArqTs/venv-wheel-3.12-manylinux_2_17_x86_64.manylinux2014_x86_64/lib/python3.12/site-packages/pyarrow/libarrow.so.1900 #15 0x00007f1a3f4ad392 in std::call_once<orc::LazyTimezone::getImpl() const::{lambda()#1}>(std::once_flag&, orc::LazyTimezone::getImpl() const::{lambda()#1}&&)::{lambda()#2}::_FUN() () from /tmp/arrow-HEAD.ArqTs/venv-wheel-3.12-manylinux_2_17_x86_64.manylinux2014_x86_64/lib/python3.12/site-packages/pyarrow/libarrow.so.1900 #16 0x00007f1a4298bec3 in __pthread_once_slow (once_control=0xa5ca7c8, init_routine=0x7f1a3ce69420 <__once_proxy>) at ./nptl/pthread_once.c:116 #17 0x00007f1a3f4a9ad0 in orc::LazyTimezone::getEpoch() const () from /tmp/arrow-HEAD.ArqTs/venv-wheel-3.12-manylinux_2_17_x86_64.manylinux2014_x86_64/lib/python3.12/site-packages/pyarrow/libarrow.so.1900 #18 0x00007f1a3f4e76b1 in orc::TimestampColumnReader::TimestampColumnReader(orc::Type const&, orc::StripeStreams&, bool) () from /tmp/arrow-HEAD.ArqTs/venv-wheel-3.12-manylinux_2_17_x86_64.manylinux2014_x86_64/lib/python3.12/site-packages/pyarrow/libarrow.so.1900 #19 0x00007f1a3f4e84ad in orc::buildReader(orc::Type const&, orc::StripeStreams&, bool, bool, bool) () from /tmp/arrow-HEAD.ArqTs/venv-wheel-3.12-manylinux_2_17_x86_64.manylinux2014_x86_64/lib/python3.12/site-packages/pyarrow/libarrow.so.1900 #20 0x00007f1a3f4e8dd7 in orc::StructColumnReader::StructColumnReader(orc::Type const&, orc::StripeStreams&, bool, bool) () from /tmp/arrow-HEAD.ArqTs/venv-wheel-3.12-manylinux_2_17_x86_64.manylinux2014_x86_64/lib/python3.12/site-packages/pyarrow/libarrow.so.1900 #21 0x00007f1a3f4e8532 in orc::buildReader(orc::Type const&, orc::StripeStreams&, bool, bool, bool) () from /tmp/arrow-HEAD.ArqTs/venv-wheel-3.12-manylinux_2_17_x86_64.manylinux2014_x86_64/lib/python3.12/site-packages/pyarrow/libarrow.so.1900 #22 0x00007f1a3f4925e9 in orc::RowReaderImpl::startNextStripe() () from /tmp/arrow-HEAD.ArqTs/venv-wheel-3.12-manylinux_2_17_x86_64.manylinux2014_x86_64/lib/python3.12/site-packages/pyarrow/libarrow.so.1900 #23 0x00007f1a3f492c9d in orc::RowReaderImpl::next(orc::ColumnVectorBatch&) () from /tmp/arrow-HEAD.ArqTs/venv-wheel-3.12-manylinux_2_17_x86_64.manylinux2014_x86_64/lib/python3.12/site-packages/pyarrow/libarrow.so.1900 #24 0x00007f1a3e6b251f in arrow::adapters::orc::ORCFileReader::Impl::ReadBatch(orc::RowReaderOptions const&, std::shared_ptr<arrow::Schema> const&, long) () from /tmp/arrow-HEAD.ArqTs/venv-wheel-3.12-manylinux_2_17_x86_64.manylinux2014_x86_64/lib/python3.12/site-packages/pyarrow/libarrow.so.1900 ``` ### What changes are included in this PR? Catch C++ exceptions when iterating ORC batches instead of letting them slip through. ### Are these changes tested? Yes. ### Are there any user-facing changes? No. * GitHub Issue: apache#40633 Authored-by: Antoine Pitrou <[email protected]> Signed-off-by: Sutou Kouhei <[email protected]>
Added support for making generator sources suitable for as-of-join. Also refactored a bit and added docstrings.