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

Code before:
  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;
  }

  const std::string max_spilling_file_size_str =
      max_spilling_file_size_ <= 0
          ? "unlimited"
          : absl::StrFormat("%lld", static_cast<long long>(max_spilling_file_size_));
  RAY_LOG(DEBUG) << absl::StrFormat(
      "Choosing objects to spill with minimum total size %lld, max fused file size %s "
      "or with total # of objects = %lld",
      static_cast<long long>(min_spilling_size_),
      max_spilling_file_size_str,
      static_cast<long long>(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 (refactor):
Since you're improving the code, can you move this to the ctor of LocalObjectManager? It's good to log static configuration once on startup instead of every time we try to spill. 

Code after:
  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;
  }

  if (idx == objects_pending_spill_.size() && bytes_to_spill < min_spilling_size_ &&
      !objects_pending_spill_.empty()) {
    // 1. We've gone through all objects and it didn't hit max_fused_object_count_.
    // 2. The total size of the current objects is less than min_spilling_size.
    // 3. There are other objects already being spilled.
    // Let those spill requests finish before we try to spill the current