Repository: ray-project/ray
File: src/ray/raylet/node_manager.cc (C++)

Code before:
        worker_killing_policy_->SelectWorkersToKill(workers, system_memory);
    for (std::pair<std::shared_ptr<WorkerInterface>, bool>
             worker_to_kill_and_should_retry : workers_to_kill_and_should_retry) {
      std::shared_ptr<WorkerInterface> worker_to_kill =
          worker_to_kill_and_should_retry.first;
      bool should_retry = worker_to_kill_and_should_retry.second;
      /// TODO: (clarng) expose these strings in the frontend python error as well.
      std::string oom_kill_details = this->CreateOomKillMessageDetails(
          worker_to_kill, this->self_node_id_, system_memory, usage_threshold);
      std::string oom_kill_suggestions =
          this->CreateOomKillMessageSuggestions(worker_to_kill, should_retry);

      RAY_LOG(INFO)
          << "Killing worker with task "
          << worker_to_kill->GetGrantedLease().GetLeaseSpecification().DebugString()
          << "\n\n"
          << oom_kill_details << "\n\n"
          << oom_kill_suggestions;

      std::stringstream worker_exit_message_ss;
      worker_exit_message_ss << "Task was killed due to the node running low on memory.\n"
                             << oom_kill_details << "\n"
                             << oom_kill_suggestions;
      std::string worker_exit_message = worker_exit_message_ss.str();

      // Rerpot the event to the dashboard.
      RAY_EVENT_EVERY_MS(ERROR, "Out of Memory", 10 * 1000) << worker_exit_message;

      // Mark the worker as failure and raise an exception from a caller.
      rpc::RayErrorInfo worker_failure_reason;
      worker_failure_reason.set_error_message(worker_exit_message);
      worker_failure_reason.set_error_type(rpc::ErrorType::OUT_OF_MEMORY);
      SetWorkerFailureReason(
          worker_to_kill->GetGrantedLeaseId(), worker_failure_reason, should_retry);

      /// since we print the process memory in the message. Destroy should be called
      /// as soon as possible to free up memory.
      DestroyWorker(worker_to_kill,
                    rpc::WorkerExitType::NODE_OUT_OF_MEMORY,
                    worker_exit_message,
                    true /* force */);

      // Record worker killing metrics.
      if (worker_to_kill->GetWorkerType() == rpc::WorkerType::DRIVER) {
        // TODO(sang): Add the job entrypoint to the name.
        memory_manager_worker_eviction_total_gauge_.Record(
            1, {{"Type", "MemoryManager.DriverEviction.Total"}, {"Name", ""}});
      } else if (worker_to_kill->GetActorId().IsNil()) {
        const auto &ray_lease = worker_to_kill->GetGrantedLease();
        memory_manager_worker_eviction_total_gauge_.Record(
            1,

Review comment (suggestion):
```suggestion
      // Report the event to the dashboard.
```

Code after:
              MemoryMonitorUtils::TakePerProcessMemorySnapshot();
          std::vector<std::shared_ptr<WorkerInterface>> workers =
              worker_pool_.GetAllRegisteredWorkers();
          if (workers.empty()) {
            RAY_LOG_EVERY_MS(WARNING, 5000)
                << "Memory usage above threshold but no workers are available for "
                   "killing."
                << "This could be due to worker memory leak and"
                << "idle worker are occupying most of the memory.";
            memory_monitor_->Enable();
            return;
          }
          std::pair<std::shared_ptr<WorkerInterface>, bool>
              worker_to_kill_and_should_retry =
                  worker_killing_policy_->SelectWorkerToKill(workers,
                                                             process_memory_snapshot);
          std::shared_ptr<WorkerInterface> worker_to_kill =
              worker_to_kill_and_should_retry.first;
          bool should_retry = worker_to_kill_and_should_retry.second;
          if (worker_to_kill == nullptr) {
            RAY_LOG_EVERY_MS(WARNING, 5000) << "Worker killer did not select a worker to "
                                               "kill even though memory usage is high.";
          } else {
            // Compute the memory usage threshold
            int64_t total_memory_bytes = system_memory.total_bytes;
            int64_t computed_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold(
                total_memory_bytes,
                RayConfig::instance().memory_usage_threshold(),
                RayConfig::instance().min_memory_free_bytes());
            float computed_threshold_fraction =
                static_cast<float>(computed_threshold_bytes) /
                static_cast<float>(total_memory_bytes);

            worker_being_killed_ = worker_to_kill;

            std::string oom_kill_details =
                CreateOomKillMessageDetails(worker_to_kill,
                                            self_node_id_,
                                            system_memory,
                                            process_memory_snapshot,
                                            computed_threshold_fraction);
            std::string oom_kill_suggestions =
                CreateOomKillMessageSuggestions(worker_to_kill, should_retry);

            RAY_LOG(INFO) << absl::StrFormat(
                "Killing worker with task %s, kill details: %s, suggestions: %s",
                worker_to_kill->GetGrantedLease().GetLeaseSpecification().DebugString(),
                oom_kill_details,
                oom_kill_suggestions);

            std::string worker_exit_message = absl::StrFormat(