[native] Make PrestoServer::driverExecutor_ abstract type

Summary: Make PrestoServer::driverExecutor_  abstract folly::Executor type. This is to provide flexibility to make driver executor customizable, for example, for monitoring. This change also makes other executors unique_ptr because PrestoServer as the application backing server, should hold sole ownership of all application related executors (in fact the current usage does not need any features from shared_ptr).

Reviewed By: amitkdutta

Differential Revision: D79155996

Author: tanjialiang

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

@@ -488,8 +488,8 @@ void PrestoServer::run() {
       });
 
   PRESTO_STARTUP_LOG(INFO) << "Driver CPU executor '"
-                           << driverExecutor_->getName() << "' has "
-                           << driverExecutor_->numThreads() << " threads.";
+                           << driverCpuExecutor_->getName() << "' has "
+                           << driverCpuExecutor_->numThreads() << " threads.";
   if (httpServer_->getExecutor()) {
     PRESTO_STARTUP_LOG(INFO)
         << "HTTP Server IO executor '" << httpServer_->getExecutor()->getName()
@@ -519,7 +519,7 @@ void PrestoServer::run() {
   auto* memoryAllocator = velox::memory::memoryManager()->allocator();
   auto* asyncDataCache = velox::cache::AsyncDataCache::getInstance();
   periodicTaskManager_ = std::make_unique<PeriodicTaskManager>(
-      driverExecutor_.get(),
+      driverCpuExecutor_,
       spillerExecutor_.get(),
       httpSrvIoExecutor_.get(),
       httpSrvCpuExecutor_.get(),
@@ -648,10 +648,10 @@ void PrestoServer::run() {
   unregisterVeloxCudf();
 
   PRESTO_SHUTDOWN_LOG(INFO)
-      << "Joining Driver CPU Executor '" << driverExecutor_->getName()
-      << "': threads: " << driverExecutor_->numActiveThreads() << "/"
-      << driverExecutor_->numThreads()
-      << ", task queue: " << driverExecutor_->getTaskQueueSize();
+      << "Joining Driver CPU Executor '" << driverCpuExecutor_->getName()
+      << "': threads: " << driverCpuExecutor_->numActiveThreads() << "/"
+      << driverCpuExecutor_->numThreads()
+      << ", task queue: " << driverCpuExecutor_->getTaskQueueSize();
   // Schedule release of SessionPools held by HttpClients before the exchange
   // HTTP IO executor threads are joined.
   driverExecutor_.reset();
@@ -747,7 +747,7 @@ void PrestoServer::yieldTasks() {
     return;
   }
   static std::atomic<int32_t> numYields = 0;
-  const auto numQueued = driverExecutor_->getTaskQueueSize();
+  const auto numQueued = driverCpuExecutor_->getTaskQueueSize();
   if (numQueued > 0) {
     numYields += taskManager_->yieldTasks(numQueued, timeslice);
   }
@@ -796,8 +796,10 @@ void PrestoServer::initializeThreadPools() {
     threadFactory = std::make_shared<folly::NamedThreadFactory>("Driver");
   }
 
-  driverExecutor_ = std::make_shared<folly::CPUThreadPoolExecutor>(
+  auto driverExecutor = std::make_unique<folly::CPUThreadPoolExecutor>(
       numDriverCpuThreads, threadFactory);
+  driverCpuExecutor_ = driverExecutor.get();
+  driverExecutor_ = std::move(driverExecutor);
 
   const auto numIoThreads = std::max<size_t>(
       systemConfig->httpServerNumIoThreadsHwMultiplier() * hwConcurrency, 1);
@@ -806,13 +808,13 @@ void PrestoServer::initializeThreadPools() {
 
   const auto numCpuThreads = std::max<size_t>(
       systemConfig->httpServerNumCpuThreadsHwMultiplier() * hwConcurrency, 1);
-  httpSrvCpuExecutor_ = std::make_shared<folly::CPUThreadPoolExecutor>(
+  httpSrvCpuExecutor_ = std::make_unique<folly::CPUThreadPoolExecutor>(
       numCpuThreads, std::make_shared<folly::NamedThreadFactory>("HTTPSrvCpu"));
 
   const auto numSpillerCpuThreads = std::max<size_t>(
       systemConfig->spillerNumCpuThreadsHwMultiplier() * hwConcurrency, 0);
   if (numSpillerCpuThreads > 0) {
-    spillerExecutor_ = std::make_shared<folly::CPUThreadPoolExecutor>(
+    spillerExecutor_ = std::make_unique<folly::CPUThreadPoolExecutor>(
         numSpillerCpuThreads,
         std::make_shared<folly::NamedThreadFactory>("Spiller"));
   }
@@ -821,7 +823,7 @@ void PrestoServer::initializeThreadPools() {
       systemConfig->exchangeHttpClientNumIoThreadsHwMultiplier() *
           std::thread::hardware_concurrency(),
       1);
-  exchangeHttpIoExecutor_ = std::make_shared<folly::IOThreadPoolExecutor>(
+  exchangeHttpIoExecutor_ = std::make_unique<folly::IOThreadPoolExecutor>(
       numExchangeHttpClientIoThreads,
       std::make_shared<folly::NamedThreadFactory>("ExchangeIO"));
 
@@ -841,7 +843,7 @@ void PrestoServer::initializeThreadPools() {
           std::thread::hardware_concurrency(),
       1);
 
-  exchangeHttpCpuExecutor_ = std::make_shared<folly::CPUThreadPoolExecutor>(
+  exchangeHttpCpuExecutor_ = std::make_unique<folly::CPUThreadPoolExecutor>(
       numExchangeHttpClientCpuThreads,
       std::make_shared<folly::NamedThreadFactory>("ExchangeCPU"));
 
@@ -1035,7 +1037,7 @@ size_t PrestoServer::numDriverThreads() const {
   VELOX_CHECK(
       driverExecutor_ != nullptr,
       "Driver executor is expected to be not null, but it is null!");
-  return driverExecutor_->numThreads();
+  return driverCpuExecutor_->numThreads();
 }
 
 void PrestoServer::detachWorker() {
@@ -1441,7 +1443,7 @@ void PrestoServer::enableWorkerStatsReporting() {
 
 void PrestoServer::initVeloxPlanValidator() {
   VELOX_CHECK_NULL(planValidator_);
-  planValidator_ = std::make_shared<VeloxPlanValidator>();
+  planValidator_ = std::make_unique<VeloxPlanValidator>();
 }
 
 VeloxPlanValidator* PrestoServer::getVeloxPlanValidator() {

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

@@ -198,7 +198,6 @@ class PrestoServer {
 
   void registerStatsCounters();
 
- protected:
   void updateAnnouncerDetails();
 
   void addServerPeriodicTasks();
@@ -242,24 +241,31 @@ class PrestoServer {
   std::unique_ptr<folly::IOThreadPoolExecutor> connectorIoExecutor_;
 
   // Executor for exchange data over http.
-  std::shared_ptr<folly::IOThreadPoolExecutor> exchangeHttpIoExecutor_;
+  std::unique_ptr<folly::IOThreadPoolExecutor> exchangeHttpIoExecutor_;
 
   // Executor for exchange request processing.
-  std::shared_ptr<folly::CPUThreadPoolExecutor> exchangeHttpCpuExecutor_;
+  std::unique_ptr<folly::CPUThreadPoolExecutor> exchangeHttpCpuExecutor_;
 
   // Executor for HTTP request dispatching
   std::shared_ptr<folly::IOThreadPoolExecutor> httpSrvIoExecutor_;
 
   // Executor for HTTP request processing after dispatching
-  std::shared_ptr<folly::CPUThreadPoolExecutor> httpSrvCpuExecutor_;
-
-  // Executor for query engine driver executions.
-  std::shared_ptr<folly::CPUThreadPoolExecutor> driverExecutor_;
+  std::unique_ptr<folly::CPUThreadPoolExecutor> httpSrvCpuExecutor_;
+
+  // Executor for query engine driver executions. 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 
+  // 'driverCpuExecutor()' method.
+  std::unique_ptr<folly::Executor> driverExecutor_;
+  // Raw pointer pointing to the underlying folly::CPUThreadPoolExecutor of 
+  // 'driverExecutor_'.
+  folly::CPUThreadPoolExecutor* driverCpuExecutor_;
 
   // Executor for spilling.
-  std::shared_ptr<folly::CPUThreadPoolExecutor> spillerExecutor_;
+  std::unique_ptr<folly::CPUThreadPoolExecutor> spillerExecutor_;
 
-  std::shared_ptr<VeloxPlanValidator> planValidator_;
+  std::unique_ptr<VeloxPlanValidator> planValidator_;
 
   std::unique_ptr<http::HttpClientConnectionPool> exchangeSourceConnectionPool_;