Skip to content

Commit ea6bbf5

Browse files
committed
GH-49433: [C++] Buffer ARROW_LOG output to prevent thread interleaving
When using the ARROW_LOG macros from multiple threads, messages were getting mingled together in stderr because the operator<< was writing directly to the global stream piece-by-piece. This commit introduces an internal std::ostringstream buffer to CerrLog so that messages are accumulated locally and flushed atomically upon destruction.
1 parent 008e082 commit ea6bbf5

2 files changed

Lines changed: 31 additions & 3 deletions

File tree

cpp/src/arrow/util/logging.cc

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@
2424
#endif
2525
#include <cstdlib>
2626
#include <iostream>
27+
#include <mutex>
28+
#include <sstream>
2729

2830
#ifdef ARROW_USE_GLOG
2931

@@ -67,7 +69,9 @@ class CerrLog {
6769

6870
virtual ~CerrLog() {
6971
if (has_logged_) {
70-
std::cerr << std::endl;
72+
static std::mutex cerr_mutex;
73+
std::lock_guard<std::mutex> lock(cerr_mutex);
74+
std::cerr << buffer_.str() << std::endl;
7175
}
7276
if (severity_ == ArrowLogLevel::ARROW_FATAL) {
7377
PrintBackTrace();
@@ -77,21 +81,22 @@ class CerrLog {
7781

7882
std::ostream& Stream() {
7983
has_logged_ = true;
80-
return std::cerr;
84+
return buffer_;
8185
}
8286

8387
template <class T>
8488
CerrLog& operator<<(const T& t) {
8589
if (severity_ != ArrowLogLevel::ARROW_DEBUG) {
8690
has_logged_ = true;
87-
std::cerr << t;
91+
buffer_ << t;
8892
}
8993
return *this;
9094
}
9195

9296
protected:
9397
const ArrowLogLevel severity_;
9498
bool has_logged_;
99+
std::ostringstream buffer_;
95100

96101
void PrintBackTrace() {
97102
#ifdef ARROW_WITH_BACKTRACE

cpp/src/arrow/util/logging_test.cc

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818
#include <chrono>
1919
#include <cstdint>
2020
#include <iostream>
21+
#include <thread>
22+
#include <vector>
2123

2224
#include <gtest/gtest.h>
2325

@@ -95,6 +97,27 @@ TEST(ArrowCheck, PayloadEvaluatedOnFailure) {
9597
ASSERT_TRUE(tracer.was_printed);
9698
}
9799

100+
TEST(ArrowLog, MultiThreadedLogging) {
101+
constexpr int kNumThreads = 10;
102+
constexpr int kNumMessges = 10;
103+
std::vector<std::thread> threads;
104+
105+
threads.reserve(kNumThreads);
106+
107+
for (int i = 0; i < kNumThreads; ++i) {
108+
threads.emplace_back([i]() {
109+
for (int j = 0; j < kNumMessges; ++j) {
110+
ARROW_LOG(INFO) << "Thread" << i << " message " << j
111+
<< " - testing thread safety.";
112+
}
113+
});
114+
}
115+
116+
for (auto& thread : threads) {
117+
thread.join();
118+
}
119+
}
120+
98121
} // namespace util
99122

100123
TEST(DcheckMacros, DoNotEvaluateReleaseMode) {

0 commit comments

Comments
 (0)