GH-48137: [C++] Restore ThreadPool state when a worker fails to start - #51107
Conversation
|
|
There was a problem hiding this comment.
Pull request overview
Fixes exception-safety holes in Arrow’s C++ ThreadPool worker-launch path so that a failed std::thread construction cannot leave the pool in a wedged state (stale workers_ entries and inflated tasks_queued_or_running_) that would hang Shutdown() / WaitForIdle() and eventually prevent new workers from being started.
Changes:
- Make
LaunchWorkersUnlocked()erase the just-appendedworkers_entry ifstd::threadconstruction throws, then rethrow. - Move
tasks_queued_or_running_increment inSpawnReal()to after a successful worker launch attempt (keeping the launch heuristic equivalent). - Add a fork-based regression test that forces thread creation failure and asserts the pool returns to a clean state and can still shut down.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.
| File | Description |
|---|---|
| cpp/src/arrow/util/thread_pool.h | Adds FRIEND_TEST access for the new fork-safety regression test. |
| cpp/src/arrow/util/thread_pool.cc | Restores ThreadPool internal invariants when worker thread creation throws; avoids counter leaks on launch failure. |
| cpp/src/arrow/util/thread_pool_test.cc | Adds FailedWorkerLaunch regression test that validates state restoration after forced thread creation failure. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
github-actions
Bot
added
awaiting committer review
|
All three done. |
There was a problem hiding this comment.
🟡 Changes recommended
The implementation and test expectations currently conflict with the PR description’s stated exception-propagation behavior, and the new test alters a process-wide rlimit in-process rather than isolating it in a forked child (risking flakiness).
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
- Files reviewed: 3/3 changed files
- Comments generated: 2
- Review effort level: Lite
| } catch (const std::exception& e) { | ||
| state_->workers_.erase(it); | ||
| return Status::UnknownError("Failed to launch worker thread: ", e.what()); | ||
| } |
| 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) { |
|
|
|
@github-actions crossbow submit -g cpp |
|
Revision: 0fa803a Submitted crossbow builds: ursacomputing/crossbow @ actions-b40897411d
|
pitrou
left a comment
pitrou
left a comment
There was a problem hiding this comment.
Thanks for the update @advitrocks9 ! LGTM now.
|
(note: CI failures are unrelated) |
Rationale for this change
LaunchWorkersUnlockedappends an entry tostate_->workers_before constructing the threadthat owns it, and only the worker itself erases that entry. If the
std::threadconstructorfails, the entry stays behind with nothing left to remove it, so
Shutdownwaits forever onworkers_.empty(). The destructor takes the same path. The failure also escapedSpawnRealafter
tasks_queued_or_running_had been incremented, soWaitForIdlenever returned either,and once stale entries filled
workers_to capacity the pool stopped launching workers whileSpawnstill returned OK for tasks nothing would run.What changes are included in this PR?
LaunchWorkersUnlockedreturns aStatus. A failed thread construction erases the entry it hadreserved and returns an error, which
SpawnRealandSetCapacitypropagate. The task counter isincremented after the launch rather than before, so a failed launch cannot leak a count.
Are these changes tested?
TestThreadPool.FailedWorkerLaunchlowersRLIMIT_NPROCto 1, spawns a task, restores the softlimit, and then checks the pool reports no workers and no tasks and still shuts down. It skips on
macOS, where
RLIMIT_NPROCcounts processes rather than threads, and skips anywhere else thelowered limit does not stop thread creation, such as under root.
Are there any user-facing changes?
Yes.
Spawn,SubmitandSetCapacityused to let astd::system_errorescape when the OSrefused a new thread. They return an error
Statusnow.ThreadPool::Makeis unaffected, sinceworker threads are only started on demand and a new pool starts none.