GH-48137: [C++] Restore ThreadPool state when a worker fails to start (#51107)

### Rationale for this change

`LaunchWorkersUnlocked` appends an entry to `state_->workers_` before constructing the thread
that owns it, and only the worker itself erases that entry. If the `std::thread` constructor
fails, the entry stays behind with nothing left to remove it, so `Shutdown` waits forever on
`workers_.empty()`. The destructor takes the same path. The failure also escaped `SpawnReal`
after `tasks_queued_or_running_` had been incremented, so `WaitForIdle` never returned either,
and once stale entries filled `workers_` to capacity the pool stopped launching workers while
`Spawn` still returned OK for tasks nothing would run.

### What changes are included in this PR?

`LaunchWorkersUnlocked` returns a `Status`. A failed thread construction erases the entry it had
reserved and returns an error, which `SpawnReal` and `SetCapacity` propagate. The task counter is
incremented after the launch rather than before, so a failed launch cannot leak a count.

### Are these changes tested?

`TestThreadPool.FailedWorkerLaunch` lowers `RLIMIT_NPROC` to 1, spawns a task, restores the soft
limit, and then checks the pool reports no workers and no tasks and still shuts down. It skips on
macOS, where `RLIMIT_NPROC` counts processes rather than threads, and skips anywhere else the
lowered limit does not stop thread creation, such as under root.

### Are there any user-facing changes?

Yes. `Spawn`, `Submit` and `SetCapacity` used to let a `std::system_error` escape when the OS
refused a new thread. They return an error `Status` now. `ThreadPool::Make` is unaffected, since
worker threads are only started on demand and a new pool starts none.

* GitHub Issue: #48137

Authored-by: Advit Arora <advitarora2@gmail.com>
Signed-off-by: Antoine Pitrou <antoine@python.org>

Author: advitrocks9

cpp/src/arrow/util/thread_pool.cc

@@ -23,6 +23,7 @@
 #include <list>
 #include <mutex>
 #include <string>
+#include <system_error>
 #include <thread>
 #include <vector>
 
@@ -580,7 +581,7 @@ Status ThreadPool::SetCapacity(int threads) {
                                 threads - static_cast<int>(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,17 +693,23 @@ 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> state = sp_state_;
 
   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 (const std::exception& e) {
+      state_->workers_.erase(it);
+      return Status::UnknownError("Failed to launch worker thread: ", e.what());
+    }
   }
+  return Status::OK();
 }
 
 Status ThreadPool::SpawnReal(TaskHints hints, FnOnce<void()> task, StopToken stop_token,
@@ -729,12 +736,12 @@ Status ThreadPool::SpawnReal(TaskHints hints, FnOnce<void()> task, StopToken sto
       return Status::Invalid("operation forbidden during or after shutdown");
     }
     CollectFinishedWorkersUnlocked();
-    state_->tasks_queued_or_running_++;
-    if (static_cast<int>(state_->workers_.size()) < state_->tasks_queued_or_running_ &&
+    if (static_cast<int>(state_->workers_.size()) <= state_->tasks_queued_or_running_ &&
         state_->desired_capacity_ > static_cast<int>(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(
         QueuedTask{{std::move(task), std::move(stop_token), std::move(stop_callback)},
                    hints.priority,

cpp/src/arrow/util/thread_pool.h

@@ -496,6 +496,7 @@ class ARROW_EXPORT ThreadPool : public Executor {
 
  protected:
   FRIEND_TEST(TestThreadPool, SetCapacity);
+  FRIEND_TEST(TestThreadPool, FailedWorkerLaunch);
   FRIEND_TEST(TestGlobalThreadPool, Capacity);
   ARROW_FRIEND_EXPORT friend ThreadPool* GetCpuThreadPool();
 
@@ -507,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();
 

cpp/src/arrow/util/thread_pool_test.cc

@@ -16,6 +16,7 @@
 // under the License.
 
 #ifndef _WIN32
+#  include <sys/resource.h>
 #  include <sys/types.h>
 #  include <unistd.h>
 #endif
@@ -832,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) {