update

Signed-off-by: You-Cheng Lin <mses010108@gmail.com>

Author: owenowenisme

src/ray/common/event_memory_monitor.cc

@@ -192,7 +192,8 @@ void EventMemoryMonitor::MonitoringThreadMain() {
 
       if (high_modified && IsEnabled()) {
         Disable();
-        kill_workers_callback_();
+        kill_workers_callback_(
+            "cgroup memory.events 'high' counter increased (event-driven)");
       }
     } else {
       RAY_LOG(ERROR) << absl::StrFormat(

src/ray/common/memory_monitor_interface.h

@@ -19,6 +19,7 @@
 #include <cstdint>
 #include <functional>
 #include <ostream>
+#include <string>
 
 #include "absl/container/flat_hash_map.h"
 #include "ray/util/compat.h"
@@ -49,8 +50,10 @@ using ProcessesMemorySnapshot = absl::flat_hash_map<pid_t, int64_t>;
 
 /**
  * @brief Callback to trigger worker oom killing when under memory pressure.
+ * @param trigger_reason A human-readable description of why the monitor triggered
+ *        the kill (e.g. threshold exceeded, cgroup event, PSI pressure).
  */
-using KillWorkersCallback = std::function<void()>;
+using KillWorkersCallback = std::function<void(const std::string &trigger_reason)>;
 
 /**
  * @brief implementations of this interface monitors the memory usage of the node

src/ray/common/pressure_memory_monitor.cc

@@ -161,7 +161,8 @@ void PressureMemoryMonitor::MonitoringThreadMain() {
     if (fds[0].revents & POLLPRI) {
       if (IsEnabled()) {
         Disable();
-        kill_workers_callback_();
+        kill_workers_callback_(
+            "cgroup memory.pressure PSI trigger fired (pressure-driven)");
       }
     } else if (fds[0].revents & POLLERR) {
       RAY_LOG(ERROR) << "Got POLLERR while monitoring memory pressure. "

src/ray/common/tests/event_memory_monitor_test.cc

@@ -98,7 +98,9 @@ TEST_F(EventMemoryMonitorTest, TestCallbackCalledWhenHighEventChanges) {
   WriteMemoryEventsFile(events_file_->GetPath(), 0, 0);
 
   auto callback_latch = std::make_shared<boost::latch>(1);
-  KillWorkersCallback callback = [callback_latch]() { callback_latch->count_down(); };
+  KillWorkersCallback callback = [callback_latch](const std::string &) {
+    callback_latch->count_down();
+  };
 
   StatusSetOr<std::unique_ptr<EventMemoryMonitor>, StatusT::IOError> result =
       EventMemoryMonitor::Create(std::move(mock_cgroup_dir_->GetPath()),
@@ -116,7 +118,9 @@ TEST_F(EventMemoryMonitorTest, TestNoCallbackWhenValuesUnchanged) {
   WriteMemoryEventsFile(events_file_->GetPath(), 0, 0);
 
   auto callback_latch = std::make_shared<boost::latch>(1);
-  KillWorkersCallback callback = [callback_latch]() { callback_latch->count_down(); };
+  KillWorkersCallback callback = [callback_latch](const std::string &) {
+    callback_latch->count_down();
+  };
 
   StatusSetOr<std::unique_ptr<EventMemoryMonitor>, StatusT::IOError> result =
       EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), callback);
@@ -133,7 +137,9 @@ TEST_F(EventMemoryMonitorTest, TestNoCallbackWhenIrrelevantEventChanges) {
   WriteMemoryEventsFile(events_file_->GetPath(), 0, 0);
 
   auto callback_latch = std::make_shared<boost::latch>(1);
-  KillWorkersCallback callback = [callback_latch]() { callback_latch->count_down(); };
+  KillWorkersCallback callback = [callback_latch](const std::string &) {
+    callback_latch->count_down();
+  };
 
   StatusSetOr<std::unique_ptr<EventMemoryMonitor>, StatusT::IOError> result =
       EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), callback);
@@ -153,16 +159,17 @@ TEST_F(EventMemoryMonitorTest, TestMultipleCallbacksOnMultipleChanges) {
   auto latch2 = std::make_shared<boost::latch>(1);
   auto latch3 = std::make_shared<boost::latch>(1);
   std::atomic<int> callback_count{0};
-  KillWorkersCallback callback = [&callback_count, latch1, latch2, latch3]() {
-    int count = ++callback_count;
-    if (count == 1) {
-      latch1->count_down();
-    } else if (count == 2) {
-      latch2->count_down();
-    } else if (count == 3) {
-      latch3->count_down();
-    }
-  };
+  KillWorkersCallback callback =
+      [&callback_count, latch1, latch2, latch3](const std::string &) {
+        int count = ++callback_count;
+        if (count == 1) {
+          latch1->count_down();
+        } else if (count == 2) {
+          latch2->count_down();
+        } else if (count == 3) {
+          latch3->count_down();
+        }
+      };
 
   StatusSetOr<std::unique_ptr<EventMemoryMonitor>, StatusT::IOError> result =
       EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), callback);

src/ray/common/tests/pressure_memory_monitor_test.cc

@@ -162,7 +162,9 @@ TEST_F(PressureMemoryMonitorTest,
   close(listener);
 
   std::shared_ptr<boost::latch> has_called_once = std::make_shared<boost::latch>(1);
-  auto kill_workers_callback = [has_called_once]() { has_called_once->count_down(); };
+  auto kill_workers_callback = [has_called_once](const std::string &) {
+    has_called_once->count_down();
+  };
   std::unique_ptr<PressureMemoryMonitor> monitor =
       std::make_unique<PressureMemoryMonitor>(
           mock_cgroup_dir_->GetPath(), listener_fd, kill_workers_callback);

src/ray/common/tests/threshold_memory_monitor_test.cc

@@ -54,7 +54,7 @@ TEST_F(ThresholdMemoryMonitorTest, TestMonitorTriggerCanDetectMemoryUsage) {
   MakeThresholdMemoryMonitor(
       0 /*memory_usage_threshold_bytes*/,
       1 /*refresh_interval_ms*/,
-      [has_checked_once]() { has_checked_once->count_down(); },
+      [has_checked_once](const std::string &) { has_checked_once->count_down(); },
       "" /*root_cgroup_path*/);
   has_checked_once->wait();
 }
@@ -78,7 +78,7 @@ TEST_F(ThresholdMemoryMonitorTest,
   MakeThresholdMemoryMonitor(
       memory_usage_threshold_bytes,  // (70%)
       1 /*refresh_interval_ms*/,
-      [has_checked_once]() { has_checked_once->count_down(); },
+      [has_checked_once](const std::string &) { has_checked_once->count_down(); },
       cgroup_dir /*root_cgroup_path*/);
 
   has_checked_once->wait();

src/ray/common/threshold_memory_monitor.cc

@@ -60,7 +60,14 @@ ThresholdMemoryMonitor::ThresholdMemoryMonitor(KillWorkersCallback kill_workers_
 
         if (is_usage_above_threshold && IsEnabled()) {
           Disable();
-          kill_workers_callback_();
+          std::string trigger_reason = absl::StrFormat(
+              "Memory usage %dB exceeded threshold of %dB (%.1f%% of %dB total)",
+              cur_memory_snapshot.used_bytes,
+              memory_usage_threshold_bytes_,
+              static_cast<float>(memory_usage_threshold_bytes_) /
+                  static_cast<float>(cur_memory_snapshot.total_bytes) * 100,
+              cur_memory_snapshot.total_bytes);
+          kill_workers_callback_(trigger_reason);
         }
       },
       monitor_interval_ms,

src/ray/raylet/node_manager.cc

@@ -3068,12 +3068,12 @@ void NodeManager::ReleaseKillWorkerInProgress() {
 
 // Picks the workers and kills the process if the memory usage is above the threshold.
 KillWorkersCallback NodeManager::CreateKillWorkersCallback() {
-  return [this]() {
+  return [this](const std::string &trigger_reason) {
     if (!MarkKillWorkerInProgress()) {
       return;
     }
     io_service_.post(
-        [this]() {
+        [this, trigger_reason]() {
           std::vector<std::shared_ptr<WorkerInterface>> workers =
               worker_pool_.GetAllRegisteredWorkers(/* filter_dead_workers */ true,
                                                    /* filter_io_workers */ true);
@@ -3100,25 +3100,13 @@ KillWorkersCallback NodeManager::CreateKillWorkersCallback() {
             return;
           }
 
-          // Compute the memory usage threshold
-          int64_t total_memory_bytes = system_memory_snapshot.total_bytes;
-          int64_t computed_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold(
-              total_memory_bytes,
-              RayConfig::instance().memory_usage_threshold(),
-              RayConfig::instance().min_memory_free_bytes(),
-              initial_config_.enable_resource_isolation,
-              *cgroup_manager_);
-          float computed_threshold_fraction =
-              static_cast<float>(computed_threshold_bytes) /
-              static_cast<float>(total_memory_bytes);
-
           std::string oom_kill_details = CreateOomKillMessageDetails(
               workers_to_kill_and_should_retry,
               self_node_id_,
               system_memory_snapshot,
               store_client_->GetMemoryUsage().value_or("Not available"),
               process_memory_snapshot,
-              computed_threshold_fraction);
+              trigger_reason);
           std::string oom_kill_suggestions =
               CreateOomKillMessageSuggestions(workers_to_kill_and_should_retry);
 
@@ -3188,7 +3176,7 @@ std::string NodeManager::CreateOomKillMessageDetails(
     const SystemMemorySnapshot &system_memory_snapshot,
     const std::string &object_store_memory_usage,
     const ProcessesMemorySnapshot &process_memory_snapshot,
-    float usage_threshold) const {
+    const std::string &trigger_reason) const {
   if (workers_to_kill.empty()) {
     return "";
   }
@@ -3246,8 +3234,8 @@ std::string NodeManager::CreateOomKillMessageDetails(
   }
 
   return absl::StrFormat(
-      "Memory on the node (IP: %s, ID: %s) was %sGB / %sGB (%f), "
-      "which exceeds the memory usage threshold of %f; "
+      "Memory on the node (IP: %s, ID: %s) was %sGB / %sGB (%f); "
+      "Trigger: %s; "
       "Object store memory usage: [%s]; "
       "Ray killed %d worker(s) based on the killing policy: "
       "[%s]; "
@@ -3259,7 +3247,7 @@ std::string NodeManager::CreateOomKillMessageDetails(
       used_bytes_gb,
       total_bytes_gb,
       usage_fraction,
-      usage_threshold,
+      trigger_reason,
       absl::StrReplaceAll(object_store_memory_usage, {{"\n", "; "}}),
       workers_to_kill.size(),
       absl::StrJoin(worker_details, "; "),

src/ray/raylet/node_manager.h

@@ -786,7 +786,7 @@ class NodeManager : public rpc::NodeManagerServiceHandler,
    * @param node_id The ID of the node.
    * @param system_memory_snapshot The snapshot of the system memory.
    * @param process_memory_snapshot The snapshot of the process memory.
-   * @param usage_threshold The memory limit.
+   * @param trigger_reason The reason the memory monitor triggered the kill.
    * @return The detail message for the workers that are killed due to memory running low.
    */
   std::string CreateOomKillMessageDetails(
@@ -796,7 +796,7 @@ class NodeManager : public rpc::NodeManagerServiceHandler,
       const SystemMemorySnapshot &system_memory_snapshot,
       const std::string &object_store_memory_usage,
       const ProcessesMemorySnapshot &process_memory_snapshot,
-      float usage_threshold) const;
+      const std::string &trigger_reason) const;
 
   /**
    * @param workers_to_kill The workers to print the kill suggestions for.