Skip to content

Commit 20738bc

Browse files
Jimmy Luxiaoxmeng
Jimmy Lu
authored and
xiaoxmeng
committed
[native] Follow up on task watchdog
1 parent 6af66f1 commit 20738bc

7 files changed

Lines changed: 50 additions & 50 deletions

File tree

‎presto-native-execution/presto_cpp/main/PeriodicTaskManager.cpp‎

Lines changed: 22 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -72,17 +72,15 @@ PeriodicTaskManager::PeriodicTaskManager(
7272
const std::unordered_map<
7373
std::string,
7474
std::shared_ptr<velox::connector::Connector>>& connectors,
75-
PrestoServer* server,
76-
size_t stuckDriverThresholdMs)
75+
PrestoServer* server)
7776
: driverCPUExecutor_(driverCPUExecutor),
7877
httpExecutor_(httpExecutor),
7978
taskManager_(taskManager),
8079
memoryAllocator_(memoryAllocator),
8180
asyncDataCache_(asyncDataCache),
8281
arbitrator_(velox::memory::memoryManager()->arbitrator()),
8382
connectors_(connectors),
84-
server_(server),
85-
stuckDriverThresholdMs_(stuckDriverThresholdMs) {}
83+
server_(server) {}
8684

8785
void PeriodicTaskManager::start() {
8886
// If executors are null, don't bother starting this task.
@@ -124,12 +122,12 @@ void PeriodicTaskManager::start() {
124122

125123
addWatchdogTask();
126124

127-
onceRunner_.start();
125+
oneTimeRunner_.start();
128126
}
129127

130128
void PeriodicTaskManager::stop() {
131-
onceRunner_.cancelAllFunctionsAndWait();
132-
onceRunner_.shutdown();
129+
oneTimeRunner_.cancelAllFunctionsAndWait();
130+
oneTimeRunner_.shutdown();
133131
repeatedRunner_.stop();
134132
}
135133

@@ -463,11 +461,11 @@ class HiveConnectorStatsReporter {
463461
RECORD_METRIC_VALUE(curSizeMetricName_, stats.curSize);
464462
RECORD_METRIC_VALUE(numAccumulativeHitsMetricName_, stats.numHits);
465463
RECORD_METRIC_VALUE(numAccumulativeLookupsMetricName_, stats.numLookups);
466-
RECORD_METRIC_VALUE(numHitsMetricName_, stats.numHits - oldNumHits_);
467-
oldNumHits_ = stats.numHits;
464+
RECORD_METRIC_VALUE(numHitsMetricName_, stats.numHits - lastNumHits_);
465+
lastNumHits_ = stats.numHits;
468466
RECORD_METRIC_VALUE(
469-
numLookupsMetricName_, stats.numLookups - oldNumLookups_);
470-
oldNumLookups_ = stats.numLookups;
467+
numLookupsMetricName_, stats.numLookups - lastNumLookups_);
468+
lastNumLookups_ = stats.numLookups;
471469
}
472470

473471
private:
@@ -479,8 +477,8 @@ class HiveConnectorStatsReporter {
479477
const std::string numAccumulativeLookupsMetricName_;
480478
const std::string numHitsMetricName_;
481479
const std::string numLookupsMetricName_;
482-
size_t oldNumHits_{0};
483-
size_t oldNumLookups_{0};
480+
size_t lastNumHits_{0};
481+
size_t lastNumLookups_{0};
484482
};
485483

486484
} // namespace
@@ -663,29 +661,27 @@ void PeriodicTaskManager::addHttpEndpointLatencyStatsTask() {
663661

664662
void PeriodicTaskManager::addWatchdogTask() {
665663
addTask(
666-
[this,
667-
deadlockedTasks = std::vector<std::string>(),
668-
opCalls = std::vector<velox::exec::Task::OpCallInfo>()]() mutable {
669-
deadlockedTasks.clear();
670-
opCalls.clear();
671-
if (!taskManager_->getLongRunningOpCalls(
672-
stuckDriverThresholdMs_, deadlockedTasks, opCalls)) {
664+
[this] {
665+
std::vector<std::string> deadlockTasks;
666+
std::vector<velox::exec::Task::OpCallInfo> stuckOpCalls;
667+
if (!taskManager_->getStuckOpCalls(deadlockTasks, stuckOpCalls)) {
673668
LOG(ERROR)
674669
<< "Cannot take lock on task manager, likely starving or deadlocked";
675-
RECORD_METRIC_VALUE(kCounterNumTasksDeadlock, 1);
670+
RECORD_METRIC_VALUE(kCounterNumTaskManagerLockTimeOut, 1);
676671
detachWorker();
677672
return;
678673
}
679-
for (auto& taskId : deadlockedTasks) {
674+
RECORD_METRIC_VALUE(kCounterNumTaskManagerLockTimeOut, 0);
675+
for (const auto& taskId : deadlockTasks) {
680676
LOG(ERROR) << "Starving or deadlocked task: " << taskId;
681677
}
682-
RECORD_METRIC_VALUE(kCounterNumTasksDeadlock, deadlockedTasks.size());
683-
for (auto& call : opCalls) {
678+
RECORD_METRIC_VALUE(kCounterNumTasksDeadlock, deadlockTasks.size());
679+
for (const auto& call : stuckOpCalls) {
684680
LOG(ERROR) << "Stuck operator: tid=" << call.tid
685681
<< " taskId=" << call.taskId << " opId=" << call.opId;
686682
}
687-
RECORD_METRIC_VALUE(kCounterNumStuckDrivers, opCalls.size());
688-
if (!deadlockedTasks.empty() || !opCalls.empty()) {
683+
RECORD_METRIC_VALUE(kCounterNumStuckDrivers, stuckOpCalls.size());
684+
if (!deadlockTasks.empty() || !stuckOpCalls.empty()) {
689685
detachWorker();
690686
}
691687
},

‎presto-native-execution/presto_cpp/main/PeriodicTaskManager.h‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
#include <folly/experimental/ThreadedRepeatingFunctionRunner.h>
1818
#include "velox/common/memory/Memory.h"
1919
#include "velox/exec/Spill.h"
20+
#include "velox/exec/Task.h"
2021

2122
namespace folly {
2223
class CPUThreadPoolExecutor;
@@ -50,8 +51,7 @@ class PeriodicTaskManager {
5051
const std::unordered_map<
5152
std::string,
5253
std::shared_ptr<velox::connector::Connector>>& connectors,
53-
PrestoServer* server,
54-
size_t stuckDriverThresholdMs);
54+
PrestoServer* server);
5555

5656
~PeriodicTaskManager() {
5757
stop();
@@ -88,8 +88,8 @@ class PeriodicTaskManager {
8888
template <typename TFunc>
8989
void
9090
addTaskOnce(TFunc&& func, size_t periodMicros, const std::string& taskName) {
91-
onceRunner_.cancelFunction(taskName);
92-
onceRunner_.addFunctionOnce(
91+
oneTimeRunner_.cancelFunction(taskName);
92+
oneTimeRunner_.addFunctionOnce(
9393
std::forward<TFunc>(func),
9494
taskName,
9595
std::chrono::microseconds{periodMicros});
@@ -146,7 +146,6 @@ class PeriodicTaskManager {
146146
std::string,
147147
std::shared_ptr<velox::connector::Connector>>& connectors_;
148148
PrestoServer* const server_;
149-
const size_t stuckDriverThresholdMs_;
150149

151150
// Cache related stats
152151
int64_t lastMemoryCacheHits_{0};
@@ -169,8 +168,8 @@ class PeriodicTaskManager {
169168
velox::common::SpillStats lastSpillStats_;
170169
velox::memory::MemoryArbitrator::Stats lastArbitratorStats_;
171170

172-
// CAUTION: Declare last since the threads access other members of `this`.
173-
folly::FunctionScheduler onceRunner_;
171+
// NOTE: declare last since the threads access other members of `this`.
172+
folly::FunctionScheduler oneTimeRunner_;
174173
folly::ThreadedRepeatingFunctionRunner repeatedRunner_;
175174
};
176175

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -482,8 +482,7 @@ void PrestoServer::run() {
482482
memoryAllocator,
483483
asyncDataCache,
484484
velox::connector::getAllConnectors(),
485-
this,
486-
systemConfig->driverStuckOperatorThresholdMs());
485+
this);
487486
addServerPeriodicTasks();
488487
addAdditionalPeriodicTasks();
489488
periodicTaskManager_->start();

‎presto-native-execution/presto_cpp/main/TaskManager.cpp‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1041,20 +1041,21 @@ DriverCountStats TaskManager::getDriverCountStats() const {
10411041
return driverCountStats;
10421042
}
10431043

1044-
bool TaskManager::getLongRunningOpCalls(
1045-
size_t thresholdDurationMs,
1046-
std::vector<std::string>& deadlockedTasks,
1047-
std::vector<velox::exec::Task::OpCallInfo>& opCalls) const {
1048-
std::chrono::milliseconds lockTimeout(thresholdDurationMs);
1049-
auto taskMap = taskMap_.rlock(lockTimeout);
1044+
bool TaskManager::getStuckOpCalls(
1045+
std::vector<std::string>& deadlockTasks,
1046+
std::vector<velox::exec::Task::OpCallInfo>& stuckOpCalls) const {
1047+
const auto thresholdDurationMs =
1048+
SystemConfig::instance()->driverStuckOperatorThresholdMs();
1049+
const std::chrono::milliseconds lockTimeoutMs(thresholdDurationMs);
1050+
auto taskMap = taskMap_.rlock(lockTimeoutMs);
10501051
if (!taskMap) {
10511052
return false;
10521053
}
10531054
for (const auto& [_, prestoTask] : *taskMap) {
10541055
if (prestoTask->task != nullptr &&
10551056
!prestoTask->task->getLongRunningOpCalls(
1056-
lockTimeout, thresholdDurationMs, opCalls)) {
1057-
deadlockedTasks.push_back(prestoTask->task->taskId());
1057+
lockTimeoutMs, thresholdDurationMs, stuckOpCalls)) {
1058+
deadlockTasks.push_back(prestoTask->task->taskId());
10581059
}
10591060
}
10601061
return true;

‎presto-native-execution/presto_cpp/main/TaskManager.h‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -146,13 +146,14 @@ class TaskManager {
146146
// in exec/Task.h).
147147
std::array<size_t, 5> getTaskNumbers(size_t& numTasks) const;
148148

149-
/// Populate the blocked tasks (failing to take lock on the mutex), and long
150-
/// running operator calls across all drivers in all tasks. Returns false if
151-
/// a lock on the taskMap cannot be taken, otherwise returns true.
152-
bool getLongRunningOpCalls(
153-
size_t thresholdDurationMs,
154-
std::vector<std::string>& deadlockedTasks,
155-
std::vector<velox::exec::Task::OpCallInfo>& opCalls) const;
149+
/// Invoked to check the stuck operation calls in the system. If the function
150+
/// fails to get the stuck call information from a task due to the lock
151+
/// timeout, it adds the task to 'blockedTasks'. Otherwise, it adds all stuck
152+
/// call information to 'stuckOpCalls'. The function returns false if a lock
153+
/// on the taskMap cannot be taken, otherwise returns true.
154+
bool getStuckOpCalls(
155+
std::vector<std::string>& deadlockTasks,
156+
std::vector<velox::exec::Task::OpCallInfo>& stuckOpCalls) const;
156157

157158
/// Build directory path for spilling for the given task.
158159
/// Always returns non-empty string.

‎presto-native-execution/presto_cpp/main/common/Counters.cpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,8 @@ void registerPrestoMetrics() {
6161
DEFINE_METRIC(
6262
kCounterNumTasksWithStuckOperator, facebook::velox::StatType::AVG);
6363
DEFINE_METRIC(kCounterNumTasksDeadlock, facebook::velox::StatType::AVG);
64+
DEFINE_METRIC(
65+
kCounterNumTaskManagerLockTimeOut, facebook::velox::StatType::AVG);
6466
DEFINE_METRIC(kCounterNumRunningDrivers, facebook::velox::StatType::AVG);
6567
DEFINE_METRIC(kCounterNumBlockedDrivers, facebook::velox::StatType::AVG);
6668
DEFINE_METRIC(kCounterNumStuckDrivers, facebook::velox::StatType::AVG);

‎presto-native-execution/presto_cpp/main/common/Counters.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,8 @@ constexpr folly::StringPiece kCounterNumTasksWithStuckOperator{
7070
"presto_cpp.num_tasks_with_stuck_operator"};
7171
constexpr folly::StringPiece kCounterNumTasksDeadlock{
7272
"presto_cpp.num_tasks_deadlock"};
73+
constexpr folly::StringPiece kCounterNumTaskManagerLockTimeOut{
74+
"presto_cpp.num_tasks_manager_lock_timeout"};
7375
constexpr folly::StringPiece kCounterNumRunningDrivers{
7476
"presto_cpp.num_running_drivers"};
7577
constexpr folly::StringPiece kCounterNumBlockedDrivers{

0 commit comments

Comments
 (0)