Skip to content

Commit 360276c

Browse files
authored
feat: Allow flexible thread pool monitoring for Spiller cpu pool (#26695)
Summary: as title Differential Revision: D87850298 ## Release Notes ``` == NO RELEASE NOTE == ```
1 parent 327032c commit 360276c

File tree

2 files changed

+18
-9
lines changed

2 files changed

+18
-9
lines changed

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

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -549,8 +549,8 @@ void PrestoServer::run() {
549549
}
550550
if (spillerExecutor_ != nullptr) {
551551
PRESTO_STARTUP_LOG(INFO)
552-
<< "Spiller CPU executor '" << spillerExecutor_->getName() << "', has "
553-
<< spillerExecutor_->numThreads() << " threads.";
552+
<< "Spiller CPU executor '" << spillerCpuExecutor_->getName()
553+
<< "', has " << spillerCpuExecutor_->numThreads() << " threads.";
554554
} else {
555555
PRESTO_STARTUP_LOG(INFO) << "Spill executor was not configured.";
556556
}
@@ -561,7 +561,7 @@ void PrestoServer::run() {
561561
auto* asyncDataCache = velox::cache::AsyncDataCache::getInstance();
562562
periodicTaskManager_ = std::make_unique<PeriodicTaskManager>(
563563
driverCpuExecutor_,
564-
spillerExecutor_.get(),
564+
spillerCpuExecutor_,
565565
httpSrvIoExecutor_.get(),
566566
httpSrvCpuExecutor_.get(),
567567
exchangeHttpIoExecutor_.get(),
@@ -837,10 +837,11 @@ void PrestoServer::initializeThreadPools() {
837837
threadFactory = std::make_shared<folly::NamedThreadFactory>("Driver");
838838
}
839839

840-
auto driverExecutor = std::make_unique<folly::CPUThreadPoolExecutor>(
840+
driverExecutor_ = std::make_unique<folly::CPUThreadPoolExecutor>(
841841
numDriverCpuThreads, threadFactory);
842-
driverCpuExecutor_ = driverExecutor.get();
843-
driverExecutor_ = std::move(driverExecutor);
842+
driverCpuExecutor_ =
843+
velox::checked_pointer_cast<folly::CPUThreadPoolExecutor>(
844+
driverExecutor_.get());
844845

845846
const auto numIoThreads = std::max<size_t>(
846847
systemConfig->httpServerNumIoThreadsHwMultiplier() * hwConcurrency, 1);
@@ -858,8 +859,10 @@ void PrestoServer::initializeThreadPools() {
858859
spillerExecutor_ = std::make_unique<folly::CPUThreadPoolExecutor>(
859860
numSpillerCpuThreads,
860861
std::make_shared<folly::NamedThreadFactory>("Spiller"));
862+
spillerCpuExecutor_ =
863+
velox::checked_pointer_cast<folly::CPUThreadPoolExecutor>(
864+
spillerExecutor_.get());
861865
}
862-
863866
const auto numExchangeHttpClientIoThreads = std::max<size_t>(
864867
systemConfig->exchangeHttpClientNumIoThreadsHwMultiplier() *
865868
std::thread::hardware_concurrency(),

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

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -268,8 +268,14 @@ class PrestoServer {
268268
// 'driverExecutor_'.
269269
folly::CPUThreadPoolExecutor* driverCpuExecutor_;
270270

271-
// Executor for spilling.
272-
std::unique_ptr<folly::CPUThreadPoolExecutor> spillerExecutor_;
271+
// Executor for spilling. The underlying thread pool executor is a
272+
// folly::CPUThreadPoolExecutor. The executor is stored as abstract type to
273+
// provide flexibility of thread pool monitoring. The underlying
274+
// folly::CPUThreadPoolExecutor can be obtained through 'spillerCpuExecutor_'.
275+
std::unique_ptr<folly::Executor> spillerExecutor_;
276+
// Raw pointer pointing to the underlying folly::CPUThreadPoolExecutor of
277+
// 'spillerExecutor_'.
278+
folly::CPUThreadPoolExecutor* spillerCpuExecutor_;
273279

274280
std::unique_ptr<VeloxPlanValidator> planValidator_;
275281

0 commit comments

Comments
 (0)