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

Code before:
  RAY_CHECK(commands_sync_message.SerializeToString(&serialized_commands_sync_msg));

  // Populate the sync message.
  syncer::RaySyncMessage msg;
  msg.set_version(gc_command_sync_version_);
  msg.set_node_id(self_node_id_.Binary());
  msg.set_message_type(syncer::MessageType::COMMANDS);
  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"

Review comment (suggestion):
I would use `auto` only if you think it makes the code more readable. I prefer explicit types even if they're more annoying to write out. One reason is so I can grep for usages of a class. 

https://google.github.io/styleguide/cppguide.html#Type_deduction

Code after:

  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.
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);