apache / apache/arrow

[C++] IpcWriter to memory stream race condition

Open
#40,419 0 comments 0 reactions 0 assignees View on GitHub
Component: C++ Type: bug
Dominant language
C++
Stars
17.1k
Forks
4.3k
Avg merge
3d 13h
Merged PRs (30d)
88

Description

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

Arrow version: 15.0.1
Platform: x64-windows

My Acero use case:

dataset `scan` node
\\/
filter/projection (not relevant)
\\/
`consuming_sink`
\\/
Custom `SinkNodeConsumer` writting batches to `RecordBatchWriter`
`RecordBatchWriter` is an `ipc::MakeStreamWriter` with `BufferOutputStream`

This use case creates multiple threads writting to `BufferOutputStream` which causes race in `BufferOutputStream::Write `

SinkNodeConsumer:

```
struct RecordBatchWriterNodeConsumer : public ac::SinkNodeConsumer {
RecordBatchWriterNodeConsumer(std::shared_ptr _writer)
: writer(_writer){}
arrow::Status Init(const std::shared_ptr& _schema,
ac::BackpressureControl* backpressure_control,
ac::ExecPlan* plan) override {
schema = _schema;
return arrow::Status::OK();
}

arrow::Status Consume(cp::ExecBatch batch) override {
ARROW_ASSIGN_OR_RAISE(auto rb, batch.ToRecordBatch(schema));
//const std::lock_guard lock(consume_mutex);
return writer->WriteRecordBatch(*rb);
}

arrow::Future<> Finish() override {
return writer->Close();
}

std::shared_ptr writer;
std::shared_ptr schema;
//std::mutex consume_mutex;
};
```

AddressSanitizer dump:

```
=================================================================
==22728==ERROR: AddressSanitizer: attempting double-free on 0x038f001c4800 in thread T48:
#0 0x7ffc9ae00798 in _asan_wrap_RtlValidateHeap+0x288 (\\?\C:\Users\USER\project\dist_MXP_web\libs\clang_rt.asan_dynamic-x86_64.dll+0x180040798)
#1 0x7ffc99f31a7b in arrow::BaseMemoryPoolImpl::Reallocate C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\memory_pool.cc:487
#2 0x7ffc99f321e7 in arrow::PoolBuffer::Resize C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\memory_pool.cc:891
#3 0x7ffc99ff2db2 in arrow::io::BufferOutputStream::Reserve C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\io\memory.cc:123
#4 0x7ffc99ff3544 in arrow::io::BufferOutputStream::Write C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\io\memory.cc:105
#5 0x7ffc99ff0770 in arrow::io::Writable::Write C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\io\interfaces.cc:203
#6 0x7ffc9a9d7f93 in arrow::ipc::WriteIpcPayload C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:757
#7 0x7ffc9a9d82ab in arrow::ipc::internal::PayloadStreamWriter::WritePayload C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:1406
#8 0x7ffc9a9d8756 in arrow::ipc::internal::IpcFormatWriter::WriteRecordBatch C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:1195
#9 0x7ffc9a9d8375 in arrow::ipc::internal::IpcFormatWriter::WriteRecordBatch C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:1176
#10 0x7ffcadac9af5 in RecordBatchWriterNodeConsumer::Consume C:\Users\USER\project\src\mxp\electron\cmake\src\analysis\PdwProcessing.cpp:358
#11 0x7ffcac3365e5 in arrow::acero::`anonymous namespace'::ConsumingSinkNode::Process C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\sink_node.cc:399
#12 0x7ffcac334d67 in arrow::acero::`anonymous namespace'::ConsumingSinkNode::InputReceived C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\sink_node.cc:390
#13 0x7ffcac32851a in arrow::acero::MapNode::InputReceived C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\map_node.cc:79
#14 0x7ffcac344069 in ::operator() C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\source_node.cc:157
#15 0x7ffcac349121 in std::_Func_impl_no_alloc<,arrow::Status>::_Do_call C:\Program Files (x86)\Microsoft Visual Studio\2019\Community\VC\Tools\MSVC\14.29.30133\include\functional:822
#16 0x7ffcac331b8f in std::_Call_binder,std::function >,std::tuple<> > C:\Program Files (x86)\Microsoft Visual Studio\2019\Community\VC\Tools\MSVC\14.29.30133\include\functional:1307
#17 0x7ffcac33309d in arrow::internal::FnOnce::FnImpl &,std::function > >::invoke C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\util\functional.h:152
#18 0x7ffc9a056b64 in std::thread::_Invoke >,0> C:\Program Files (x86)\Microsoft Visual Studio\2019\Community\VC\Tools\MSVC\14.29.30133\include\thread:55
#19 0x7ffd30571bb1 in configthreadlocale+0x91 (C:\WINDOWS\System32\ucrtbase.dll+0x180021bb1)
#20 0x7ffc9ae0ebde in _asan_default_suppressions__dll+0x122e (\\?\C:\Users\USER\project\dist_MXP_web\libs\clang_rt.asan_dynamic-x86_64.dll+0x18004ebde)
#21 0x7ffd32227343 in BaseThreadInitThunk+0x13 (C:\WINDOWS\System32\KERNEL32.DLL+0x180017343)
#22 0x7ffd328226b0 in RtlUserThreadStart+0x20 (C:\WINDOWS\SYSTEM32\ntdll.dll+0x1800526b0)

0x038f001c4800 is located 0 bytes inside of 65536-byte region [0x038f001c4800,0x038f001d4800)
freed by thread T52 here:
#0 0x7ffc9ae00798 in _asan_wrap_RtlValidateHeap+0x288 (\\?\C:\Users\USER\project\dist_MXP_web\libs\clang_rt.asan_dynamic-x86_64.dll+0x180040798)
#1 0x7ffc99f31a7b in arrow::BaseMemoryPoolImpl::Reallocate C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\memory_pool.cc:487
#2 0x7ffc99f321e7 in arrow::PoolBuffer::Resize C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\memory_pool.cc:891
#3 0x7ffc99ff2db2 in arrow::io::BufferOutputStream::Reserve C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\io\memory.cc:123
#4 0x7ffc99ff3544 in arrow::io::BufferOutputStream::Write C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\io\memory.cc:105
#5 0x7ffc99ff0770 in arrow::io::Writable::Write C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\io\interfaces.cc:203
#6 0x7ffc9a9d7f93 in arrow::ipc::WriteIpcPayload C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:757
#7 0x7ffc9a9d82ab in arrow::ipc::internal::PayloadStreamWriter::WritePayload C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:1406
#8 0x7ffc9a9d8756 in arrow::ipc::internal::IpcFormatWriter::WriteRecordBatch C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:1195
#9 0x7ffc9a9d8375 in arrow::ipc::internal::IpcFormatWriter::WriteRecordBatch C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:1176
#10 0x7ffcadac9af5 in RecordBatchWriterNodeConsumer::Consume C:\Users\USER\project\src\mxp\electron\cmake\src\analysis\PdwProcessing.cpp:358
#11 0x7ffcac3365e5 in arrow::acero::`anonymous namespace'::ConsumingSinkNode::Process C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\sink_node.cc:399
#12 0x7ffcac334d67 in arrow::acero::`anonymous namespace'::ConsumingSinkNode::InputReceived C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\sink_node.cc:390
#13 0x7ffcac32851a in arrow::acero::MapNode::InputReceived C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\map_node.cc:79
#14 0x7ffcac344069 in ::operator() C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\source_node.cc:157
#15 0x7ffcac349121 in std::_Func_impl_no_alloc<,arrow::Status>::_Do_call C:\Program Files (x86)\Microsoft Visual Studio\2019\Community\VC\Tools\MSVC\14.29.30133\include\functional:822
#16 0x7ffcac331b8f in std::_Call_binder,std::function >,std::tuple<> > C:\Program Files (x86)\Microsoft Visual Studio\2019\Community\VC\Tools\MSVC\14.29.30133\include\functional:1307
#17 0x7ffcac33309d in arrow::internal::FnOnce::FnImpl &,std::function > >::invoke C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\util\functional.h:152
#18 0x7ffc9a056b64 in std::thread::_Invoke >,0> C:\Program Files (x86)\Microsoft Visual Studio\2019\Community\VC\Tools\MSVC\14.29.30133\include\thread:55
#19 0x7ffd30571bb1 in configthreadlocale+0x91 (C:\WINDOWS\System32\ucrtbase.dll+0x180021bb1)
#20 0x7ffc9ae0ebde in _asan_default_suppressions__dll+0x122e (\\?\C:\Users\USER\project\dist_MXP_web\libs\clang_rt.asan_dynamic-x86_64.dll+0x18004ebde)
#21 0x7ffd32227343 in BaseThreadInitThunk+0x13 (C:\WINDOWS\System32\KERNEL32.DLL+0x180017343)
#22 0x7ffd328226b0 in RtlUserThreadStart+0x20 (C:\WINDOWS\SYSTEM32\ntdll.dll+0x1800526b0)

previously allocated by thread T52 here:
#0 0x7ffc9ae00a1a in _asan_wrap_RtlValidateHeap+0x50a (\\?\C:\Users\USER\project\dist_MXP_web\libs\clang_rt.asan_dynamic-x86_64.dll+0x180040a1a)
#1 0x7ffc99f3197c in arrow::BaseMemoryPoolImpl::Reallocate C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\memory_pool.cc:487
#2 0x7ffc99f321e7 in arrow::PoolBuffer::Resize C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\memory_pool.cc:891
#3 0x7ffc99ff2db2 in arrow::io::BufferOutputStream::Reserve C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\io\memory.cc:123
#4 0x7ffc99ff3544 in arrow::io::BufferOutputStream::Write C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\io\memory.cc:105
#5 0x7ffc99ff0770 in arrow::io::Writable::Write C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\io\interfaces.cc:203
#6 0x7ffc9a9d7f93 in arrow::ipc::WriteIpcPayload C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:757
#7 0x7ffc9a9d82ab in arrow::ipc::internal::PayloadStreamWriter::WritePayload C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:1406
#8 0x7ffc9a9d8756 in arrow::ipc::internal::IpcFormatWriter::WriteRecordBatch C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:1195
#9 0x7ffc9a9d8375 in arrow::ipc::internal::IpcFormatWriter::WriteRecordBatch C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\ipc\writer.cc:1176
#10 0x7ffcadac9af5 in RecordBatchWriterNodeConsumer::Consume C:\Users\USER\project\src\mxp\electron\cmake\src\analysis\PdwProcessing.cpp:358
#11 0x7ffcac3365e5 in arrow::acero::`anonymous namespace'::ConsumingSinkNode::Process C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\sink_node.cc:399
#12 0x7ffcac334d67 in arrow::acero::`anonymous namespace'::ConsumingSinkNode::InputReceived C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\sink_node.cc:390
#13 0x7ffcac32851a in arrow::acero::MapNode::InputReceived C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\map_node.cc:79
#14 0x7ffcac344069 in ::operator() C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\acero\source_node.cc:157
#15 0x7ffcac349121 in std::_Func_impl_no_alloc<,arrow::Status>::_Do_call C:\Program Files (x86)\Microsoft Visual Studio\2019\Community\VC\Tools\MSVC\14.29.30133\include\functional:822
#16 0x7ffcac331b8f in std::_Call_binder,std::function >,std::tuple<> > C:\Program Files (x86)\Microsoft Visual Studio\2019\Community\VC\Tools\MSVC\14.29.30133\include\functional:1307
#17 0x7ffcac33309d in arrow::internal::FnOnce::FnImpl &,std::function > >::invoke C:\Users\USER\project\vcpkg\buildtrees\arrow\src\e-arrow-15-f337ad0391.clean\cpp\src\arrow\util\functional.h:152
#18 0x7ffc9a056b64 in std::thread::_Invoke >,0> C:\Program Files (x86)\Microsoft Visual Studio\2019\Community\VC\Tools\MSVC\14.29.30133\include\thread:55
#19 0x7ffd30571bb1 in configthreadlocale+0x91 (C:\WINDOWS\System32\ucrtbase.dll+0x180021bb1)
#20 0x7ffc9ae0ebde in _asan_default_suppressions__dll+0x122e (\\?\C:\Users\USER\project\dist_MXP_web\libs\clang_rt.asan_dynamic-x86_64.dll+0x18004ebde)
#21 0x7ffd32227343 in BaseThreadInitThunk+0x13 (C:\WINDOWS\System32\KERNEL32.DLL+0x180017343)
#22 0x7ffd328226b0 in RtlUserThreadStart+0x20 (C:\WINDOWS\SYSTEM32\ntdll.dll+0x1800526b0)
```

This can be fixed by adding lock in RecordBatchWriterNodeConsumer::Consume, however this not seem like a good solution. I think either:
arrow::io::BufferOutputStream::Write should have write lock
or:
StreamWriter should have a write lock (as it writes to stream)
or maybe both?

### Component(s)

C++

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.