From 96ac49d5cb6b5656246e0f92f3b17ab05e213d1c Mon Sep 17 00:00:00 2001 From: Advit Arora Date: Tue, 25 Aug 2026 11:38:49 +0530 Subject: [PATCH 1/2] GH-48137: [C++] Restore ThreadPool state when a worker fails to start --- cpp/src/arrow/util/thread_pool.cc | 17 ++-- cpp/src/arrow/util/thread_pool.h | 1 + cpp/src/arrow/util/thread_pool_test.cc | 105 +++++++++++++++++++++++++ 3 files changed, 117 insertions(+), 6 deletions(-) diff --git a/cpp/src/arrow/util/thread_pool.cc b/cpp/src/arrow/util/thread_pool.cc index 4fbce97c2ff4..ddfd67078b57 100644 --- a/cpp/src/arrow/util/thread_pool.cc +++ b/cpp/src/arrow/util/thread_pool.cc @@ -698,10 +698,15 @@ void ThreadPool::LaunchWorkersUnlocked(int threads) { for (int i = 0; i < threads; i++) { state_->workers_.emplace_back(); auto it = --(state_->workers_.end()); - *it = std::thread([this, state, it] { - SetCurrentThreadPool(this); - WorkerLoop(state, it); - }); + try { + *it = std::thread([this, state, it] { + SetCurrentThreadPool(this); + WorkerLoop(state, it); + }); + } catch (...) { + state_->workers_.erase(it); + throw; + } } } @@ -729,12 +734,12 @@ Status ThreadPool::SpawnReal(TaskHints hints, FnOnce task, StopToken sto return Status::Invalid("operation forbidden during or after shutdown"); } CollectFinishedWorkersUnlocked(); - state_->tasks_queued_or_running_++; - if (static_cast(state_->workers_.size()) < state_->tasks_queued_or_running_ && + if (static_cast(state_->workers_.size()) <= state_->tasks_queued_or_running_ && state_->desired_capacity_ > static_cast(state_->workers_.size())) { // We can still spin up more workers so spin up a new worker LaunchWorkersUnlocked(/*threads=*/1); } + state_->tasks_queued_or_running_++; state_->pending_tasks_.push( QueuedTask{{std::move(task), std::move(stop_token), std::move(stop_callback)}, hints.priority, diff --git a/cpp/src/arrow/util/thread_pool.h b/cpp/src/arrow/util/thread_pool.h index 201b8cef790d..1015ca9039fd 100644 --- a/cpp/src/arrow/util/thread_pool.h +++ b/cpp/src/arrow/util/thread_pool.h @@ -496,6 +496,7 @@ class ARROW_EXPORT ThreadPool : public Executor { protected: FRIEND_TEST(TestThreadPool, SetCapacity); + FRIEND_TEST(TestThreadPoolForkSafety, FailedWorkerLaunch); FRIEND_TEST(TestGlobalThreadPool, Capacity); ARROW_FRIEND_EXPORT friend ThreadPool* GetCpuThreadPool(); diff --git a/cpp/src/arrow/util/thread_pool_test.cc b/cpp/src/arrow/util/thread_pool_test.cc index c1391c8be883..455c4dc5a36a 100644 --- a/cpp/src/arrow/util/thread_pool_test.cc +++ b/cpp/src/arrow/util/thread_pool_test.cc @@ -16,10 +16,17 @@ // under the License. #ifndef _WIN32 +# include # include +# include # include #endif +#ifdef __APPLE__ +# include +# include +#endif + #include #include #include @@ -28,6 +35,7 @@ #include #include #include +#include #include #include @@ -1049,6 +1057,103 @@ TEST_F(TestThreadPoolForkSafety, NestedChild) { } } +namespace { + +constexpr int kChildOk = 0; +constexpr int kLaunchDidNotFail = 1; +constexpr int kStateNotRestored = 2; +constexpr int kShutdownFailed = 3; +constexpr int kCouldNotForceFailure = 42; + +# ifdef __APPLE__ + +void* ParkUntilExit(void*) { + SleepFor(3600); + return nullptr; +} + +bool ExhaustTaskThreads() { + int max_threads = 0; + size_t size = sizeof(max_threads); + if (sysctlbyname("kern.num_taskthreads", &max_threads, &size, nullptr, 0) != 0) { + return false; + } + pthread_attr_t attr; + if (pthread_attr_init(&attr) != 0) { + return false; + } + bool creation_failed = false; + if (pthread_attr_setstacksize(&attr, 32 * 1024) == 0 && + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) == 0) { + for (int i = 0; i < max_threads && !creation_failed; ++i) { + pthread_t thread; + creation_failed = pthread_create(&thread, &attr, ParkUntilExit, nullptr) != 0; + } + } + pthread_attr_destroy(&attr); + return creation_failed; +} + +# endif + +bool ForceThreadCreationFailure() { + if (geteuid() == 0) { + return false; + } + struct rlimit limit; + if (getrlimit(RLIMIT_NPROC, &limit) != 0) { + return false; + } + limit.rlim_cur = 1; + if (setrlimit(RLIMIT_NPROC, &limit) != 0) { + return false; + } + try { + std::thread([] { SleepFor(3600); }).detach(); + } catch (const std::system_error&) { + return true; + } +# ifdef __APPLE__ + return ExhaustTaskThreads(); +# else + return false; +# endif +} + +} // namespace + +TEST_F(TestThreadPoolForkSafety, FailedWorkerLaunch) { +# ifndef ARROW_ENABLE_THREADING + GTEST_SKIP() << "Test requires threading support"; +# endif + auto child_pid = fork(); + if (child_pid == 0) { + alarm(60); + auto pool = this->MakeThreadPool(4); + if (!ForceThreadCreationFailure()) { + std::exit(kCouldNotForceFailure); + } + try { + ARROW_UNUSED(pool->Spawn([] {})); + std::exit(kLaunchDidNotFail); + } catch (const std::system_error&) { + } + if (pool->GetActualCapacity() != 0 || pool->GetNumTasks() != 0) { + std::exit(kStateNotRestored); + } + std::exit(pool->Shutdown().ok() ? kChildOk : kShutdownFailed); + } + + ASSERT_GT(child_pid, 0); + int child_status = 0; + ASSERT_EQ(waitpid(child_pid, &child_status, 0), child_pid); + ASSERT_TRUE(WIFEXITED(child_status)) << "Child status = " << child_status; + if (WEXITSTATUS(child_status) == kCouldNotForceFailure) { + GTEST_SKIP() << "Could not make thread creation fail"; + } + ASSERT_EQ(WEXITSTATUS(child_status), kChildOk); +} + #endif TEST(TestGlobalThreadPool, Capacity) { From 0fa803ac514188f7b6d616f9847620166d506040 Mon Sep 17 00:00:00 2001 From: Advit Arora Date: Thu, 3 Sep 2026 01:25:14 +0530 Subject: [PATCH 2/2] GH-48137: [C++] Return a Status when a worker thread fails to start --- cpp/src/arrow/util/thread_pool.cc | 12 ++- cpp/src/arrow/util/thread_pool.h | 4 +- cpp/src/arrow/util/thread_pool_test.cc | 134 ++++++------------------- 3 files changed, 39 insertions(+), 111 deletions(-) diff --git a/cpp/src/arrow/util/thread_pool.cc b/cpp/src/arrow/util/thread_pool.cc index ddfd67078b57..00e09b6573fc 100644 --- a/cpp/src/arrow/util/thread_pool.cc +++ b/cpp/src/arrow/util/thread_pool.cc @@ -23,6 +23,7 @@ #include #include #include +#include #include #include @@ -580,7 +581,7 @@ Status ThreadPool::SetCapacity(int threads) { threads - static_cast(state_->workers_.size())); if (required > 0) { // Some tasks are pending, spawn the number of needed threads immediately - LaunchWorkersUnlocked(required); + RETURN_NOT_OK(LaunchWorkersUnlocked(required)); } else if (required < 0) { // Excess threads are running, wake them so that they stop state_->cv_.notify_all(); @@ -692,7 +693,7 @@ static void SetCurrentThreadPool(ThreadPool* pool) { current_thread_pool_ = pool bool ThreadPool::OwnsThisThread() { return GetCurrentThreadPool() == this; } -void ThreadPool::LaunchWorkersUnlocked(int threads) { +Status ThreadPool::LaunchWorkersUnlocked(int threads) { std::shared_ptr state = sp_state_; for (int i = 0; i < threads; i++) { @@ -703,11 +704,12 @@ void ThreadPool::LaunchWorkersUnlocked(int threads) { SetCurrentThreadPool(this); WorkerLoop(state, it); }); - } catch (...) { + } catch (const std::exception& e) { state_->workers_.erase(it); - throw; + return Status::UnknownError("Failed to launch worker thread: ", e.what()); } } + return Status::OK(); } Status ThreadPool::SpawnReal(TaskHints hints, FnOnce task, StopToken stop_token, @@ -737,7 +739,7 @@ Status ThreadPool::SpawnReal(TaskHints hints, FnOnce task, StopToken sto if (static_cast(state_->workers_.size()) <= state_->tasks_queued_or_running_ && state_->desired_capacity_ > static_cast(state_->workers_.size())) { // We can still spin up more workers so spin up a new worker - LaunchWorkersUnlocked(/*threads=*/1); + RETURN_NOT_OK(LaunchWorkersUnlocked(/*threads=*/1)); } state_->tasks_queued_or_running_++; state_->pending_tasks_.push( diff --git a/cpp/src/arrow/util/thread_pool.h b/cpp/src/arrow/util/thread_pool.h index 1015ca9039fd..b106b577abbb 100644 --- a/cpp/src/arrow/util/thread_pool.h +++ b/cpp/src/arrow/util/thread_pool.h @@ -496,7 +496,7 @@ class ARROW_EXPORT ThreadPool : public Executor { protected: FRIEND_TEST(TestThreadPool, SetCapacity); - FRIEND_TEST(TestThreadPoolForkSafety, FailedWorkerLaunch); + FRIEND_TEST(TestThreadPool, FailedWorkerLaunch); FRIEND_TEST(TestGlobalThreadPool, Capacity); ARROW_FRIEND_EXPORT friend ThreadPool* GetCpuThreadPool(); @@ -508,7 +508,7 @@ class ARROW_EXPORT ThreadPool : public Executor { // Collect finished worker threads, making sure the OS threads have exited void CollectFinishedWorkersUnlocked(); // Launch a given number of additional workers - void LaunchWorkersUnlocked(int threads); + Status LaunchWorkersUnlocked(int threads); // Get the current actual capacity int GetActualCapacity(); diff --git a/cpp/src/arrow/util/thread_pool_test.cc b/cpp/src/arrow/util/thread_pool_test.cc index 455c4dc5a36a..7c7498838a64 100644 --- a/cpp/src/arrow/util/thread_pool_test.cc +++ b/cpp/src/arrow/util/thread_pool_test.cc @@ -18,15 +18,9 @@ #ifndef _WIN32 # include # include -# include # include #endif -#ifdef __APPLE__ -# include -# include -#endif - #include #include #include @@ -35,7 +29,6 @@ #include #include #include -#include #include #include @@ -840,6 +833,36 @@ TEST_F(TestThreadPool, SetCapacity) { ASSERT_EQ(pool->GetCapacity(), 7); } #endif + +#if defined(ARROW_ENABLE_THREADING) && !defined(_WIN32) +TEST_F(TestThreadPool, FailedWorkerLaunch) { +# ifdef __APPLE__ + GTEST_SKIP() << "RLIMIT_NPROC does not limit thread creation on macOS"; +# else + auto pool = this->MakeThreadPool(4); + + struct rlimit limit; + ASSERT_EQ(getrlimit(RLIMIT_NPROC, &limit), 0); + const rlim_t soft_limit = limit.rlim_cur; + limit.rlim_cur = 1; + if (setrlimit(RLIMIT_NPROC, &limit) != 0) { + GTEST_SKIP() << "Could not lower RLIMIT_NPROC"; + } + const Status st = pool->Spawn([] {}); + limit.rlim_cur = soft_limit; + ASSERT_EQ(setrlimit(RLIMIT_NPROC, &limit), 0); + + if (st.ok()) { + GTEST_SKIP() << "Lowering RLIMIT_NPROC did not prevent thread creation"; + } + ASSERT_RAISES(UnknownError, st); + ASSERT_EQ(pool->GetActualCapacity(), 0); + ASSERT_EQ(pool->GetNumTasks(), 0); + ASSERT_OK(pool->Shutdown()); +# endif +} +#endif + // Test Submit() functionality TEST_F(TestThreadPool, Submit) { @@ -1057,103 +1080,6 @@ TEST_F(TestThreadPoolForkSafety, NestedChild) { } } -namespace { - -constexpr int kChildOk = 0; -constexpr int kLaunchDidNotFail = 1; -constexpr int kStateNotRestored = 2; -constexpr int kShutdownFailed = 3; -constexpr int kCouldNotForceFailure = 42; - -# ifdef __APPLE__ - -void* ParkUntilExit(void*) { - SleepFor(3600); - return nullptr; -} - -bool ExhaustTaskThreads() { - int max_threads = 0; - size_t size = sizeof(max_threads); - if (sysctlbyname("kern.num_taskthreads", &max_threads, &size, nullptr, 0) != 0) { - return false; - } - pthread_attr_t attr; - if (pthread_attr_init(&attr) != 0) { - return false; - } - bool creation_failed = false; - if (pthread_attr_setstacksize(&attr, 32 * 1024) == 0 && - pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) == 0) { - for (int i = 0; i < max_threads && !creation_failed; ++i) { - pthread_t thread; - creation_failed = pthread_create(&thread, &attr, ParkUntilExit, nullptr) != 0; - } - } - pthread_attr_destroy(&attr); - return creation_failed; -} - -# endif - -bool ForceThreadCreationFailure() { - if (geteuid() == 0) { - return false; - } - struct rlimit limit; - if (getrlimit(RLIMIT_NPROC, &limit) != 0) { - return false; - } - limit.rlim_cur = 1; - if (setrlimit(RLIMIT_NPROC, &limit) != 0) { - return false; - } - try { - std::thread([] { SleepFor(3600); }).detach(); - } catch (const std::system_error&) { - return true; - } -# ifdef __APPLE__ - return ExhaustTaskThreads(); -# else - return false; -# endif -} - -} // namespace - -TEST_F(TestThreadPoolForkSafety, FailedWorkerLaunch) { -# ifndef ARROW_ENABLE_THREADING - GTEST_SKIP() << "Test requires threading support"; -# endif - auto child_pid = fork(); - if (child_pid == 0) { - alarm(60); - auto pool = this->MakeThreadPool(4); - if (!ForceThreadCreationFailure()) { - std::exit(kCouldNotForceFailure); - } - try { - ARROW_UNUSED(pool->Spawn([] {})); - std::exit(kLaunchDidNotFail); - } catch (const std::system_error&) { - } - if (pool->GetActualCapacity() != 0 || pool->GetNumTasks() != 0) { - std::exit(kStateNotRestored); - } - std::exit(pool->Shutdown().ok() ? kChildOk : kShutdownFailed); - } - - ASSERT_GT(child_pid, 0); - int child_status = 0; - ASSERT_EQ(waitpid(child_pid, &child_status, 0), child_pid); - ASSERT_TRUE(WIFEXITED(child_status)) << "Child status = " << child_status; - if (WEXITSTATUS(child_status) == kCouldNotForceFailure) { - GTEST_SKIP() << "Could not make thread creation fail"; - } - ASSERT_EQ(WEXITSTATUS(child_status), kChildOk); -} - #endif TEST(TestGlobalThreadPool, Capacity) {