GeneratorStream & Acero deadlock

gistfile1.txt

Thread 85 (Thread 0x7f67519fe6c0 (LWP 1204905) "python3.11"):
#0  0x00007f679a2a9169 in __futex_abstimed_wait_common64 (private=0, cancel=true, abstime=0x7f67519fd3a0, op=137, expected=0, futex_word=0x7f679a8e7d50 <_PyRuntime+432>) at futex-internal.c:57
#1  __futex_abstimed_wait_common (futex_word=futex_word@entry=0x7f679a8e7d50 <_PyRuntime+432>, expected=expected@entry=0, clockid=clockid@entry=1, abstime=abstime@entry=0x7f67519fd3a0, private=private@entry=0, cancel=cancel@entry=true) at futex-internal.c:87
#2  0x00007f679a2a91ef in __GI___futex_abstimed_wait_cancelable64 (futex_word=futex_word@entry=0x7f679a8e7d50 <_PyRuntime+432>, expected=expected@entry=0, clockid=clockid@entry=1, abstime=abstime@entry=0x7f67519fd3a0, private=private@entry=0) at futex-internal.c:139
#3  0x00007f679a2abe72 in __pthread_cond_wait_common (abstime=0x7f67519fd3a0, clockid=1, mutex=0x7f679a2ab830 <__condvar_cleanup_waiting>, cond=0x7f679a8e7d28 <_PyRuntime+392>) at pthread_cond_wait.c:503
#4  ___pthread_cond_timedwait64 (cond=cond@entry=0x7f679a8e7d28 <_PyRuntime+392>, mutex=mutex@entry=0x7f679a8e7d58 <_PyRuntime+440>, abstime=0x7f67519fd3a0) at pthread_cond_wait.c:643
#5  0x00007f679a5a0340 in PyCOND_TIMEDWAIT (us=<optimized out>, mut=<optimized out>, cond=0x7f679a8e7d28 <_PyRuntime+392>) at /usr/src/debug/python3.11-3.11.7-2.fc39.x86_64/Python/condvar.h:73
#6  take_gil (tstate=tstate@entry=0x7f67800339f0) at /usr/src/debug/python3.11-3.11.7-2.fc39.x86_64/Python/ceval_gil.h:231
#7  0x00007f679a5b8bb6 in PyEval_RestoreThread (tstate=0x7f67800339f0) at /usr/src/debug/python3.11-3.11.7-2.fc39.x86_64/Python/ceval.c:521
#8  0x00007f679a66895a in PyGILState_Ensure () at /usr/src/debug/python3.11-3.11.7-2.fc39.x86_64/Python/pystate.c:1712
#9  0x00007f678c59bbe3 in arrow::py::(anonymous namespace)::PythonUdfExec(arrow::compute::KernelContext*, arrow::compute::ExecSpan const&, arrow::compute::ExecResult*) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_python.so
#10 0x00007f6789c30fff in arrow::compute::detail::(anonymous namespace)::ScalarExecutor::Execute(arrow::compute::ExecBatch const&, arrow::compute::detail::ExecListener*) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow.so.1500
#11 0x00007f6789c37008 in arrow::compute::ExecuteScalarExpression(arrow::compute::Expression const&, arrow::compute::ExecBatch const&, arrow::compute::ExecContext*) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow.so.1500
#12 0x00007f678c163f1d in arrow::acero::(anonymous namespace)::ProjectNode::ProcessBatch(arrow::compute::ExecBatch) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_acero.so.1500
#13 0x00007f678c195c45 in arrow::acero::MapNode::InputReceived(arrow::acero::ExecNode*, arrow::compute::ExecBatch) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_acero.so.1500
#14 0x00007f678c176e86 in std::_Function_handler<arrow::Status (), arrow::acero::(anonymous namespace)::SourceNode::SliceAndDeliverMorsel(arrow::compute::ExecBatch const&)::{lambda()#1}>::_M_invoke(std::_Any_data const&) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_acero.so.1500
#15 0x00007f678c17d73b in arrow::internal::FnOnce<void ()>::FnImpl<std::_Bind<arrow::detail::ContinueFuture (arrow::Future<arrow::internal::Empty>, std::function<arrow::Status ()>)> >::invoke() () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_acero.so.1500
#16 0x00007f6789d36ebc in std::thread::_State_impl<std::thread::_Invoker<std::tuple<arrow::internal::ThreadPool::LaunchWorkersUnlocked(int)::{lambda()#1}> > >::_M_run() () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow.so.1500
#17 0x00007f67888e31e3 in std::execute_native_thread_routine (__p=0x7f6780034190) at ../../../../../libstdc++-v3/src/c++11/thread.cc:104
#18 0x00007f679a2ac897 in start_thread (arg=<optimized out>) at pthread_create.c:444
#19 0x00007f679a33380c in clone3 () at ../sysdeps/unix/sysv/linux/x86_64/clone3.S:78

Thread 69 (Thread 0x7f675b28a6c0 (LWP 1204680) "grpcpp_sync_ser"):
#0  0x00007f679a2a9169 in __futex_abstimed_wait_common64 (private=0, cancel=true, abstime=0x0, op=393, expected=0, futex_word=0x55dc27e5e2d8) at futex-internal.c:57
#1  __futex_abstimed_wait_common (futex_word=futex_word@entry=0x55dc27e5e2d8, expected=expected@entry=0, clockid=clockid@entry=0, abstime=abstime@entry=0x0, private=private@entry=0, cancel=cancel@entry=true) at futex-internal.c:87
#2  0x00007f679a2a91ef in __GI___futex_abstimed_wait_cancelable64 (futex_word=futex_word@entry=0x55dc27e5e2d8, expected=expected@entry=0, clockid=clockid@entry=0, abstime=abstime@entry=0x0, private=private@entry=0) at futex-internal.c:139
#3  0x00007f679a2abb09 in __pthread_cond_wait_common (abstime=0x0, clockid=0, mutex=<optimized out>, cond=0x55dc27e5e2b0) at pthread_cond_wait.c:503
#4  ___pthread_cond_wait (cond=0x55dc27e5e2b0, mutex=<optimized out>) at pthread_cond_wait.c:618
#5  0x00007f67888b32b7 in __gthread_cond_wait (__mutex=<optimized out>, __cond=<optimized out>) at /usr/src/debug/gcc-13.2.1-6.fc39.x86_64/obj-x86_64-redhat-linux/x86_64-redhat-linux/libstdc++-v3/include/x86_64-redhat-linux/bits/gthr-default.h:865
#6  std::__condvar::wait (__m=..., this=<optimized out>) at /usr/src/debug/gcc-13.2.1-6.fc39.x86_64/obj-x86_64-redhat-linux/x86_64-redhat-linux/libstdc++-v3/include/bits/std_mutex.h:171
#7  __gnu_cxx::__nothrow_wait_cv::wait (this=<optimized out>, lock=...) at ../../../../libstdc++-v3/src/c++11/compatibility-condvar.cc:99
#8  0x00007f6789d417a3 in arrow::FutureImpl::Wait() () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow.so.1500
#9  0x00007f678c1dda2f in arrow::Result<std::shared_ptr<arrow::RecordBatch> > arrow::Iterator<std::shared_ptr<arrow::RecordBatch> >::Next<arrow::GeneratorIterator<std::shared_ptr<arrow::RecordBatch> > >(void*) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_acero.so.1500
#10 0x00007f678c1c1eea in arrow::acero::DeclarationToReader(arrow::acero::Declaration, arrow::acero::QueryOptions)::PlanReader::ReadNext(std::shared_ptr<arrow::RecordBatch>*) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_acero.so.1500
#11 0x00007f677ec1d2f8 in arrow::flight::RecordBatchStream::RecordBatchStreamImpl::Next(arrow::flight::FlightPayload*) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_flight.so.1500
#12 0x00007f677ec0daff in arrow::flight::RecordBatchStream::Next() () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_flight.so.1500
#13 0x00007f678773fff6 in arrow::py::flight::PyFlightDataStream::Next() () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_python_flight.so
#14 0x00007f6784da945c in __pyx_f_7pyarrow_7_flight__data_stream_next(void*, arrow::flight::FlightPayload*) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/_flight.cpython-311-x86_64-linux-gnu.so
#15 0x00007f6784dc06fa in std::_Function_handler<arrow::Status (_object*, arrow::flight::FlightPayload*), arrow::Status (*)(void*, arrow::flight::FlightPayload*)>::_M_invoke(std::_Any_data const&, _object*&&, arrow::flight::FlightPayload*&&) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/_flight.cpython-311-x86_64-linux-gnu.so
#16 0x00007f6787744c95 in arrow::py::flight::PyGeneratorFlightDataStream::Next() () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_python_flight.so
#17 0x00007f678773fff6 in arrow::py::flight::PyFlightDataStream::Next() () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_python_flight.so
#18 0x00007f677ebee019 in arrow::flight::internal::ServerTransport::DoGet(arrow::flight::ServerCallContext const&, arrow::flight::Ticket const&, arrow::flight::internal::ServerDataStream*) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_flight.so.1500
#19 0x00007f677ebf5780 in arrow::flight::transport::grpc::(anonymous namespace)::GrpcServiceHandler::DoGet(grpc::ServerContext*, arrow::flight::protocol::Ticket const*, grpc::ServerWriter<arrow::flight::protocol::FlightData>*) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_flight.so.1500
#20 0x00007f677eba4a1e in std::_Function_handler<grpc::Status (arrow::flight::protocol::FlightService::Service*, grpc::ServerContext*, arrow::flight::protocol::Ticket const*, grpc::ServerWriter<arrow::flight::protocol::FlightData>*), arrow::flight::protocol::FlightService::Service::ServerWriter()::{lambda(arrow::flight::protocol::FlightService::Service*, grpc::ServerContext*, arrow::flight::protocol::Ticket const*, grpc::ServerWriter<arrow::flight::protocol::FlightData>*)#6}>::_M_invoke(std::_Any_data const&, arrow::flight::protocol::FlightService::Service*&&, grpc::ServerContext*&&, arrow::flight::protocol::Ticket const*&&, grpc::ServerWriter<arrow::flight::protocol::FlightData>*&&) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_flight.so.1500
#21 0x00007f677ebe7478 in grpc::internal::ServerStreamingHandler<arrow::flight::protocol::FlightService::Service, arrow::flight::protocol::Ticket, arrow::flight::protocol::FlightData>::RunHandler(grpc::internal::MethodHandler::HandlerParameter const&) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_flight.so.1500
#22 0x00007f677ec4321c in grpc::Server::SyncRequest::ContinueRunAfterInterception() () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_flight.so.1500
#23 0x00007f677ec496eb in grpc::ThreadManager::MainWorkLoop() () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_flight.so.1500
#24 0x00007f677ec4981c in grpc::ThreadManager::WorkerThread::Run() () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_flight.so.1500
#25 0x00007f677f0fefb3 in grpc_core::(anonymous namespace)::ThreadInternalsPosix::ThreadInternalsPosix(char const*, void (*)(void*), void*, bo--Type <RET> for more, q to quit, c to continue without paging--
ol*, grpc_core::Thread::Options const&)::{lambda(void*)#1}::_FUN(void*) () from /whatever/.venv/lib64/python3.11/site-packages/pyarrow/libarrow_flight.so.1500
#26 0x00007f679a2ac897 in start_thread (arg=<optimized out>) at pthread_create.c:444
#27 0x00007f679a33380c in clone3 () at ../sysdeps/unix/sysv/linux/x86_64/clone3.S:78