Remove repeat push timeout from object manager (#11874)

python/ray/includes/ray_config.pxd

@@ -49,8 +49,6 @@ cdef extern from "ray/common/ray_config.h" nogil:
 
         int object_manager_push_timeout_ms() const
 
-        int object_manager_repeated_push_delay_ms() const
-
         uint64_t object_manager_default_chunk_size() const
 
         int num_workers_per_process_python() const

python/ray/includes/ray_config.pxi

@@ -84,10 +84,6 @@ cdef class Config:
     def object_manager_push_timeout_ms():
         return RayConfig.instance().object_manager_push_timeout_ms()
 
-    @staticmethod
-    def object_manager_repeated_push_delay_ms():
-        return RayConfig.instance().object_manager_repeated_push_delay_ms()
-
     @staticmethod
     def object_manager_default_chunk_size():
         return RayConfig.instance().object_manager_default_chunk_size()

python/ray/tests/test_multinode_failures_2.py

@@ -22,7 +22,6 @@ import ray.ray_constants as ray_constants
             "num_heartbeats_timeout": 10,
             "object_manager_pull_timeout_ms": 1000,
             "object_manager_push_timeout_ms": 1000,
-            "object_manager_repeated_push_delay_ms": 1000,
         },
     }],
     indirect=True)

python/ray/tests/test_object_manager.py

@@ -199,15 +199,11 @@ def test_actor_broadcast(ray_start_cluster_with_resource):
 def test_object_transfer_retry(ray_start_cluster):
     cluster = ray_start_cluster
 
-    repeated_push_delay = 1
-
     # Force the sending object manager to allow duplicate pushes again sooner.
     # Also, force the receiving object manager to retry the pull sooner. We
     # make the chunk size smaller in order to make it easier to test objects
     # with multiple chunks.
     config = {
-        "object_manager_repeated_push_delay_ms": repeated_push_delay * 1000,
-        "object_manager_pull_timeout_ms": repeated_push_delay * 1000 / 4,
         "object_manager_default_chunk_size": 1000,
         "object_store_full_max_retries": 1,
     }
@@ -230,13 +226,7 @@ def test_object_transfer_retry(ray_start_cluster):
     # Get the objects locally to cause them to be transferred. This is the
     # first time the objects are getting transferred, so it should happen
     # quickly.
-    start_time = time.time()
     ray.get(x_id)
-    end_time = time.time()
-    if end_time - start_time > repeated_push_delay:
-        warnings.warn("The initial transfer took longer than the repeated "
-                      "push delay, so this test may not be testing the thing "
-                      "it's supposed to test.")
 
     def not_exists():
         return not ray.worker.global_worker.core_worker.object_exists(x_id)
@@ -257,20 +247,6 @@ def test_object_transfer_retry(ray_start_cluster):
 
     # Get the object again and make sure it gets transferred.
     ray.get(x_id)
-    end_transfer_time = time.time()
-    # We should have had to wait for the repeated push delay.
-    assert end_transfer_time - start_time >= repeated_push_delay
-
-    # Force the object to be evicted again and wait longer than the repeated
-    # push delay and make sure that the object is transferred again.
-    force_eviction()
-    time.sleep(repeated_push_delay)
-
-    # Fetch the object again. This should not wait for the delay.
-    start_time = time.time()
-    ray.get(x_id)
-    end_time = time.time()
-    assert end_time - start_time < repeated_push_delay
 
 
 # The purpose of this test is to make sure we can transfer many objects. In the