fix(thread-pool): cap workers at hardware concurrency

Author: luoxiaojian

src/ailego/parallel/thread_pool.cc

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <algorithm>
 #include <zvec/ailego/logger/logger.h>
 #include <zvec/ailego/parallel/thread_pool.h>
 
@@ -80,12 +81,42 @@ static inline void UnbindThreads(std::vector<std::thread> &) {}
 namespace zvec {
 namespace ailego {
 
+namespace {
+
+uint32_t SafeWorkerCount() noexcept {
+  return std::max(std::thread::hardware_concurrency(), 1u);
+}
+
+}  // namespace
+
 ThreadPool::ThreadPool(uint32_t size, bool binding) {
-  for (uint32_t i = 0u; i < size; ++i) {
-    pool_.emplace_back(&ThreadPool::worker, this);
+  const uint32_t max_size = SafeWorkerCount();
+  const uint32_t safe_size = std::min(size, max_size);
+  if (safe_size != size) {
+    LOG_WARN(
+        "ThreadPool worker count[%u] exceeds available CPUs[%u], "
+        "clamped to %u",
+        size, max_size, safe_size);
   }
-  if (binding) {
-    this->bind();
+
+  try {
+    // Avoid vector reallocations after workers have started. If thread creation
+    // still fails, stop and join the workers already created before rethrowing.
+    pool_.reserve(safe_size);
+    for (uint32_t i = 0u; i < safe_size; ++i) {
+      pool_.emplace_back(&ThreadPool::worker, this);
+    }
+    if (binding) {
+      this->bind();
+    }
+  } catch (...) {
+    this->stop();
+    for (auto &worker : pool_) {
+      if (worker.joinable()) {
+        worker.join();
+      }
+    }
+    throw;
   }
 }
 

tests/ailego/parallel/thread_pool_test.cc

@@ -12,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <algorithm>
 #include <chrono>
 #include <iostream>
+#include <limits>
 #include <memory>
 #include <type_traits>
 #include <gtest/gtest.h>
@@ -42,8 +44,16 @@ TEST(ThreadPool, ConstructorBehavior) {
   ThreadPool single_unbound_pool(1, false);
   EXPECT_EQ(1u, single_unbound_pool.count());
 
-  ThreadPool double_bound_pool(2, true);
-  EXPECT_EQ(2u, double_bound_pool.count());
+  const uint32_t bound_count = std::min(hardware_concurrency, 2u);
+  ThreadPool bound_pool(bound_count, true);
+  EXPECT_EQ(bound_count, bound_pool.count());
+}
+
+TEST(ThreadPool, CapsExcessiveWorkerCount) {
+  const auto hardware_concurrency =
+      std::max(std::thread::hardware_concurrency(), 1u);
+  ThreadPool pool((std::numeric_limits<uint32_t>::max)(), false);
+  EXPECT_EQ(hardware_concurrency, pool.count());
 }
 
 #if defined(__linux__) && !defined(__ANDROID__)