pitrou commented on code in PR #51107:
URL: https://github.com/apache/arrow/pull/51107#discussion_r3902781387
##########
cpp/src/arrow/util/thread_pool_test.cc:
##########
@@ -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();
Review Comment:
Do we actually need to use `fork` for this test, or can we just use
`setrlimit` to temporarily lower the limit and then raise it again?
##########
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;
Review Comment:
We don't use C++ exceptions in Arrow, please let's return a `Status` here
instead.
##########
cpp/src/arrow/util/thread_pool_test.cc:
##########
@@ -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__
Review Comment:
Is this actually required for macOS? We can just as well skip the test on
Apple, instead of adding this rather complicated initialization code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
