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

Code before:

void LocalObjectManager::SpillObjectUptoMaxThroughput() {
  if (RayConfig::instance().object_spilling_config().empty()) {
    return;
  }

  // Spill as fast as we can using all our spill workers.
  bool can_spill_more = true;
  while (can_spill_more) {
    if (!TryToSpillObjects()) {
      break;
    }
    can_spill_more = num_active_workers_ < max_active_workers_;
  }
}

bool LocalObjectManager::IsSpillingInProgress() { return num_active_workers_ > 0; }

bool LocalObjectManager::TryToSpillObjects() {
  if (RayConfig::instance().object_spilling_config().empty()) {
    return false;
  }

  RAY_LOG(DEBUG) << "Choosing objects to spill with minimum total size "
                 << min_spilling_size_ << ", max fused file size "
                 << max_spilling_file_size_
                 << " or with total # of objects = " << max_fused_object_count_;
  int64_t bytes_to_spill = 0;
  std::vector<ObjectID> objects_to_spill;
  int64_t num_to_spill = 0;
  size_t idx = 0;
  for (const auto &[object_id, ray_object] : pinned_objects_) {
    if (is_plasma_object_spillable_(object_id)) {
      const int64_t object_size = ray_object->GetSize();

      // If the max file size limit is enabled, avoid fusing more objects once we'd exceed
      // it. Always allow spilling at least one object, even if it's larger than the
      // limit.
      if (max_spilling_file_size_ > 0 && !objects_to_spill.empty() &&
          bytes_to_spill + object_size > max_spilling_file_size_) {
        break;
      }

      bytes_to_spill += object_size;
      objects_to_spill.push_back(object_id);
      ++num_to_spill;
      if (num_to_spill == max_fused_object_count_) {
        break;
      }
    }
    ++idx;

Review comment (suggestion):
If max_spilling_file_size is left as default, the -1 might be confusing in this log message. One idea could be `max_spilling_file_size <= 0 ? "max" : max_spilling_file_size`.

Code after:
}

void LocalObjectManager::SpillObjectUptoMaxThroughput() {
  if (RayConfig::instance().object_spilling_config().empty()) {
    return;
  }

  // Spill as fast as we can using all our spill workers.
  bool can_spill_more = true;
  while (can_spill_more) {
    if (!TryToSpillObjects()) {
      break;
    }
    can_spill_more = num_active_workers_ < max_active_workers_;
  }
}

bool LocalObjectManager::IsSpillingInProgress() { return num_active_workers_ > 0; }

bool LocalObjectManager::TryToSpillObjects() {
  if (RayConfig::instance().object_spilling_config().empty()) {
    return false;
  }

  int64_t bytes_to_spill = 0;
  std::vector<ObjectID> objects_to_spill;
  int64_t num_to_spill = 0;
  size_t idx = 0;
  for (const auto &[object_id, ray_object] : pinned_objects_) {
    if (is_plasma_object_spillable_(object_id)) {
      const int64_t object_size = ray_object->GetSize();

      // If the max file size limit is enabled, avoid fusing more objects once we'd exceed
      // it. Always allow spilling at least one object, even if it's larger than the
      // limit.
      if (max_spilling_file_size_bytes_ > 0 && !objects_to_spill.empty() &&
          bytes_to_spill + object_size > max_spilling_file_size_bytes_) {
        break;
      }

      bytes_to_spill += object_size;
      objects_to_spill.push_back(object_id);
      ++num_to_spill;
      if (num_to_spill == max_fused_object_count_) {
        break;
      }
    }
    ++idx;
  }
  if (objects_to_spill.empty()) {
    return false;