You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@arrow.apache.org by "jtanx (via GitHub)" <gi...@apache.org> on 2025/04/25 09:05:09 UTC
[I] [C++] [Acero] Threading issues in the groupby node [arrow]
jtanx opened a new issue, #46227:
URL: https://github.com/apache/arrow/issues/46227
### Describe the bug, including details regarding any error messages, version, and platform.
We have been intermittently noticing deadlocks in a fairly simple graph that uses the aggregation node. We've had difficulty in reliably reproducing the deadlock (where the main thread ends up indefinitely waiting on the exec plan's finished future while all the other threads are idle).
In trying to make a smaller test case, I have noticed what appears to be some pretty significant threading issues in the implementation of the GroupByNode.
Referring to the attached test case built with arrow at this revision: 18e8f50729c1f7e615f9647f2c4501efef0e7b26:
[test.zip](https://github.com/user-attachments/files/19905554/test.zip)
This performs the following aggregation:
```c++
ac::TableSourceNodeOptions srcOpts(tb);
ac::AggregateNodeOptions aggOpts(
{{"hash_min", nullptr, "avg_d_kbps", "min_avg_d_kbps"}}, {"year", "devices"});
auto decl = ac::Declaration::Sequence({
{"table_source", std::move(srcOpts)},
{"aggregate", std::move(aggOpts)},
});
``
Running this under helgrind with `valgrind --log-file=helgrind.log --tool=helgrind --history-backtrace-size=100 ./testit`, it reports the following trace (among many others):
```
==42024== ----------------------------------------------------------------
==42024==
==42024== Possible data race during read of size 8 at 0x57DEA30 by thread #4
==42024== Locks held: none
==42024== at 0x406E71: _M_ptr (unique_ptr.h:199)
==42024== by 0x406E71: get (unique_ptr.h:470)
==42024== by 0x406E71: operator bool (unique_ptr.h:487)
==42024== by 0x406E71: arrow::acero::aggregate::GroupByNode::OutputResult(bool) (groupby_aggregate_node.cc:342)
==42024== by 0x4073E7: arrow::acero::aggregate::GroupByNode::InputFinished(arrow::acero::ExecNode*, int) (groupby_aggregate_node.cc:396)
==42024== by 0x38CEAF: operator() (source_node.cc:226)
==42024== by 0x38CEAF: __invoke_impl<arrow::Status, arrow::acero::(anonymous namespace)::SourceNode::StartProducing()::<lambda(arrow::Result<int>)> mutable::<lambda()>&> (invoke.h:61)
==42024== by 0x38CEAF: __invoke_r<arrow::Status, arrow::acero::(anonymous namespace)::SourceNode::StartProducing()::<lambda(arrow::Result<int>)> mutable::<lambda()>&> (invoke.h:116)
==42024== by 0x38CEAF: std::_Function_handler<arrow::Status (), arrow::acero::(anonymous namespace)::SourceNode::StartProducing()::{lambda(arrow::Result<int>)#1}::operator()(arrow::Result<int>)::{lambda()#1}>::_M_invoke(std::_Any_data const&) (std_function.h:291)
==42024== by 0x3730AF: operator() (std_function.h:591)
==42024== by 0x3730AF: operator()<std::function<arrow::Status()>&> (future.h:150)
==42024== by 0x3730AF: __invoke_impl<void, arrow::detail::ContinueFuture&, arrow::Future<arrow::internal::Empty>&, std::function<arrow::Status()>&> (invoke.h:61)
==42024== by 0x3730AF: __invoke<arrow::detail::ContinueFuture&, arrow::Future<arrow::internal::Empty>&, std::function<arrow::Status()>&> (invoke.h:96)
==42024== by 0x3730AF: __call<void, 0, 1> (functional:506)
==42024== by 0x3730AF: operator()<> (functional:591)
==42024== by 0x3730AF: arrow::internal::FnOnce<void ()>::FnImpl<std::_Bind<arrow::detail::ContinueFuture (arrow::Future<arrow::internal::Empty>, std::function<arrow::Status ()>)> >::invoke() (functional.h:152)
==42024== by 0x14B8F6C: operator() (functional.h:140)
==42024== by 0x14B8F6C: WorkerLoop (thread_pool.cc:478)
==42024== by 0x14B8F6C: operator() (thread_pool.cc:643)
==42024== by 0x14B8F6C: __invoke_impl<void, arrow::internal::ThreadPool::LaunchWorkersUnlocked(int)::<lambda()> > (invoke.h:61)
==42024== by 0x14B8F6C: __invoke<arrow::internal::ThreadPool::LaunchWorkersUnlocked(int)::<lambda()> > (invoke.h:96)
==42024== by 0x14B8F6C: _M_invoke<0> (std_thread.h:292)
==42024== by 0x14B8F6C: operator() (std_thread.h:299)
==42024== by 0x14B8F6C: std::thread::_State_impl<std::thread::_Invoker<std::tuple<arrow::internal::ThreadPool::LaunchWorkersUnlocked(int)::{lambda()#1}> > >::_M_run() (std_thread.h:244)
==42024== by 0x4B69DB3: ??? (in /usr/lib/x86_64-linux-gnu/libstdc++.so.6.0.33)
==42024== by 0x4854B7A: ??? (in /usr/libexec/valgrind/vgpreload_helgrind-amd64-linux.so)
==42024== by 0x4EAEAA3: start_thread (pthread_create.c:447)
==42024== by 0x4F3BA33: clone (clone.S:100)
==42024==
==42024== This conflicts with a previous write of size 8 by thread #3
==42024== Locks held: none
==42024== at 0x405C8A: reset (unique_ptr.h:209)
==42024== by 0x405C8A: operator= (unique_ptr.h:191)
==42024== by 0x405C8A: operator= (unique_ptr.h:243)
==42024== by 0x405C8A: operator= (unique_ptr.h:414)
==42024== by 0x405C8A: arrow::acero::aggregate::GroupByNode::InitLocalStateIfNeeded(arrow::acero::aggregate::GroupByNode::ThreadLocalState*) (groupby_aggregate_node.cc:428)
==42024== by 0x407C45: arrow::acero::aggregate::GroupByNode::Consume(arrow::compute::ExecSpan) (groupby_aggregate_node.cc:218)
==42024== by 0x408A2C: operator() (groupby_aggregate_node.cc:376)
==42024== by 0x408A2C: HandleSegments<arrow::acero::aggregate::GroupByNode::InputReceived(arrow::acero::ExecNode*, arrow::compute::ExecBatch)::<lambda(const arrow::compute::ExecBatch&, const arrow::compute::Segment&)> > (aggregate_internal.h:139)
==42024== by 0x408A2C: arrow::acero::aggregate::GroupByNode::InputReceived(arrow::acero::ExecNode*, arrow::compute::ExecBatch) (groupby_aggregate_node.cc:382)
==42024== by 0x3943D5: operator() (source_node.cc:158)
==42024== by 0x3943D5: __invoke_impl<arrow::Status, arrow::acero::(anonymous namespace)::SourceNode::SliceAndDeliverMorsel(const arrow::compute::ExecBatch&)::<lambda()>&> (invoke.h:61)
==42024== by 0x3943D5: __invoke_r<arrow::Status, arrow::acero::(anonymous namespace)::SourceNode::SliceAndDeliverMorsel(const arrow::compute::ExecBatch&)::<lambda()>&> (invoke.h:116)
==42024== by 0x3943D5: std::_Function_handler<arrow::Status (), arrow::acero::(anonymous namespace)::SourceNode::SliceAndDeliverMorsel(arrow::compute::ExecBatch const&)::{lambda()#1}>::_M_invoke(std::_Any_data const&) (std_function.h:291)
==42024== by 0x3730AF: operator() (std_function.h:591)
==42024== by 0x3730AF: operator()<std::function<arrow::Status()>&> (future.h:150)
==42024== by 0x3730AF: __invoke_impl<void, arrow::detail::ContinueFuture&, arrow::Future<arrow::internal::Empty>&, std::function<arrow::Status()>&> (invoke.h:61)
==42024== by 0x3730AF: __invoke<arrow::detail::ContinueFuture&, arrow::Future<arrow::internal::Empty>&, std::function<arrow::Status()>&> (invoke.h:96)
==42024== by 0x3730AF: __call<void, 0, 1> (functional:506)
==42024== by 0x3730AF: operator()<> (functional:591)
==42024== by 0x3730AF: arrow::internal::FnOnce<void ()>::FnImpl<std::_Bind<arrow::detail::ContinueFuture (arrow::Future<arrow::internal::Empty>, std::function<arrow::Status ()>)> >::invoke() (functional.h:152)
==42024== by 0x14B8F6C: operator() (functional.h:140)
==42024== by 0x14B8F6C: WorkerLoop (thread_pool.cc:478)
==42024== by 0x14B8F6C: operator() (thread_pool.cc:643)
==42024== by 0x14B8F6C: __invoke_impl<void, arrow::internal::ThreadPool::LaunchWorkersUnlocked(int)::<lambda()> > (invoke.h:61)
==42024== by 0x14B8F6C: __invoke<arrow::internal::ThreadPool::LaunchWorkersUnlocked(int)::<lambda()> > (invoke.h:96)
==42024== by 0x14B8F6C: _M_invoke<0> (std_thread.h:292)
==42024== by 0x14B8F6C: operator() (std_thread.h:299)
==42024== by 0x14B8F6C: std::thread::_State_impl<std::thread::_Invoker<std::tuple<arrow::internal::ThreadPool::LaunchWorkersUnlocked(int)::{lambda()#1}> > >::_M_run() (std_thread.h:244)
==42024== by 0x4B69DB3: ??? (in /usr/lib/x86_64-linux-gnu/libstdc++.so.6.0.33)
==42024== by 0x4854B7A: ??? (in /usr/libexec/valgrind/vgpreload_helgrind-amd64-linux.so)
==42024== by 0x4EAEAA3: start_thread (pthread_create.c:447)
==42024== by 0x4F3BA33: clone (clone.S:100)
==42024== Address 0x57dea30 is 0 bytes inside a block of size 352 alloc'd
==42024== at 0x4849023: operator new(unsigned long) (in /usr/libexec/valgrind/vgpreload_helgrind-amd64-linux.so)
==42024== by 0x40CF76: allocate (new_allocator.h:151)
==42024== by 0x40CF76: allocate (alloc_traits.h:482)
==42024== by 0x40CF76: _M_allocate (stl_vector.h:381)
==42024== by 0x40CF76: _M_allocate (stl_vector.h:378)
==42024== by 0x40CF76: std::vector<arrow::acero::aggregate::GroupByNode::ThreadLocalState, std::allocator<arrow::acero::aggregate::GroupByNode::ThreadLocalState> >::_M_default_append(unsigned long) (vector.tcc:663)
==42024== by 0x40D1D2: resize (stl_vector.h:1016)
==42024== by 0x40D1D2: arrow::acero::aggregate::GroupByNode::StartProducing() (aggregate_internal.h:289)
==42024== by 0x32B8EA: operator() (exec_plan.cc:175)
==42024== by 0x32B8EA: arrow::internal::FnOnce<arrow::Status (arrow::util::AsyncTaskScheduler*)>::FnImpl<arrow::acero::(anonymous namespace)::ExecPlanImpl::StartProducing()::{lambda(arrow::util::AsyncTaskScheduler*)#1}>::invoke(arrow::util::AsyncTaskScheduler*&&) (functional.h:152)
==42024== by 0x143A479: operator() (functional.h:140)
==42024== by 0x143A479: arrow::util::AsyncTaskScheduler::Make(arrow::internal::FnOnce<arrow::Status (arrow::util::AsyncTaskScheduler*)>, arrow::internal::FnOnce<void (arrow::Status const&)>, arrow::StopToken) (async_util.cc:471)
==42024== by 0x329640: arrow::acero::(anonymous namespace)::ExecPlanImpl::StartProducing() (exec_plan.cc:193)
==42024== by 0x332AAB: StartProducing (exec_plan.cc:439)
==42024== by 0x332AAB: arrow::acero::(anonymous namespace)::DeclarationToTableImpl(arrow::acero::Declaration, arrow::acero::QueryOptions, arrow::internal::Executor*) (exec_plan.cc:662)
==42024== by 0x333433: operator() (exec_plan.cc:787)
==42024== by 0x333433: arrow::internal::FnOnce<arrow::Future<std::shared_ptr<arrow::Table> > (arrow::internal::Executor*)>::FnImpl<arrow::acero::DeclarationToTable(arrow::acero::Declaration, bool, arrow::MemoryPool*, arrow::compute::FunctionRegistry*)::{lambda(arrow::internal::Executor*)#1}>::invoke(arrow::internal::Executor*&&) (functional.h:152)
==42024== by 0x342B55: operator() (functional.h:140)
==42024== by 0x342B55: arrow::Future<std::shared_ptr<arrow::Table> >::SyncType arrow::internal::RunSynchronously<arrow::Future<std::shared_ptr<arrow::Table> >, std::shared_ptr<arrow::Table> >(arrow::internal::FnOnce<arrow::Future<std::shared_ptr<arrow::Table> > (arrow::internal::Executor*)>, bool) (thread_pool.h:587)
==42024== by 0x334F63: arrow::acero::DeclarationToTable(arrow::acero::Declaration, bool, arrow::MemoryPool*, arrow::compute::FunctionRegistry*) (exec_plan.cc:789)
==42024== by 0x320A8F: RunTest() (testit.cc:32)
==42024== by 0x30B824: main (testit.cc:39)
==42024== Block was alloc'd by thread #1
==42024==
==42024== ----------------------------------------------------------------
```
Inspecting how [`InputRecieved`](https://github.com/apache/arrow/blob/4d566e60c5a12f047e1c0f36b9733a56513a7d43/cpp/src/arrow/acero/groupby_aggregate_node.cc#L391) and [`InputFinished`](https://github.com/apache/arrow/blob/4d566e60c5a12f047e1c0f36b9733a56513a7d43/cpp/src/arrow/acero/groupby_aggregate_node.cc#L401) are implemented in the groupby node, along with what the source node does - it uses the task scheduler to call [`InputReceived()` on the downstream node](https://github.com/apache/arrow/blob/4d566e60c5a12f047e1c0f36b9733a56513a7d43/cpp/src/arrow/acero/source_node.cc#L134) and similarly so with [`InputFinished()`](https://github.com/apache/arrow/blob/4d566e60c5a12f047e1c0f36b9733a56513a7d43/cpp/src/arrow/acero/source_node.cc#L224-L227) - I can see how this race can happen.
Presumably the task scheduler does not guarantee any sort of ordering over which is called first between InputReceived() or InputFinished() - furthermore they can be called simultaneously off different threads. This implies that both of these methods in the groupby node must be threadsafe - which therefore implies that [OutputResult](https://github.com/apache/arrow/blob/4d566e60c5a12f047e1c0f36b9733a56513a7d43/cpp/src/arrow/acero/groupby_aggregate_node.cc#L346-L354) must also be threadsafe. But the way that `OutputResult` is currently implemented is clearly not - it first modifies the thread-local state (potentially swapping entries; these entries could also potentially be populated during this iteration if the other threads have just spawned) and further merges the thread-local state.
I think this explains the above race - InputFinished is being called from one thread while another thread is in the middle of calling InputReceived - and both end up in OutputResult when they should not. This is also causing other issues such as the execution plan to be marked as finished even before the source node has called InputFinished on the aggregation node - which means that valgrind is also observing that InputFinished could be called well after things have been destroyed.
So it looks like there is a clear issue here - but I am not familiar enough with the intended execution/threading model to say what the fix should be. Some suggestions here would be appreciated
### Component(s)
C++
--
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: issues-unsubscribe@arrow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [C++] [Acero] Threading issues in the groupby node [arrow]
Posted by "zanmato1984 (via GitHub)" <gi...@apache.org>.
zanmato1984 commented on issue #46227:
URL: https://github.com/apache/arrow/issues/46227#issuecomment-2840273227
Hi @jtanx , thanks for the reporting and the further digging. As you mentioned these seem quite like false positives as helgrind has known limitation in understanding modern C++ `atomic`s.
> (although I'm not sure about [this case](https://github.com/apache/arrow/blob/4d566e60c5a12f047e1c0f36b9733a56513a7d43/cpp/src/arrow/acero/groupby_aggregate_node.cc#L377-L378))
This is also OK. The `OutputResult` here is only invoked single-threaded, see [1]. In other words, if ran multi-threaded, the condition `if (!segment.extends && segment.offset == 0)` never holds.
So I'm closing this as a non-issue. Feel free to reopen if you find anything else. Thanks!
[1] https://github.com/apache/arrow/blob/fc0862a25fa2d6aa17df14daa9ad37430d432425/cpp/src/arrow/acero/groupby_aggregate_node.cc#L128-L132
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [C++] [Acero] Threading issues in the groupby node [arrow]
Posted by "zanmato1984 (via GitHub)" <gi...@apache.org>.
zanmato1984 closed issue #46227: [C++] [Acero] Threading issues in the groupby node
URL: https://github.com/apache/arrow/issues/46227
--
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: issues-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [C++] [Acero] Threading issues in the groupby node [arrow]
Posted by "jtanx (via GitHub)" <gi...@apache.org>.
jtanx commented on issue #46227:
URL: https://github.com/apache/arrow/issues/46227#issuecomment-2830043076
After some further digging, I see that the synchronization is basically accounted for by `input_counter_` (although I'm not sure about [this case](https://github.com/apache/arrow/blob/4d566e60c5a12f047e1c0f36b9733a56513a7d43/cpp/src/arrow/acero/groupby_aggregate_node.cc#L377-L378)). I'm guessing that these are more likely to be false positives from helgrind not handling atomic variable access correctly.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org