[C++] Fix ASAN issue in the arrow-dataset-dataset-writer-test

[wgtmac]

Describe the bug, including details regarding any error messages, version, and platform.

https://github.com/apache/arrow/actions/runs/12739705161/job/35503788331?pr=45226

=================================================================
==9882==ERROR: LeakSanitizer: detected memory leaks

Indirect leak of 512 byte(s) in 1 object(s) allocated from:
 #0 0x55777af7ebed in operator new(unsigned long) (/build/cpp/debug/arrow-dataset-dataset-writer-test+0x419bed) (BuildId: 22e936dc51f9bb18fc25ab7de9ec7c7ba97cc2c2)
 #1 0x55777b1376c0 in __gnu_cxx::new_allocator<std::shared_ptr<arrow::RecordBatch> >::allocate(unsigned long, void const*) /usr/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/ext/new_allocator.h:127:27
 #2 0x55777b13754c in std::allocator_traits<std::allocator<std::shared_ptr<arrow::RecordBatch> > >::allocate(std::allocator<std::shared_ptr<arrow::RecordBatch> >&, unsigned long) /usr/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/alloc_traits.h:464:20
 #3 0x7f11941d29dc in std::_Deque_base<std::shared_ptr<arrow::RecordBatch>, std::allocator<std::shared_ptr<arrow::RecordBatch> > >::_M_allocate_node() /usr/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_deque.h:562:9
 #4 0x7f119420bdbd in std::_Deque_base<std::shared_ptr<arrow::RecordBatch>, std::allocator<std::shared_ptr<arrow::RecordBatch> > >::_M_create_nodes(std::shared_ptr<arrow::RecordBatch>**, std::shared_ptr<arrow::RecordBatch>**) /usr/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_deque.h:663:21
 #5 0x7f119420ae17 in std::_Deque_base<std::shared_ptr<arrow::RecordBatch>, std::allocator<std::shared_ptr<arrow::RecordBatch> > >::_M_initialize_map(unsigned long) /usr/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_deque.h:637:4
 #6 0x7f119420a2f4 in std::_Deque_base<std::shared_ptr<arrow::RecordBatch>, std::allocator<std::shared_ptr<arrow::RecordBatch> > >::_Deque_base() /usr/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_deque.h:439:9
 #7 0x7f119420a15f in std::deque<std::shared_ptr<arrow::RecordBatch>, std::allocator<std::shared_ptr<arrow::RecordBatch> > >::deque() /usr/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/stl_deque.h:834:7
 #8 0x7f1194209e95 in arrow::dataset::internal::(anonymous namespace)::DatasetWriterFileQueue::DatasetWriterFileQueue(std::shared_ptr<arrow::Schema> const&, arrow::dataset::FileSystemDatasetWriteOptions const&, arrow::dataset::internal::(anonymous namespace)::DatasetWriterState*) /arrow/cpp/src/arrow/dataset/dataset_writer.cc:136:12
 #9 0x7f1194208165 in std::_MakeUniq<arrow::dataset::internal::(anonymous namespace)::DatasetWriterFileQueue>::__single_object std::make_unique<arrow::dataset::internal::(anonymous namespace)::DatasetWriterFileQueue, std::shared_ptr<arrow::Schema>&, arrow::dataset::FileSystemDatasetWriteOptions const&, arrow::dataset::internal::(anonymous namespace)::DatasetWriterState*&>(std::shared_ptr<arrow::Schema>&, arrow::dataset::FileSystemDatasetWriteOptions const&, arrow::dataset::internal::(anonymous namespace)::DatasetWriterState*&) /usr/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/unique_ptr.h:962:34
 #10 0x7f1194206a02 in arrow::dataset::internal::(anonymous namespace)::DatasetWriterDirectoryQueue::OpenFileQueue(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) /arrow/cpp/src/arrow/dataset/dataset_writer.cc:343:9
 #11 0x7f119418667a in arrow::dataset::internal::(anonymous namespace)::DatasetWriterDirectoryQueue::StartWrite(std::shared_ptr<arrow::RecordBatch> const&) /arrow/cpp/src/arrow/dataset/dataset_writer.cc:306:7
 #12 0x7f119417fcdd in arrow::dataset::internal::DatasetWriter::DatasetWriterImpl::DoWriteRecordBatch(std::shared_ptr<arrow::RecordBatch>, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) /arrow/cpp/src/arrow/dataset/dataset_writer.cc:660:27
 #13 0x7f119417c5b8 in arrow::dataset::internal::DatasetWriter::DatasetWriterImpl::WriteAndCheckBackpressure(std::shared_ptr<arrow::RecordBatch>, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) /arrow/cpp/src/arrow/dataset/dataset_writer.cc:539:14
 #14 0x7f119417b6ba in arrow::dataset::internal::DatasetWriter::DatasetWriterImpl::WriteRecordBatch(std::shared_ptr<arrow::RecordBatch>, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&)::'lambda'()::operator()() /arrow/cpp/src/arrow/dataset/dataset_writer.cc:565:15
 #15 0x7f1194179b2b in arrow::util::AsyncTaskScheduler::SimpleTask<arrow::dataset::internal::DatasetWriter::DatasetWriterImpl::WriteRecordBatch(std::shared_ptr<arrow::RecordBatch>, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&)::'lambda'()>::operator()() /arrow/cpp/src/arrow/util/async_util.h:153:53
 #16 0x7f117fb90b19 in arrow::util::(anonymous namespace)::AsyncTaskGroupImpl::AddTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >)::WrapperTask::operator()() /arrow/cpp/src/arrow/util/async_util.cc:433:9
 #17 0x7f117fb7d938 in arrow::util::(anonymous namespace)::ThrottledAsyncTaskSchedulerImpl::SubmitTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >, int, bool)::'lambda'()::operator()() /arrow/cpp/src/arrow/util/async_util.cc:351:11
 #18 0x7f117fb7d356 in arrow::util::AsyncTaskScheduler::SimpleTask<arrow::util::(anonymous namespace)::ThrottledAsyncTaskSchedulerImpl::SubmitTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >, int, bool)::'lambda'()>::operator()() /arrow/cpp/src/arrow/util/async_util.h:153:53
 #19 0x7f117fb57813 in arrow::util::(anonymous namespace)::AsyncTaskSchedulerImpl::DoSubmitTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >) /arrow/cpp/src/arrow/util/async_util.cc:197:38
 #20 0x7f117fb55d31 in arrow::util::(anonymous namespace)::AsyncTaskSchedulerImpl::SubmitTaskUnlocked(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >, std::unique_lock<std::mutex>&&) /arrow/cpp/src/arrow/util/async_util.cc:267:12
 #21 0x7f117fb5548d in arrow::util::(anonymous namespace)::AsyncTaskSchedulerImpl::AddTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >) /arrow/cpp/src/arrow/util/async_util.cc:173:5
 #22 0x7f117fb7bf9a in bool arrow::util::AsyncTaskScheduler::AddSimpleTask<arrow::util::(anonymous namespace)::ThrottledAsyncTaskSchedulerImpl::SubmitTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >, int, bool)::'lambda'()>(arrow::util::(anonymous namespace)::ThrottledAsyncTaskSchedulerImpl::SubmitTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >, int, bool)::'lambda'(), std::basic_string_view<char, std::char_traits<char> >) /arrow/cpp/src/arrow/util/async_util.h:171:12
 #23 0x7f117fb7544b in arrow::util::(anonymous namespace)::ThrottledAsyncTaskSchedulerImpl::SubmitTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >, int, bool) /arrow/cpp/src/arrow/util/async_util.cc:348:21
 #24 0x7f117fb7216a in arrow::util::(anonymous namespace)::ThrottledAsyncTaskSchedulerImpl::AddTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >) /arrow/cpp/src/arrow/util/async_util.cc:331:14
 #25 0x7f117fb87b6f in arrow::util::(anonymous namespace)::AsyncTaskGroupImpl::AddTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >) /arrow/cpp/src/arrow/util/async_util.cc:446:21
 #26 0x7f117fb999f8 in arrow::util::ThrottledAsyncTaskGroup::AddTask(std::unique_ptr<arrow::util::AsyncTaskScheduler::Task, std::default_delete<arrow::util::AsyncTaskScheduler::Task> >) /arrow/cpp/src/arrow/util/async_util.cc:511:25
 #27 0x7f119417796a in bool arrow::util::AsyncTaskScheduler::AddSimpleTask<arrow::dataset::internal::DatasetWriter::DatasetWriterImpl::WriteRecordBatch(std::shared_ptr<arrow::RecordBatch>, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&)::'lambda'()>(arrow::dataset::internal::DatasetWriter::DatasetWriterImpl::WriteRecordBatch(std::shared_ptr<arrow::RecordBatch>, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&)::'lambda'(), std::basic_string_view<char, std::char_traits<char> >) /arrow/cpp/src/arrow/util/async_util.h:171:12
 #28 0x7f119416c4cb in arrow::dataset::internal::DatasetWriter::DatasetWriterImpl::WriteRecordBatch(std::shared_ptr<arrow::RecordBatch>, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) /arrow/cpp/src/arrow/dataset/dataset_writer.cc:562:19
 #29 0x7f119416bdec in arrow::dataset::internal::DatasetWriter::WriteRecordBatch(std::shared_ptr<arrow::RecordBatch>, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) /arrow/cpp/src/arrow/dataset/dataset_writer.cc:725:17

Component(s)

C++

[wgtmac]

@kou @zanmato1984 Have you seen this? It is not stable to reproduce but I have seen it for several times.

[zanmato1984]

I remember seeing it regularly in the CI jobs of my recent several PRs. Eg. in #45108, https://github.com/apache/arrow/actions/runs/12644487138/job/35232182454?pr=45108

[amoeba]

I saw this today in CI for #46180 and was also able to reproduce it locally (debian sid, amd64) once but not after repeated calls.

Edit: I can pretty consistently reproduce the leak on the above system if re-run the test in a while loop.

[pitrou]

This seems to be fixed by #46711, though I'm trying to run multiple iterations to confirm.

[amoeba]

Since I had a very reliable local reproduction, I tested #46711 I think it's at least way better. I previous could get ASAN to fail within a couple seconds of running,

while true; ASAN_OPTIONS=detect_leaks=1 debug/arrow-dataset-dataset-writer-test || break; end;

With your PR, I was able to run that loop for a few minutes but it did hang eventually:

...>8...
[ FAILED ] 1 test, listed below:
[ FAILED ] DatasetWriterTestFixture.MaxRowsOneWriteBackpresure

 1 FAILED TEST
==36691==LeakSanitizer has encountered a fatal error.
==36691==HINT: For debugging, try setting environment variable LSAN_OPTIONS=verbosity=1:log_threads=1
==36691==HINT: LeakSanitizer does not work under ptrace (strace, gdb, etc)

I did manage to get a stack trace,

gdb bt dump

(gdb) bt
#0 0x00007ffb7b0a49ee in ?? () from /lib/x86_64-linux-gnu/libc.so.6
#1 0x00007ffb7b099668 in ?? () from /lib/x86_64-linux-gnu/libc.so.6
#2 0x00007ffb7b099c9c in ?? () from /lib/x86_64-linux-gnu/libc.so.6
#3 0x00007ffb7b09c515 in pthread_cond_clockwait () from /lib/x86_64-linux-gnu/libc.so.6
#4 0x00007ffb81ae8f51 in std::__condvar::wait_until (this=0x50e000019e90, __m=..., __clock=1, __abs_time=...)
 at /usr/include/c++/14/bits/std_mutex.h:185
#5 0x00007ffb81aec05d in std::condition_variable::__wait_until_impl<std::chrono::duration<long, std::ratio<1l, 1000000000l> > > (
 this=0x50e000019e90, __lock=..., __atime=std::chrono::_V2::steady_clock time_point = { 1052772720473ns })
 at /usr/include/c++/14/condition_variable:203
#6 0x00007ffb81aeb70b in std::condition_variable::wait_until<std::chrono::duration<long, std::ratio<1l, 1000000000l> > > (this=0x50e000019e90,
 __lock=..., __atime=std::chrono::_V2::steady_clock time_point = { 1052772720473ns }) at /usr/include/c++/14/condition_variable:113
#7 0x00007ffb81b2d2be in std::condition_variable::wait_until<std::chrono::_V2::steady_clock, std::chrono::duration<long, std::ratio<1l, 1000000000l> >, arrow::ConcreteFutureImpl::DoWait(double)::{lambda()#1}>(std::unique_lock<std::mutex>&, std::chrono::time_point<std::chrono::_V2::steady_clock, std::chrono::duration<long, std::ratio<1l, 1000000000l> > > const&, arrow::ConcreteFutureImpl::DoWait(double)::{lambda()#1}) (
 this=0x50e000019e90, __lock=..., __atime=std::chrono::_V2::steady_clock time_point = { 1052772720473ns }, __p=...)
 at /usr/include/c++/14/condition_variable:154
#8 0x00007ffb81b2adc0 in std::condition_variable::wait_for<double, std::ratio<1l, 1l>, arrow::ConcreteFutureImpl::DoWait(double)::{lambda()#1}>(std::unique_lock<std::mutex>&, std::chrono::duration<double, std::ratio<1l, 1l> > const&, arrow::ConcreteFutureImpl::DoWait(double)::{lambda()#1}) (
 this=0x50e000019e90, __lock=..., __rtime=std::chrono::duration = { 64s }, __p=...) at /usr/include/c++/14/condition_variable:177
#9 0x00007ffb81b29af6 in arrow::ConcreteFutureImpl::DoWait (this=0x50e000019e20, seconds=64)
 at /home/bryce/src/apache/arrow/cpp/src/arrow/util/future.cc:179
#10 0x00007ffb81b24c05 in arrow::FutureImpl::Wait (this=0x50e000019e20, seconds=64)
 at /home/bryce/src/apache/arrow/cpp/src/arrow/util/future.cc:222
#11 0x0000561a897f7083 in arrow::Future<arrow::internal::Empty>::Wait (this=0x5140000086b8, seconds=64)
 at /home/bryce/src/apache/arrow/cpp/src/arrow/util/future.h:395
#12 0x0000561a897ee279 in arrow::dataset::internal::DatasetWriterTestFixture::EndWriterChecked (this=0x514000008640, writer=0x50200000a610)
 at /home/bryce/src/apache/arrow/cpp/src/arrow/dataset/dataset_writer_test.cc:116
#13 0x0000561a897ca492 in arrow::dataset::internal::DatasetWriterTestFixture_MaxRowsOneWriteBackpresure_Test::TestBody (this=0x514000008640)
 at /home/bryce/src/apache/arrow/cpp/src/arrow/dataset/dataset_writer_test.cc:304
#14 0x00007ffb87a2094e in testing::internal::HandleSehExceptionsInMethodIfSupported<testing::Test, void> (object=0x514000008640,
 method=&virtual testing::Test::TestBody(), location=0x7ffb87a37fdb "the test body")
 at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/src/gtest.cc:2653
#15 0x00007ffb87a18d4b in testing::internal::HandleExceptionsInMethodIfSupported<testing::Test, void> (object=0x514000008640,
 method=&virtual testing::Test::TestBody(), location=0x7ffb87a37fdb "the test body")
 at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/src/gtest.cc:2689
#16 0x00007ffb879f35d3 in testing::Test::Run (this=0x514000008640)
 at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/src/gtest.cc:2728
#17 0x00007ffb879f3fd6 in testing::TestInfo::Run (this=0x512000002d40)
 at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/src/gtest.cc:2874
#18 0x00007ffb879f48c4 in testing::TestSuite::Run (this=0x512000002440)
 at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/src/gtest.cc:3052
#19 0x00007ffb87a04a82 in testing::internal::UnitTestImpl::RunAllTests (this=0x517000000400)
 at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/src/gtest.cc:6004
#20 0x00007ffb87a21a04 in testing::internal::HandleSehExceptionsInMethodIfSupported<testing::internal::UnitTestImpl, bool> (object=0x517000000400,
 method=(bool (testing::internal::UnitTestImpl::*)(class testing::internal::UnitTestImpl * const)) 0x7ffb87a046a6 <testing::internal::UnitTestImpl::RunAllTests()>, location=0x7ffb87a38a80 "auxiliary test code (environments or event listeners)")
 at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/src/gtest.cc:2653
#21 0x00007ffb87a19f33 in testing::internal::HandleExceptionsInMethodIfSupported<testing::internal::UnitTestImpl, bool> (object=0x517000000400,
 method=(bool (testing::internal::UnitTestImpl::*)(class testing::internal::UnitTestImpl * const)) 0x7ffb87a046a6 <testing::internal::UnitTestImpl::RunAllTests()>, location=0x7ffb87a38a80 "auxiliary test code (environments or event listeners)")
 at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/src/gtest.cc:2689
#22 0x00007ffb87a0320e in testing::UnitTest::Run (this=0x7ffb87a66e00 <testing::UnitTest::GetInstance()::instance>)
 at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/src/gtest.cc:5583
#23 0x00007ffb87abe1a2 in RUN_ALL_TESTS () at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/include/gtest/gtest.h:2334
#24 0x00007ffb87abe18e in main (argc=1, argv=0x7fff22e64898)
 at /home/bryce/src/apache/arrow/cpp/build/_deps/googletest-src/googletest/src/gtest_main.cc:64

I'll keep trying to reproduce that now.

[amoeba]

I didn't reproduce whatever the hang was I reported above so maybe that was just a fluke.

[pitrou]

I did manage to get a stack trace,

It doesn't seem to be from the right thread, though? Good practice is to use thread apply all bt to get a trace of all threads at the point of crash.

Edit: ah, sorry, it was a hang, not a crash. But regardless, without the stack trace from the other threads, it's not possible to understand what could have happened.

[amoeba]

Yeah you're right about needing to dump all threads. I'm not in good practice with this stuff apparently. :)

I'll leave it running for a while today to see if I can reproduce the hang.

[amoeba]

After about 5 minutes, the test hang again. I then,

• attached gdb to the test process
• ran thread apply all bt (output included in gist)
• when I detached gdb, the test then printed a bunch of output, see gist. kinda odd

[pitrou]

Did you try to run the test for as long on the main branch?