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

Code before:
  msg.set_sync_message(std::move(serialized_commands_sync_msg));

  return std::make_optional(std::move(msg));
}

// Picks the worker with the latest submitted task and kills the process
// if the memory usage is above the threshold. Allows one in-flight
// process kill at a time as killing a process could sometimes take
// seconds.
// Assumes that the callback is running on the node manager's io_service thread.
MemoryUsageRefreshCallback NodeManager::CreateMemoryUsageRefreshCallback() {
  return [this](bool is_usage_above_threshold,
                MemorySnapshot system_memory,
                float usage_threshold) {
    if (!is_usage_above_threshold) {
      return;
    }
    system_memory.process_used_bytes = MemoryMonitor::GetProcessMemoryUsage();
    auto 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.";
      return;
    }
    auto workers_to_kill_and_should_retry =
        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;

Review comment (suggestion):
I don't think we need to log here. We can have high memory usage (through plasma) even if there are no workers. This is a little too ad-hoc for leak detection. 

If anything, this tells us that the node is only serving as a data store for plasma and nothing else. This would be good motivation for primary copy migration, but that's still a little further down the road. 

Code after:
// seconds.
KillWorkersCallback NodeManager::CreateKillWorkersCallback() {
  return [this](const SystemMemorySnapshot &system_memory_snapshot) {
    io_service_.post(
        [this, system_memory = system_memory_snapshot]() {
          if (worker_being_killed_ != nullptr) {
            // If the worker previously being killed has not died yet, wait until it is
            // killed before selecting a new worker to prevent double killing.
            if (worker_being_killed_->GetProcess().IsAlive()) {
              RAY_LOG_EVERY_MS(INFO, 1000)
                      .WithField(worker_being_killed_->GetGrantedLeaseId())
                      .WithField(worker_being_killed_->WorkerId())
                  << "Memory usage above threshold. "
                  << "Still waiting for worker eviction to free up memory. "
                  << "worker pid: " << worker_being_killed_->GetProcess().GetId();
              memory_monitor_->Enable();
              return;
            } else {
              RAY_LOG(INFO)
                      .WithField(worker_being_killed_->WorkerId())
                      .WithField(worker_being_killed_->GetGrantedLeaseId())
                  << "Worker evicted and process killed to reclaim memory. "
                  << "worker pid: " << worker_being_killed_->GetProcess().GetId();
              worker_being_killed_ = nullptr;
            }
          }

          ProcessesMemorySnapshot process_memory_snapshot =
              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 {