GH-49272: [C++][CI] Fix intermittent segfault in arrow-json-test with MinGW

cpp/src/arrow/json/reader_test.cc

@@ -290,6 +290,28 @@ TEST(ReaderTest, MultipleChunksParallel) {
  AssertTablesEqual(*serial, *threaded);
 }
 
+// Regression test for intermittent threading crashes on MinGW.
+// Run this test multiple times manually to stress-test:
+// while build/debug/arrow-json-test
+// --gtest_filter=ReaderTest.MultipleChunksParallelRegression; do :; done
+// See https://github.com/apache/arrow/issues/49272
+TEST(ReaderTest, MultipleChunksParallelRegression) {
+ int64_t count = 1 << 10;
+ ReadOptions read_options;
+ read_options.block_size = static_cast<int>(count / 2);
+ read_options.use_threads = true;
+ ParseOptions parse_options;
+
+ std::string json;
+ for (int64_t i = 0; i < count; ++i) {
+ json += "{\"a\":" + std::to_string(i) + "}\n";
+ }
+
+ ASSERT_OK_AND_ASSIGN(auto table,
+ ReadToTable(std::move(json), read_options, parse_options));
+ ASSERT_EQ(table->num_rows(), count);
+}
+
 TEST(ReaderTest, ListArrayWithFewValues) {
  // ARROW-7647
  ParseOptions parse_options;

cpp/src/arrow/util/thread_pool.cc

@@ -31,6 +31,7 @@
 #include "arrow/util/io_util.h"
 #include "arrow/util/logging_internal.h"
 #include "arrow/util/mutex.h"
+#include "arrow/util/windows_compatibility.h"
 
 #include "arrow/util/tracing_internal.h"
 
@@ -630,9 +631,66 @@ void ThreadPool::CollectFinishedWorkersUnlocked() {
  state_->finished_workers_.clear();
 }
 
+// MinGW's __emutls implementation for C++ thread_local has known race conditions
+// during thread creation. When a new worker thread is spawned and immediately
+// writes to a thread_local variable (here: current_thread_pool_), __emutls may
+// not have finished initializing TLS for that thread, causing it to dereference
+// a stale/invalid pointer and segfault. This is a known upstream GCC/MinGW bug:
+// https://gcc.gnu.org/bugzilla/show_bug.cgi?id=78605
+// The crash surfaces specifically in arrow-json-test (ReaderTest.MultipleChunksParallel)
+// because that test creates a fresh ThreadPool and immediately dispatches work,
+// hitting the narrow startup race before __emutls can initialize. Other tests
+// that rely on the already-warmed global thread pool do not trigger this window.
+// Use native Win32 TLS (TlsAlloc/TlsGetValue/TlsSetValue) to bypass __emutls.
+// See also: https://github.com/apache/arrow/issues/49272
+# ifdef __MINGW32__
+
+namespace {
+DWORD GetPoolTlsIndex() {
+ static DWORD index = [] {
+ DWORD i = TlsAlloc();
+ if (i == TLS_OUT_OF_INDEXES) {
+ ARROW_LOG(FATAL) << "TlsAlloc failed for thread pool TLS: "
+ << WinErrorMessage(GetLastError());
+ }
+ return i;
+ }();
+ return index;
+}
+} // namespace
+
+static ThreadPool* GetCurrentThreadPool() {
+ // Preserve the caller's last-error value while also detecting TLS failures.
+ DWORD original_error = GetLastError();
+ // Ensure a successful TlsGetValue() leaves GetLastError() == 0.
+ SetLastError(0);
+ auto* pool = static_cast<ThreadPool*>(TlsGetValue(GetPoolTlsIndex()));
+ DWORD tls_error = GetLastError();
+ if (tls_error != 0) {
+ // No need to restore original_error here: ARROW_LOG(FATAL) aborts the process.
+ ARROW_LOG(FATAL) << "TlsGetValue failed for thread pool TLS: "
+ << WinErrorMessage(tls_error);
+ }
+ // Restore the caller's last-error value.
+ SetLastError(original_error);
+ return pool;
+}
+
+static void SetCurrentThreadPool(ThreadPool* pool) {
+ BOOL ok = TlsSetValue(GetPoolTlsIndex(), pool);
+ if (!ok) {
+ ARROW_LOG(FATAL) << "TlsSetValue failed for thread pool TLS: "
+ << WinErrorMessage(GetLastError());
+ }
+}
+# else
 thread_local ThreadPool* current_thread_pool_ = nullptr;
 
-bool ThreadPool::OwnsThisThread() { return current_thread_pool_ == this; }
+static ThreadPool* GetCurrentThreadPool() { return current_thread_pool_; }
+static void SetCurrentThreadPool(ThreadPool* pool) { current_thread_pool_ = pool; }
+# endif
+
+bool ThreadPool::OwnsThisThread() { return GetCurrentThreadPool() == this; }
 
 void ThreadPool::LaunchWorkersUnlocked(int threads) {
  std::shared_ptr<State> state = sp_state_;
@@ -641,7 +699,7 @@ void ThreadPool::LaunchWorkersUnlocked(int threads) {
  state_->workers_.emplace_back();
  auto it = --(state_->workers_.end());
  *it = std::thread([this, state, it] {
-current_thread_pool_ = this;
+SetCurrentThreadPool(this);
  WorkerLoop(state, it);
  });
  }

cpp/src/arrow/util/thread_pool_test.cc

@@ -44,6 +44,7 @@
 #include "arrow/util/macros.h"
 #include "arrow/util/test_common.h"
 #include "arrow/util/thread_pool.h"
+#include "arrow/util/windows_compatibility.h"
 
 namespace arrow {
 namespace internal {
@@ -660,6 +661,34 @@ TEST_F(TestThreadPool, OwnsCurrentThread) {
  ASSERT_FALSE(one_failed);
 }
 
+#ifdef _WIN32
+TEST_F(TestThreadPool, OwnsThisThreadPreservesLastError) {
+# ifndef ARROW_ENABLE_THREADING
+ GTEST_SKIP() << "Test requires threading support";
+# endif
+ auto pool = this->MakeThreadPool(4);
+
+ // Verify from outside the pool: OwnsThisThread() must not clobber
+ // the calling thread's last-error state.
+ ::SetLastError(ERROR_FILE_NOT_FOUND);
+ ASSERT_FALSE(pool->OwnsThisThread());
+ ASSERT_EQ(::GetLastError(), static_cast<DWORD>(ERROR_FILE_NOT_FOUND));
+
+ // Verify from inside a pool thread.
+ std::atomic<bool> error_preserved{true};
+ ASSERT_OK(pool->Spawn([&] {
+ ::SetLastError(ERROR_ACCESS_DENIED);
+ ASSERT_TRUE(pool->OwnsThisThread());
+ if (::GetLastError() != ERROR_ACCESS_DENIED) {
+ error_preserved = false;
+ }
+ }));
+
+ ASSERT_OK(pool->Shutdown());
+ ASSERT_TRUE(error_preserved.load());
+}
+#endif
+
 TEST_F(TestThreadPool, StressSpawnThreaded) {
 #ifndef ARROW_ENABLE_THREADING
  GTEST_SKIP() << "Test requires threading support";