[Prestissimo] Add stall thread monitoring to Spiller cpu pool

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(),
@@ -854,9 +854,11 @@ void PrestoServer::initializeThreadPools() {
   const auto numSpillerCpuThreads = std::max<size_t>(
       systemConfig->spillerNumCpuThreadsHwMultiplier() * hwConcurrency, 0);
   if (numSpillerCpuThreads > 0) {
-    spillerExecutor_ = std::make_unique<folly::CPUThreadPoolExecutor>(
+    auto spillerExecutor = std::make_unique<folly::CPUThreadPoolExecutor>(
         numSpillerCpuThreads,
         std::make_shared<folly::NamedThreadFactory>("Spiller"));
+    spillerCpuExecutor_ = spillerExecutor.get();
+    spillerExecutor_ = std::move(spillerExecutor);
   }
 
   const auto numExchangeHttpClientIoThreads = std::max<size_t>(
@@ -1793,7 +1795,7 @@ void PrestoServer::registerDynamicFunctions() {
 
 void PrestoServer::createTaskManager() {
   taskManager_ = std::make_unique<TaskManager>(
-      driverExecutor_.get(), httpSrvCpuExecutor_.get(), spillerExecutor_.get());
+      driverCpuExecutor_, httpSrvCpuExecutor_.get(), spillerCpuExecutor_);
 }
 
 void PrestoServer::reportNodeStats(proxygen::ResponseHandler* downstream) {

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

@@ -269,7 +269,8 @@ class PrestoServer {
   folly::CPUThreadPoolExecutor* driverCpuExecutor_;
 
   // Executor for spilling.
-  std::unique_ptr<folly::CPUThreadPoolExecutor> spillerExecutor_;
+  std::unique_ptr<folly::Executor> spillerExecutor_;
+  folly::CPUThreadPoolExecutor* spillerCpuExecutor_;
 
   std::unique_ptr<VeloxPlanValidator> planValidator_;