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

Code before:
    int64_t after_version, syncer::MessageType message_type) const {
  // This method is only called for the COMMANDS channel, as the RESOURCE_VIEW
  // channel goes through the LocalResourceManager.
  RAY_CHECK_EQ(message_type, syncer::MessageType::COMMANDS);

  // Serialize the COMMANDS message to a byte string to be nested inside the sync message.
  std::string serialized_commands_sync_msg;
  syncer::CommandsSyncMessage commands_sync_message;
  commands_sync_message.set_should_global_gc(true);
  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.

Review comment (suggestion):
Maybe make this comment a little stronger. 

NOTE: The callback is NOT thread-safe and must be run on the io_service thread. 

Code after:
  commands_sync_message.set_should_global_gc(true);
  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.
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."