[native] Add optional Task queuing framework.

Author: spershin

presto-docs/src/main/sphinx/presto_cpp/properties.rst

@@ -617,6 +617,25 @@ memory use. Ignored if zero.
 CPU threshold in % above which the worker is considered overloaded in terms of
 CPU use. Ignored if zero.
 
+``worker-overloaded-cooldown-period-sec``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``integer``
+* **Default value:** ``5``
+
+Specifies how many seconds worker has to be not overloaded (in terms of
+memory and CPU) before its status changes to not overloaded.
+This is to prevent spiky fluctuation of the overloaded status.
+
+``worker-overloaded-task-queuing-enabled``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``false``
+
+If true, the worker starts queuing new tasks when overloaded, and
+starts them gradually when it stops being overloaded.
+
 Environment Variables As Values For Worker Properties
 -----------------------------------------------------
 

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

@@ -334,19 +334,23 @@ void PeriodicTaskManager::updateTaskStats() {
       kCounterNumTasksBytesProcessed, taskManager_->getBytesProcessed());
   RECORD_METRIC_VALUE(
       kCounterNumTasksRunning,
-      taskNumbers[static_cast<int>(velox::exec::TaskState::kRunning)]);
+      taskNumbers[static_cast<int>(PrestoTaskState::kRunning)]);
   RECORD_METRIC_VALUE(
       kCounterNumTasksFinished,
-      taskNumbers[static_cast<int>(velox::exec::TaskState::kFinished)]);
+      taskNumbers[static_cast<int>(PrestoTaskState::kFinished)]);
   RECORD_METRIC_VALUE(
       kCounterNumTasksCancelled,
-      taskNumbers[static_cast<int>(velox::exec::TaskState::kCanceled)]);
+      taskNumbers[static_cast<int>(PrestoTaskState::kCanceled)]);
   RECORD_METRIC_VALUE(
       kCounterNumTasksAborted,
-      taskNumbers[static_cast<int>(velox::exec::TaskState::kAborted)]);
+      taskNumbers[static_cast<int>(PrestoTaskState::kAborted)]);
   RECORD_METRIC_VALUE(
       kCounterNumTasksFailed,
-      taskNumbers[static_cast<int>(velox::exec::TaskState::kFailed)]);
+      taskNumbers[static_cast<int>(PrestoTaskState::kFailed)]);
+  RECORD_METRIC_VALUE(
+      kCounterNumTasksPlanned,
+      taskNumbers[static_cast<int>(PrestoTaskState::kPlanned)]);
+  RECORD_METRIC_VALUE(kCounterNumTasksQueued, taskManager_->numQueuedTasks());
 
   const auto driverCounts = taskManager_->getDriverCounts();
   RECORD_METRIC_VALUE(kCounterNumQueuedDrivers, driverCounts.numQueuedDrivers);

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

@@ -1425,6 +1425,9 @@ void PrestoServer::populateMemAndCPUInfo() {
     poolInfo.reservedBytes += bytes;
   });
   RECORD_METRIC_VALUE(kCounterNumQueryContexts, numContexts);
+  RECORD_METRIC_VALUE(
+      kCounterMemoryManagerTotalBytes,
+      velox::memory::memoryManager()->getTotalBytes());
   cpuMon_.update();
   checkOverload();
   **memoryInfo_.wlock() = std::move(memoryInfo);
@@ -1439,41 +1442,74 @@ void PrestoServer::checkOverload() {
     const auto currentUsedMemoryBytes = (memoryChecker_ != nullptr)
         ? memoryChecker_->cachedSystemUsedMemoryBytes()
         : 0;
-    const bool isMemOverloaded =
+    const bool memOverloaded =
         (currentUsedMemoryBytes > overloadedThresholdMemBytes);
-    if (isMemOverloaded) {
-      LOG(WARNING) << "Server memory is overloaded. Currently used: "
+    if (memOverloaded && !memOverloaded_) {
+      LOG(WARNING) << "OVERLOAD: Server memory is overloaded. Currently used: "
                    << velox::succinctBytes(currentUsedMemoryBytes)
+                   << ", including tracked: "
+                   << velox::succinctBytes(
+                          velox::memory::memoryManager()->getTotalBytes())
                    << ", threshold: "
                    << velox::succinctBytes(overloadedThresholdMemBytes);
-    } else if (isMemOverloaded_) {
-      LOG(INFO) << "Server memory is no longer overloaded. Currently used: "
-                << velox::succinctBytes(currentUsedMemoryBytes)
-                << ", threshold: "
-                << velox::succinctBytes(overloadedThresholdMemBytes);
+    } else if (!memOverloaded && memOverloaded_) {
+      LOG(INFO)
+          << "OVERLOAD: Server memory is no longer overloaded. Currently used: "
+          << velox::succinctBytes(currentUsedMemoryBytes)
+          << ", including tracked: "
+          << velox::succinctBytes(
+                 velox::memory::memoryManager()->getTotalBytes())
+          << ", threshold: "
+          << velox::succinctBytes(overloadedThresholdMemBytes);
     }
-    RECORD_METRIC_VALUE(kCounterOverloadedMem, isMemOverloaded ? 100 : 0);
-    isMemOverloaded_ = isMemOverloaded;
+    RECORD_METRIC_VALUE(kCounterOverloadedMem, memOverloaded ? 100 : 0);
+    memOverloaded_ = memOverloaded;
   }
 
   const auto overloadedThresholdCpuPct =
       systemConfig->workerOverloadedThresholdCpuPct();
   if (overloadedThresholdCpuPct > 0) {
     const auto currentUsedCpuPct = cpuMon_.getCPULoadPct();
-    const bool isCpuOverloaded =
-        (currentUsedCpuPct > overloadedThresholdCpuPct);
-    if (isCpuOverloaded) {
-      LOG(WARNING) << "Server CPU is overloaded. Currently used: "
+    const bool cpuOverloaded = (currentUsedCpuPct > overloadedThresholdCpuPct);
+    if (cpuOverloaded && !cpuOverloaded_) {
+      LOG(WARNING) << "OVERLOAD: Server CPU is overloaded. Currently used: "
                    << currentUsedCpuPct
                    << "%, threshold: " << overloadedThresholdCpuPct << "%";
-    } else if (isCpuOverloaded_) {
-      LOG(INFO) << "Server CPU is no longer overloaded. Currently used: "
-                << currentUsedCpuPct
-                << "%, threshold: " << overloadedThresholdCpuPct << "%";
+    } else if (!cpuOverloaded && cpuOverloaded_) {
+      LOG(INFO)
+          << "OVERLOAD: Server CPU is no longer overloaded. Currently used: "
+          << currentUsedCpuPct << "%, threshold: " << overloadedThresholdCpuPct
+          << "%";
     }
-    RECORD_METRIC_VALUE(kCounterOverloadedCpu, isCpuOverloaded ? 100 : 0);
-    isCpuOverloaded_ = isCpuOverloaded;
+    RECORD_METRIC_VALUE(kCounterOverloadedCpu, cpuOverloaded ? 100 : 0);
+    cpuOverloaded_ = cpuOverloaded;
   }
+
+  // Determine if the server is overloaded. We require both memory and CPU to be
+  // not overloaded for some time (continuous period) to consider the server as
+  // not overloaded.
+  const uint64_t currentTimeSecs = velox::getCurrentTimeSec();
+  if (memOverloaded_ || cpuOverloaded_) {
+    lastOverloadedTimeInSecs_ = currentTimeSecs;
+  }
+  VELOX_CHECK_GE(currentTimeSecs, lastOverloadedTimeInSecs_);
+  const bool serverOverloaded =
+      ((cpuOverloaded_ || memOverloaded_) ||
+       ((currentTimeSecs - lastOverloadedTimeInSecs_) <
+        systemConfig->workerOverloadedCooldownPeriodSec()));
+
+  if (serverOverloaded && !serverOverloaded_) {
+    LOG(WARNING) << "OVERLOAD: Server is overloaded.";
+  } else if (!serverOverloaded && serverOverloaded_) {
+    LOG(INFO) << "OVERLOAD: Server is no longer overloaded.";
+  }
+  RECORD_METRIC_VALUE(kCounterOverloaded, serverOverloaded ? 100 : 0);
+  serverOverloaded_ = serverOverloaded;
+
+  if (systemConfig->workerOverloadedTaskQueuingEnabled()) {
+    taskManager_->setServerOverloaded(serverOverloaded_);
+  }
+  taskManager_->maybeStartNextQueuedTask();
 }
 
 static protocol::Duration getUptime(

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

@@ -275,8 +275,16 @@ class PrestoServer {
   std::unique_ptr<PeriodicTaskManager> periodicTaskManager_;
   std::unique_ptr<PrestoServerOperations> prestoServerOperations_;
   std::unique_ptr<PeriodicMemoryChecker> memoryChecker_;
-  bool isMemOverloaded_{false};
-  bool isCpuOverloaded_{false};
+
+  // Last known memory overloaded status.
+  bool memOverloaded_{false};
+  // Last known CPU overloaded status.
+  bool cpuOverloaded_{false};
+  // Current worker overloaded status. It can still be true when memory and CPU
+  // overloaded flags are not due to cooldown period.
+  bool serverOverloaded_{false};
+  // Last time point (in seconds) when the worker was overloaded.
+  uint64_t lastOverloadedTimeInSecs_{0};
 
   // We update these members asynchronously and return in http requests w/o
   // delay.

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

@@ -24,6 +24,25 @@ using namespace facebook::velox;
 
 namespace facebook::presto {
 
+std::string prestoTaskStateString(PrestoTaskState state) {
+  switch (state) {
+    case PrestoTaskState::kRunning:
+      return "Running";
+    case PrestoTaskState::kFinished:
+      return "Finished";
+    case PrestoTaskState::kCanceled:
+      return "Canceled";
+    case PrestoTaskState::kAborted:
+      return "Aborted";
+    case PrestoTaskState::kFailed:
+      return "Failed";
+    case PrestoTaskState::kPlanned:
+      return "Planned";
+    default:
+      return fmt::format("UNKNOWN[{}]", static_cast<int>(state));
+  }
+}
+
 namespace {
 
 #define TASK_STATS_SUM(taskStats, statsName, taskStatusSum)      \
@@ -37,7 +56,25 @@ namespace {
     }                                                            \
   } while (0)
 
-protocol::TaskState toPrestoTaskState(exec::TaskState state) {
+// Convert Velox task state to Presto task state.
+PrestoTaskState toPrestoTaskState(exec::TaskState state) {
+  switch (state) {
+    case exec::TaskState::kRunning:
+      return PrestoTaskState::kRunning;
+    case exec::TaskState::kFinished:
+      return PrestoTaskState::kFinished;
+    case exec::TaskState::kCanceled:
+      return PrestoTaskState::kCanceled;
+    case exec::TaskState::kAborted:
+      return PrestoTaskState::kAborted;
+    case exec::TaskState::kFailed:
+      return PrestoTaskState::kFailed;
+  }
+  // Should not be here.
+  return PrestoTaskState::kAborted;
+}
+
+protocol::TaskState toProtocolTaskState(exec::TaskState state) {
   switch (state) {
     case exec::TaskState::kRunning:
       return protocol::TaskState::RUNNING;
@@ -466,6 +503,21 @@ PrestoTask::PrestoTask(
   createTimeMs = getCurrentTimeMs();
 }
 
+PrestoTaskState PrestoTask::taskState() const {
+  if (task != nullptr) {
+    const auto prestoTaskState = toPrestoTaskState(task->state());
+    // Velox Task is created with 'Running' state even though it is not running
+    // until start() is called. Here we check for this and return 'Planned'
+    // state if it is the case.
+    if (prestoTaskState == PrestoTaskState::kRunning && !taskStarted) {
+      return PrestoTaskState::kPlanned;
+    }
+    return prestoTaskState;
+  }
+  // Fallback to 'aborted' if there is no Velox task.
+  return PrestoTaskState::kAborted;
+}
+
 void PrestoTask::updateHeartbeatLocked() {
   lastHeartbeatMs = velox::getCurrentTimeMs();
   info.lastHeartbeatInMillis = lastHeartbeatMs;
@@ -527,7 +579,7 @@ protocol::TaskStatus PrestoTask::updateStatusLocked() {
 
   const auto veloxTaskStats = task->taskStats();
 
-  info.taskStatus.state = toPrestoTaskState(task->state());
+  info.taskStatus.state = toProtocolTaskState(task->state());
 
   // Presto has a Driver per split. When splits represent partitions
   // of data, there is a queue of them per Task. We represent
@@ -856,14 +908,14 @@ void PrestoTask::updateExecutionInfoLocked(
 }
 
 /*static*/ std::string PrestoTask::taskStatesToString(
-    const std::array<size_t, 5>& taskStates) {
+    const std::array<size_t, 6>& taskStates) {
   std::string str;
   for (size_t i = 0; i < taskStates.size(); ++i) {
     if (taskStates[i] != 0) {
       folly::toAppend(
           fmt::format(
               "{}={} ",
-              velox::exec::taskStateString(velox::exec::TaskState(i)),
+              prestoTaskStateString(PrestoTaskState(i)),
               taskStates[i]),
           &str);
     }

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

@@ -25,6 +25,19 @@ struct RuntimeMetric;
 
 namespace facebook::presto {
 
+/// Velox Task does not have Planned state, so we add this enum to have this
+/// state.
+enum class PrestoTaskState : int {
+  kRunning = 0,
+  kFinished = 1,
+  kCanceled = 2,
+  kAborted = 3,
+  kFailed = 4,
+  kPlanned = 5
+};
+
+std::string prestoTaskStateString(PrestoTaskState state);
+
 template <typename T>
 struct PromiseHolder {
   explicit PromiseHolder(folly::Promise<T> p) : promise(std::move(p)) {}
@@ -139,6 +152,10 @@ struct PrestoTask {
       const std::string& nodeId,
       long startProcessCpuTime = 0);
 
+  /// Returns current task state, including 'planning'.
+  /// If Velox task is null, it returns 'aborted'.
+  PrestoTaskState taskState() const;
+
   /// Updates when this task was touched last time.
   void updateHeartbeatLocked();
 
@@ -169,7 +186,7 @@ struct PrestoTask {
 
   /// Turns the task numbers (per state) into a string.
   static std::string taskStatesToString(
-      const std::array<size_t, 5>& taskStates);
+      const std::array<size_t, 6>& taskStates);
 
   /// Invoked to update presto task status from the updated velox task stats.
   protocol::TaskStatus updateStatusLocked();