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

Code before:
      instrumented_io_context &io_service,
      size_t free_objects_batch_size,
      int64_t free_objects_period_ms,
      IOWorkerPoolInterface &io_worker_pool,
      rpc::CoreWorkerClientPool &owner_client_pool,
      int max_io_workers,
      bool is_external_storage_type_fs,
      int64_t max_fused_object_count,
      std::function<void(const std::vector<ObjectID> &)> on_objects_freed,
      std::function<bool(const ray::ObjectID &)> is_plasma_object_spillable,
      pubsub::SubscriberInterface *core_worker_subscriber,
      IObjectDirectory *object_directory,
      ray::observability::MetricInterface &object_store_memory_gauge,
      ray::raylet::SpillManagerMetrics &spill_manager_metrics)
      : self_node_id_(node_id),
        self_node_address_(std::move(self_node_address)),
        self_node_port_(self_node_port),
        io_service_(io_service),
        free_objects_period_ms_(free_objects_period_ms),
        free_objects_batch_size_(free_objects_batch_size),
        io_worker_pool_(io_worker_pool),
        owner_client_pool_(owner_client_pool),
        on_objects_freed_(std::move(on_objects_freed)),
        last_free_objects_at_ms_(current_time_ms()),
        min_spilling_size_(RayConfig::instance().min_spilling_size()),
        max_spilling_file_size_(RayConfig::instance().max_spilling_file_size()),
        num_active_workers_(0),
        max_active_workers_(max_io_workers),
        is_plasma_object_spillable_(std::move(is_plasma_object_spillable)),
        is_external_storage_type_fs_(is_external_storage_type_fs),
        max_fused_object_count_(max_fused_object_count),
        next_spill_error_log_bytes_(RayConfig::instance().verbose_spill_logs()),
        core_worker_subscriber_(core_worker_subscriber),
        object_directory_(object_directory),
        object_store_memory_gauge_(object_store_memory_gauge),
        spill_manager_metrics_(spill_manager_metrics) {}

  /// Pin objects.
  ///
  /// Also wait for the objects' owner to free the object.  The objects will be
  /// released when the owner at the given address fails or replies that the
  /// object can be evicted.
  ///
  /// \param object_ids The objects to be pinned.
  /// \param objects Pointers to the objects to be pinned. The pointer should
  /// be kept in scope until the object can be released.
  /// \param owner_address The owner of the objects to be pinned.
  /// \param generator_id When it's set, this means that it was a dynamically
  /// created ObjectID, so we need to notify the owner of the outer ObjectID
  /// that should already be owned by the same worker. If the outer ObjectID is
  /// still in scope, then the owner can add the dynamically created ObjectID

Review comment (suggestion):
It will be safer if we check that the max_spilling_file_size is >= to the min_spilling_size at initialization time, so we don't get any weird errors due to misconfiguration.

Code after:
      std::string self_node_address,
      int self_node_port,
      instrumented_io_context &io_service,
      size_t free_objects_batch_size,
      int64_t free_objects_period_ms,
      IOWorkerPoolInterface &io_worker_pool,
      rpc::CoreWorkerClientPool &owner_client_pool,
      int max_io_workers,
      bool is_external_storage_type_fs,
      int64_t max_fused_object_count,
      std::function<void(const std::vector<ObjectID> &)> on_objects_freed,
      std::function<bool(const ray::ObjectID &)> is_plasma_object_spillable,
      pubsub::SubscriberInterface *core_worker_subscriber,
      IObjectDirectory *object_directory,
      ray::observability::MetricInterface &object_store_memory_gauge,
      ray::raylet::SpillManagerMetrics &spill_manager_metrics)
      : self_node_id_(node_id),
        self_node_address_(std::move(self_node_address)),
        self_node_port_(self_node_port),
        io_service_(io_service),
        free_objects_period_ms_(free_objects_period_ms),
        free_objects_batch_size_(free_objects_batch_size),
        io_worker_pool_(io_worker_pool),
        owner_client_pool_(owner_client_pool),
        on_objects_freed_(std::move(on_objects_freed)),
        last_free_objects_at_ms_(current_time_ms()),
        min_spilling_size_(RayConfig::instance().min_spilling_size()),
        max_spilling_file_size_bytes_(
            RayConfig::instance().max_spilling_file_size_bytes()),
        num_active_workers_(0),
        max_active_workers_(max_io_workers),
        is_plasma_object_spillable_(std::move(is_plasma_object_spillable)),
        is_external_storage_type_fs_(is_external_storage_type_fs),
        max_fused_object_count_(max_fused_object_count),
        next_spill_error_log_bytes_(RayConfig::instance().verbose_spill_logs()),
        core_worker_subscriber_(core_worker_subscriber),
        object_directory_(object_directory),
        object_store_memory_gauge_(object_store_memory_gauge),
        spill_manager_metrics_(spill_manager_metrics) {
    if (max_spilling_file_size_bytes_ > 0) {
      RAY_CHECK_GE(max_spilling_file_size_bytes_, min_spilling_size_) << absl::StrFormat(
          "Misconfiguration: max_spilling_file_size_bytes (%lld) must be >= "
          "min_spilling_size (%lld). Set max_spilling_file_size_bytes <= 0 to disable.",
          static_cast<long long>(max_spilling_file_size_bytes_),
          static_cast<long long>(min_spilling_size_));
    }

    const std::string max_spilling_file_size_bytes_str =
        max_spilling_file_size_bytes_ <= 0
            ? "unlimited (-1)"
            : absl::StrFormat("%lld",