[native] Remove arbitrator stats reporting

presto-native-execution/presto_cpp/main/CMakeLists.txt

@@ -50,6 +50,7 @@ target_link_libraries(
   presto_operators
   velox_aggregates
   velox_caching
+  velox_common_base
   velox_core
   velox_dwio_common_exception
   velox_encode

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

@@ -20,6 +20,7 @@
 #include "presto_cpp/main/TaskManager.h"
 #include "presto_cpp/main/common/Counters.h"
 #include "presto_cpp/main/http/filters/HttpEndpointLatencyFilter.h"
+#include "velox/common/base/PeriodicStatsReporter.h"
 #include "velox/common/base/StatsReporter.h"
 #include "velox/common/base/SuccinctPrinter.h"
 #include "velox/common/caching/AsyncDataCache.h"
@@ -88,8 +89,6 @@ static constexpr size_t kConnectorPeriodGlobalCounters{
     60'000'000}; // 60 seconds.
 static constexpr size_t kOsPeriodGlobalCounters{2'000'000}; // 2 seconds
 static constexpr size_t kSpillStatsUpdateIntervalUs{60'000'000}; // 60 seconds
-static constexpr size_t kArbitratorStatsUpdateIntervalUs{
-    60'000'000}; // 60 seconds
 // Every 1 minute we print endpoint latency counters.
 static constexpr size_t kHttpEndpointLatencyPeriodGlobalCounters{
     60'000'000}; // 60 seconds.
@@ -114,6 +113,11 @@ PeriodicTaskManager::PeriodicTaskManager(
       server_(server) {}
 
 void PeriodicTaskManager::start() {
+  VELOX_CHECK_NOT_NULL(arbitrator_);
+  velox::PeriodicStatsReporter::Options opts;
+  opts.arbitrator = arbitrator_->kind() == "NOOP" ? nullptr : arbitrator_;
+  velox::startPeriodicStatsReporter(opts);
+
   // If executors are null, don't bother starting this task.
   if ((driverCPUExecutor_ != nullptr) || (httpExecutor_ != nullptr)) {
     addExecutorStatsTask();
@@ -146,11 +150,6 @@ void PeriodicTaskManager::start() {
     addHttpEndpointLatencyStatsTask();
   }
 
-  VELOX_CHECK_NOT_NULL(arbitrator_);
-  if (arbitrator_->kind() != "NOOP") {
-    addArbitratorStatsTask();
-  }
-
   if (server_ && server_->hasCoordinatorDiscoverer()) {
     numDriverThreads_ = server_->numDriverThreads();
     addWatchdogTask();
@@ -160,6 +159,7 @@ void PeriodicTaskManager::start() {
 }
 
 void PeriodicTaskManager::stop() {
+  velox::stopPeriodicStatsReporter();
   oneTimeRunner_.cancelAllFunctionsAndWait();
   oneTimeRunner_.shutdown();
   repeatedRunner_.stop();
@@ -610,50 +610,6 @@ void PeriodicTaskManager::addOperatingSystemStatsUpdateTask() {
       "os_counters");
 }
 
-void PeriodicTaskManager::addArbitratorStatsTask() {
-  addTask(
-      [this]() { updateArbitratorStatsTask(); },
-      kArbitratorStatsUpdateIntervalUs,
-      "arbitrator_stats");
-}
-
-void PeriodicTaskManager::updateArbitratorStatsTask() {
-  const auto updatedArbitratorStats = arbitrator_->stats();
-  VELOX_CHECK_GE(updatedArbitratorStats, lastArbitratorStats_);
-  const auto deltaArbitratorStats =
-      updatedArbitratorStats - lastArbitratorStats_;
-  REPORT_IF_NOT_ZERO(
-      kCounterArbitratorNumRequests, deltaArbitratorStats.numRequests);
-  REPORT_IF_NOT_ZERO(
-      kCounterArbitratorNumAborted, deltaArbitratorStats.numAborted);
-  REPORT_IF_NOT_ZERO(
-      kCounterArbitratorNumFailures, deltaArbitratorStats.numFailures);
-  REPORT_IF_NOT_ZERO(
-      kCounterArbitratorQueueTimeUs, deltaArbitratorStats.queueTimeUs);
-  REPORT_IF_NOT_ZERO(
-      kCounterArbitratorArbitrationTimeUs,
-      deltaArbitratorStats.arbitrationTimeUs);
-  REPORT_IF_NOT_ZERO(
-      kCounterArbitratorNumShrunkBytes, deltaArbitratorStats.numShrunkBytes);
-  REPORT_IF_NOT_ZERO(
-      kCounterArbitratorNumReclaimedBytes,
-      deltaArbitratorStats.numReclaimedBytes);
-  REPORT_IF_NOT_ZERO(
-      kCounterArbitratorFreeCapacityBytes,
-      deltaArbitratorStats.freeCapacityBytes);
-  REPORT_IF_NOT_ZERO(
-      kCounterArbitratorNonReclaimableAttempts,
-      deltaArbitratorStats.numNonReclaimableAttempts);
-
-  if (!deltaArbitratorStats.empty()) {
-    LOG(INFO) << "Updated memory arbitrator stats: "
-              << updatedArbitratorStats.toString();
-    LOG(INFO) << "Memory arbitrator stats change: "
-              << deltaArbitratorStats.toString();
-  }
-  lastArbitratorStats_ = updatedArbitratorStats;
-}
-
 void PeriodicTaskManager::addSpillStatsUpdateTask() {
   addTask(
       [this]() { updateSpillStatsTask(); },

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

@@ -53,10 +53,6 @@ class PeriodicTaskManager {
           std::shared_ptr<velox::connector::Connector>>& connectors,
       PrestoServer* server);
 
-  ~PeriodicTaskManager() {
-    stop();
-  }
-
   /// Invoked to start all registered, and fundamental periodic tasks running at
   /// the background.
   ///
@@ -170,7 +166,6 @@ class PeriodicTaskManager {
   int64_t lastForcedContextSwitches_{0};
   // Renabled this after update velox.
   velox::common::SpillStats lastSpillStats_;
-  velox::memory::MemoryArbitrator::Stats lastArbitratorStats_;
 
   // NOTE: declare last since the threads access other members of `this`.
   folly::FunctionScheduler oneTimeRunner_;

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

@@ -244,23 +244,6 @@ void registerPrestoMetrics() {
       100);
   DEFINE_METRIC(kCounterSpillMaxLevelExceeded, facebook::velox::StatType::SUM);
 
-  /// ================== Memory Arbitrator Counters =================
-
-  DEFINE_METRIC(kCounterArbitratorNumRequests, facebook::velox::StatType::SUM);
-  DEFINE_METRIC(kCounterArbitratorNumAborted, facebook::velox::StatType::SUM);
-  DEFINE_METRIC(kCounterArbitratorNumFailures, facebook::velox::StatType::SUM);
-  DEFINE_METRIC(kCounterArbitratorQueueTimeUs, facebook::velox::StatType::SUM);
-  DEFINE_METRIC(
-      kCounterArbitratorArbitrationTimeUs, facebook::velox::StatType::SUM);
-  DEFINE_METRIC(
-      kCounterArbitratorNumShrunkBytes, facebook::velox::StatType::SUM);
-  DEFINE_METRIC(
-      kCounterArbitratorNumReclaimedBytes, facebook::velox::StatType::SUM);
-  DEFINE_METRIC(
-      kCounterArbitratorFreeCapacityBytes, facebook::velox::StatType::AVG);
-  DEFINE_METRIC(
-      kCounterArbitratorNonReclaimableAttempts, facebook::velox::StatType::SUM);
-
   // NOTE: Metrics type exporting for file handle cache counters are in
   // PeriodicTaskManager because they have dynamic names. The following counters
   // have their type exported there:

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

@@ -165,37 +165,6 @@ constexpr folly::StringPiece kCounterMmapRawAllocBytesSmall{
 constexpr folly::StringPiece kCounterExchangeSourcePeakQueuedBytes{
     "presto_cpp.exchange_source_peak_queued_bytes"};
 
-/// ================== Memory Arbitrator Counters =================
-
-/// The number of arbitration requests.
-constexpr folly::StringPiece kCounterArbitratorNumRequests{
-    "presto_cpp.arbitrator_num_requests"};
-/// The number of aborted arbitration requests.
-constexpr folly::StringPiece kCounterArbitratorNumAborted{
-    "presto_cpp.arbitrator_num_aborted"};
-/// The number of arbitration request failures.
-constexpr folly::StringPiece kCounterArbitratorNumFailures{
-    "presto_cpp.arbitrator_num_failures"};
-/// The sum of all the arbitration request queue times in microseconds.
-constexpr folly::StringPiece kCounterArbitratorQueueTimeUs{
-    "presto_cpp.arbitrator_queue_time_us"};
-/// The sum of all the arbitration run times in microseconds.
-constexpr folly::StringPiece kCounterArbitratorArbitrationTimeUs{
-    "presto_cpp.arbitrator_arbitration_time_us"};
-/// The amount of memory bytes freed by reducing the memory pool's capacity
-/// without actually freeing memory.
-constexpr folly::StringPiece kCounterArbitratorNumShrunkBytes{
-    "presto_cpp.arbitrator_num_shrunk_bytes"};
-/// The amount of memory bytes freed by memory reclamation.
-constexpr folly::StringPiece kCounterArbitratorNumReclaimedBytes{
-    "presto_cpp.arbitrator_num_reclaimed_bytes"};
-/// The free memory capacity in bytes.
-constexpr folly::StringPiece kCounterArbitratorFreeCapacityBytes{
-    "presto_cpp.arbitrator_free_capacity_bytes"};
-/// The number of non-reclaimable operator reclaim attempts.
-constexpr folly::StringPiece kCounterArbitratorNonReclaimableAttempts{
-    "presto_cpp.arbitrator_non_reclaimable_attempts"};
-
 /// ================== Disk Spilling Counters =================
 
 /// The number of times that spilling runs on a velox operator.