[core] Simplify local/global gc logic (#58671)

Signed-off-by: dayshah <[email protected]>

Author: dayshah

python/ray/_private/gc_collect_manager.py

@@ -11,19 +11,14 @@ class PythonGCThread(threading.Thread):
     """A background thread that triggers Python garbage collection.
 
     This thread waits for GC events from CoreWorker and triggers `gc.collect()` when
-    requested, ensuring that collections are spaced out by at least
-    `min_interval_s` seconds."""
+    when requested."""
 
-    def __init__(
-        self, *, min_interval_s: int = 5, gc_collect_func: Optional[Callable] = None
-    ):
+    def __init__(self, *, gc_collect_func: Optional[Callable] = None):
         logger.debug("Starting Python GC thread")
         super().__init__(name="PythonGCThread", daemon=True)
         self._should_exit = False
-        self._last_gc_time = float("-inf")
-        self._min_gc_interval = min_interval_s
         self._gc_event = threading.Event()
-        # Set the gc_collect_func for UT, defaulting to gc.collect if None
+        # Sets the gc_collect_func (only for testing), defaults to gc.collect
         self._gc_collect_func = gc_collect_func or gc.collect
 
     def trigger_gc(self) -> None:
@@ -37,26 +32,17 @@ def run(self):
             if self._should_exit:
                 break
 
-            time_since_last_gc = time.monotonic() - self._last_gc_time
-            if time_since_last_gc < self._min_gc_interval:
-                logger.debug(
-                    f"Skipping GC, only {time_since_last_gc:.2f}s since last GC"
-                )
-                continue
-
             try:
                 start = time.monotonic()
                 num_freed = self._gc_collect_func()
-                self._last_gc_time = time.monotonic()
                 if num_freed > 0:
                     logger.debug(
                         "gc.collect() freed {} refs in {} seconds".format(
-                            num_freed, self._last_gc_time - start
+                            num_freed, time.monotonic() - start
                         )
                     )
             except Exception as e:
                 logger.error(f"Error during GC: {e}")
-                self._last_gc_time = time.monotonic()
 
     def stop(self):
         logger.debug("Stopping Python GC thread")

python/ray/_private/ray_constants.py

@@ -583,5 +583,3 @@ def gcs_actor_scheduling_enabled():
 RDT_FETCH_FAIL_TIMEOUT_SECONDS = (
     env_integer("RAY_rdt_fetch_fail_timeout_milliseconds", 60000) / 1000
 )
-
-RAY_GC_MIN_COLLECT_INTERVAL = env_float("RAY_GC_MIN_COLLECT_INTERVAL_S", 5)

python/ray/_raylet.pyx

@@ -2279,7 +2279,7 @@ cdef CRayStatus check_signals() nogil:
     return CRayStatus.OK()
 
 
-cdef void gc_collect(c_bool triggered_by_global_gc) nogil:
+cdef void gc_collect() nogil:
      with gil:
         if RayConfig.instance().start_python_gc_manager_thread():
             start = time.perf_counter()
@@ -2706,7 +2706,7 @@ cdef class CoreWorker:
 
         self._gc_thread = None
         if RayConfig.instance().start_python_gc_manager_thread():
-            self._gc_thread = PythonGCThread(min_interval_s=ray_constants.RAY_GC_MIN_COLLECT_INTERVAL)
+            self._gc_thread = PythonGCThread()
             self._gc_thread.start()
 
     def shutdown_driver(self):

python/ray/autoscaler/v2/tests/test_e2e.py

@@ -498,7 +498,7 @@ def foo():
                 "raylet_report_resources_period_milliseconds": 10000,
                 "global_gc_min_interval_s": 1,
                 "local_gc_interval_s": 1,
-                "high_plasma_storage_usage": 0.2,
+                "plasma_store_usage_trigger_gc_threshold": 0.2,
                 "raylet_check_gc_period_milliseconds": 10,
             },
         )

python/ray/includes/libcoreworker.pxd

@@ -409,7 +409,7 @@ cdef extern from "ray/core_worker/core_worker.h" nogil:
         (void(const CObjectID &) nogil) free_actor_object_callback
         (function[void()]() nogil) initialize_thread_callback
         (CRayStatus() nogil) check_signals
-        (void(c_bool) nogil) gc_collect
+        (void() nogil) gc_collect
         (c_vector[c_string](
             const c_vector[CObjectReference] &) nogil) spill_objects
         (int64_t(

python/ray/tests/test_global_gc.py

@@ -14,7 +14,6 @@
 from ray._common.test_utils import wait_for_condition
 from ray._private.gc_collect_manager import PythonGCThread
 from ray._private.internal_api import global_gc
-from ray._private.ray_constants import RAY_GC_MIN_COLLECT_INTERVAL
 
 logger = logging.getLogger(__name__)
 
@@ -248,107 +247,69 @@ def make_garbage(self):
         def has_garbage(self):
             return self.garbage() is not None
 
-    try:
-        gc.disable()
-
-        # 1) Test GC behavior for the local driver.
-
-        # 1a) Wait for the first GC to happen to avoid timing issues.
-        local_ref = weakref.ref(ObjectWithCyclicRef())
-        wait_for_condition(lambda: local_ref() is None, retry_interval_ms=10)
+    gc.disable()
 
-        # 1b) Check that GC *is not* called within the min interval.
-        local_ref = weakref.ref(ObjectWithCyclicRef())
-        time.sleep(RAY_GC_MIN_COLLECT_INTERVAL / 2)
-        assert local_ref() is not None
+    # 1) Test GC behavior for the local driver.
 
-        # 1c) Check that GC *is* called after the min interval.
-        wait_for_condition(
-            lambda: local_ref() is None,
-            timeout=RAY_GC_MIN_COLLECT_INTERVAL * 2,
-        )
+    # 1a) Wait for the first GC to happen to avoid timing issues.
+    local_ref = weakref.ref(ObjectWithCyclicRef())
+    wait_for_condition(lambda: local_ref() is None, retry_interval_ms=10)
 
-        # 2) Test GC behavior for a remote actor.
-        a = GarbageHolder.remote()
+    # 1b) Check that GC is called.
+    wait_for_condition(
+        lambda: local_ref() is None,
+        timeout=5,
+    )
 
-        # 2a) Wait for the first GC to happen to avoid timing issues.
-        ray.get(a.make_garbage.remote())
-        wait_for_condition(
-            lambda: not ray.get(a.has_garbage.remote()), retry_interval_ms=10
-        )
+    # 2) Test GC behavior for a remote actor.
+    a = GarbageHolder.remote()
 
-        # 2b) Check that GC *is not* called within the min interval.
-        ray.get(a.make_garbage.remote())
-        time.sleep(RAY_GC_MIN_COLLECT_INTERVAL / 2)
-        assert ray.get(a.has_garbage.remote())
+    # 2a) Wait for the first GC to happen to avoid timing issues.
+    ray.get(a.make_garbage.remote())
+    wait_for_condition(
+        lambda: not ray.get(a.has_garbage.remote()), retry_interval_ms=10
+    )
 
-        # 2c) Check that GC *is* called after the min interval.
-        wait_for_condition(
-            lambda: not ray.get(a.has_garbage.remote()),
-            timeout=RAY_GC_MIN_COLLECT_INTERVAL * 2,
-        )
+    # 2b) Check that GC is called.
+    wait_for_condition(
+        lambda: not ray.get(a.has_garbage.remote()),
+        timeout=5,
+    )
 
-    finally:
-        gc.enable()
+    gc.enable()
 
 
 def test_gc_manager_thread_basic_functionality():
     mock_gc_collect = Mock(return_value=10)
 
-    gc_thread = PythonGCThread(min_interval_s=1, gc_collect_func=mock_gc_collect)
+    gc_thread = PythonGCThread(gc_collect_func=mock_gc_collect)
+    gc_thread.start()
+    assert gc_thread.is_alive()
 
-    try:
-        gc_thread.start()
-        assert gc_thread.is_alive()
-
-        gc_thread.trigger_gc()
+    gc_thread.trigger_gc()
+    wait_for_condition(lambda: mock_gc_collect.call_count == 1, timeout=2)
 
-        wait_for_condition(lambda: mock_gc_collect.call_count == 1, timeout=2)
+    gc_thread.trigger_gc()
+    wait_for_condition(lambda: mock_gc_collect.call_count == 2, timeout=2)
 
-        mock_gc_collect.assert_called_once()
-
-    finally:
-        gc_thread.stop()
-        assert not gc_thread.is_alive()
-
-
-def test_gc_manager_thread_min_interval_throttling():
-    mock_gc_collect = Mock(return_value=5)
-
-    gc_thread = PythonGCThread(min_interval_s=2, gc_collect_func=mock_gc_collect)
-
-    try:
-        gc_thread.start()
-
-        for _ in range(3):
-            gc_thread.trigger_gc()
-            time.sleep(1)
-
-        wait_for_condition(lambda: mock_gc_collect.call_count == 2, timeout=2)
-
-        assert mock_gc_collect.call_count == 2
-
-    finally:
-        gc_thread.stop()
+    gc_thread.stop()
+    assert not gc_thread.is_alive()
 
 
 def test_gc_manager_thread_exception_handling():
     mock_gc_collect = Mock(side_effect=RuntimeError("GC failed"))
 
-    gc_thread = PythonGCThread(min_interval_s=5, gc_collect_func=mock_gc_collect)
+    gc_thread = PythonGCThread(gc_collect_func=mock_gc_collect)
+    gc_thread.start()
 
-    try:
-        gc_thread.start()
-
-        for _ in range(3):
-            gc_thread.trigger_gc()
-            time.sleep(0.1)
+    for _ in range(3):
+        gc_thread.trigger_gc()
+        time.sleep(0.1)
 
-        assert gc_thread.is_alive()
-        mock_gc_collect.assert_called_once()
+    assert gc_thread.is_alive()
+    assert mock_gc_collect.call_count == 3
 
-    finally:
-        gc_thread.stop()
+    gc_thread.stop()
 
 
 if __name__ == "__main__":

src/ray/common/ray_config_def.h

@@ -396,7 +396,7 @@ RAY_CONFIG(std::string, gcs_storage, "memory")
 RAY_CONFIG(uint32_t, object_store_full_delay_ms, 10)
 
 /// The threshold to trigger a global gc
-RAY_CONFIG(double, high_plasma_storage_usage, 0.7)
+RAY_CONFIG(double, plasma_store_usage_trigger_gc_threshold, 0.7)
 
 /// The amount of time between automatic local Python GC triggers.
 RAY_CONFIG(uint64_t, local_gc_interval_s, 90 * 60)
@@ -405,7 +405,7 @@ RAY_CONFIG(uint64_t, local_gc_interval_s, 90 * 60)
 RAY_CONFIG(uint64_t, local_gc_min_interval_s, 10)
 
 /// The min amount of time between triggering global_gc in raylet. This only applies
-/// to global GCs triggered due to high_plasma_storage_usage.
+/// to global GCs triggered due to plasma_store_usage_trigger_gc_threshold.
 RAY_CONFIG(uint64_t, global_gc_min_interval_s, 30)
 
 /// Duration to wait between retries for failed tasks.

src/ray/core_worker/core_worker.cc

@@ -4183,7 +4183,7 @@ void CoreWorker::HandleLocalGC(rpc::LocalGCRequest request,
                                rpc::LocalGCReply *reply,
                                rpc::SendReplyCallback send_reply_callback) {
   if (options_.gc_collect != nullptr) {
-    options_.gc_collect(request.triggered_by_global_gc());
+    options_.gc_collect();
     send_reply_callback(Status::OK(), nullptr, nullptr);
   } else {
     send_reply_callback(

src/ray/core_worker/core_worker_options.h

@@ -74,15 +74,10 @@ struct CoreWorkerOptions {
       const rpc::TensorTransport &tensor_transport)>;
 
   CoreWorkerOptions()
-      : store_socket(""),
-        raylet_socket(""),
-        enable_logging(false),
-        log_dir(""),
+      : enable_logging(false),
         install_failure_signal_handler(false),
         interactive(false),
-        node_ip_address(""),
         node_manager_port(0),
-        driver_name(""),
         task_execution_callback(nullptr),
         free_actor_object_callback(nullptr),
         check_signals(nullptr),
@@ -97,15 +92,11 @@ struct CoreWorkerOptions {
         cancel_async_actor_task(nullptr),
         actor_shutdown_callback(nullptr),
         is_local_mode(false),
-        serialized_job_config(""),
         metrics_agent_port(-1),
         runtime_env_hash(0),
         cluster_id(ClusterID::Nil()),
-        session_name(""),
-        entrypoint(""),
         worker_launch_time_ms(-1),
-        worker_launched_time_ms(-1),
-        debug_source("") {}
+        worker_launched_time_ms(-1) {}
 
   /// Type of this worker (i.e., DRIVER or WORKER).
   WorkerType worker_type;
@@ -150,7 +141,7 @@ struct CoreWorkerOptions {
   /// Application-language callback to trigger garbage collection in the language
   /// runtime. This is required to free distributed references that may otherwise
   /// be held up in garbage objects.
-  std::function<void(bool triggered_by_global_gc)> gc_collect;
+  std::function<void()> gc_collect;
   /// Application-language callback to spill objects to external storage.
   std::function<std::vector<std::string>(const std::vector<rpc::ObjectReference> &)>
       spill_objects;

src/ray/core_worker/lib/java/io_ray_runtime_RayNativeRuntime.cc

@@ -261,16 +261,12 @@ Java_io_ray_runtime_RayNativeRuntime_nativeInitialize(JNIEnv *env,
         return Status::OK();
       };
 
-  auto gc_collect = [](bool triggered_by_global_gc) {
+  auto gc_collect = []() {
     // A Java worker process usually contains more than one worker.
     // A LocalGC request is likely to be received by multiple workers in a short time.
     // Here we ensure that the 1 second interval of `System.gc()` execution is
     // guaranteed no matter how frequent the requests are received and how many workers
     // the process has.
-    if (!triggered_by_global_gc) {
-      RAY_LOG(DEBUG) << "Skipping non-global GC.";
-      return;
-    }
 
     static absl::Mutex mutex;
     static int64_t last_gc_time_ms = 0;