This is an automated email from the ASF dual-hosted git repository.
pitrou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/main by this push:
new 157575a986 GH-48137: [C++] Restore ThreadPool state when a worker
fails to start (#51107)
157575a986 is described below
commit 157575a986ccdde19dfa6888d2ea25a169a3b3de
Author: Advit Arora <[email protected]>
AuthorDate: Thu Sep 3 16:06:43 2026 +0530
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 <[email protected]>
Signed-off-by: Antoine Pitrou <[email protected]>
---
cpp/src/arrow/util/thread_pool.cc | 25 ++++++++++++++++---------
cpp/src/arrow/util/thread_pool.h | 3 ++-
cpp/src/arrow/util/thread_pool_test.cc | 31 +++++++++++++++++++++++++++++++
3 files changed, 49 insertions(+), 10 deletions(-)
diff --git a/cpp/src/arrow/util/thread_pool.cc
b/cpp/src/arrow/util/thread_pool.cc
index 4fbce97c2f..00e09b6573 100644
--- a/cpp/src/arrow/util/thread_pool.cc
+++ b/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,
diff --git a/cpp/src/arrow/util/thread_pool.h b/cpp/src/arrow/util/thread_pool.h
index ce33c4c201..582505cda9 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(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();
diff --git a/cpp/src/arrow/util/thread_pool_test.cc
b/cpp/src/arrow/util/thread_pool_test.cc
index c1391c8be8..7c7498838a 100644
--- a/cpp/src/arrow/util/thread_pool_test.cc
+++ b/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) {
