feat: Add stall thread monitoring to Spiller cpu pool (#26695)

Summary:

as title

Differential Revision: D87850298

Author: duxiao1212

presto-native-execution/presto_cpp/main/PrestoServer.cpp

@@ -548,8 +548,8 @@ void PrestoServer::run() {
   }
   if (spillerExecutor_ != nullptr) {
     PRESTO_STARTUP_LOG(INFO)
-        << "Spiller CPU executor '" << spillerExecutor_->getName() << "', has "
-        << spillerExecutor_->numThreads() << " threads.";
+        << "Spiller CPU executor '" << spillerCpuExecutor_->getName()
+        << "', has " << spillerCpuExecutor_->numThreads() << " threads.";
   } else {
     PRESTO_STARTUP_LOG(INFO) << "Spill executor was not configured.";
   }
@@ -560,7 +560,7 @@ void PrestoServer::run() {
   auto* asyncDataCache = velox::cache::AsyncDataCache::getInstance();
   periodicTaskManager_ = std::make_unique<PeriodicTaskManager>(
       driverCpuExecutor_,
-      spillerExecutor_.get(),
+      spillerCpuExecutor_,
       httpSrvIoExecutor_.get(),
       httpSrvCpuExecutor_.get(),
       exchangeHttpIoExecutor_.get(),
@@ -836,10 +836,11 @@ void PrestoServer::initializeThreadPools() {
     threadFactory = std::make_shared<folly::NamedThreadFactory>("Driver");
   }
 
-  auto driverExecutor = std::make_unique<folly::CPUThreadPoolExecutor>(
+  driverExecutor_ = std::make_unique<folly::CPUThreadPoolExecutor>(
       numDriverCpuThreads, threadFactory);
-  driverCpuExecutor_ = driverExecutor.get();
-  driverExecutor_ = std::move(driverExecutor);
+  driverCpuExecutor_ =
+      velox::checked_pointer_cast<folly::CPUThreadPoolExecutor>(
+          driverExecutor_.get());
 
   const auto numIoThreads = std::max<size_t>(
       systemConfig->httpServerNumIoThreadsHwMultiplier() * hwConcurrency, 1);
@@ -857,8 +858,10 @@ void PrestoServer::initializeThreadPools() {
     spillerExecutor_ = std::make_unique<folly::CPUThreadPoolExecutor>(
         numSpillerCpuThreads,
         std::make_shared<folly::NamedThreadFactory>("Spiller"));
+    spillerCpuExecutor_ =
+        velox::checked_pointer_cast<folly::CPUThreadPoolExecutor>(
+            spillerExecutor_.get());
   }
-
   const auto numExchangeHttpClientIoThreads = std::max<size_t>(
       systemConfig->exchangeHttpClientNumIoThreadsHwMultiplier() *
           std::thread::hardware_concurrency(),

presto-native-execution/presto_cpp/main/PrestoServer.h

@@ -268,8 +268,14 @@ class PrestoServer {
   // 'driverExecutor_'.
   folly::CPUThreadPoolExecutor* driverCpuExecutor_;
 
-  // Executor for spilling.
-  std::unique_ptr<folly::CPUThreadPoolExecutor> spillerExecutor_;
+  // Executor for spilling. The underlying thread pool executor is a
+  // folly::CPUThreadPoolExecutor. The executor is stored as abstract type to
+  // provide flexibility of thread pool monitoring. The underlying
+  // folly::CPUThreadPoolExecutor can be obtained through 'spillerCpuExecutor_'.
+  std::unique_ptr<folly::Executor> spillerExecutor_;
+  // Raw pointer pointing to the underlying folly::CPUThreadPoolExecutor of
+  // 'spillerExecutor_'.
+  folly::CPUThreadPoolExecutor* spillerCpuExecutor_;
 
   std::unique_ptr<VeloxPlanValidator> planValidator_;