Remove unused "nums_reconnect_retry" argument from GCS clients (#51298)

This had no effect...

---------

Signed-off-by: Edward Oakes <ed.nmi.oakes@gmail.com>

Author: edoakes

python/ray/_private/gcs_aio_client.py

@@ -20,7 +20,6 @@ def __init__(
         address: str = None,
         loop=None,
         executor=None,
-        nums_reconnect_retry: int = 5,
         cluster_id: Optional[str] = None,
     ):
         # This must be consistent with GcsClient.__cinit__ in _raylet.pyx

python/ray/_private/usage/usage_lib.py

@@ -854,9 +854,7 @@ def generate_report_data(
     """
     assert cluster_id
 
-    gcs_client = ray._raylet.GcsClient(
-        address=gcs_address, nums_reconnect_retry=20, cluster_id=cluster_id
-    )
+    gcs_client = ray._raylet.GcsClient(address=gcs_address, cluster_id=cluster_id)
 
     cluster_metadata = get_cluster_metadata(gcs_client)
     cluster_status_to_report = get_cluster_status_to_report(gcs_client)

python/ray/_raylet.pyx

@@ -2720,42 +2720,6 @@ cdef class EmptyProfileEvent:
         pass
 
 
-def _auto_reconnect(f):
-    @wraps(f)
-    def wrapper(self, *args, **kwargs):
-        if "TEST_RAY_COLLECT_KV_FREQUENCY" in os.environ:
-            with ray._private.utils._CALLED_FREQ_LOCK:
-                ray._private.utils._CALLED_FREQ[f.__name__] += 1
-        remaining_retry = self._nums_reconnect_retry
-        while True:
-            try:
-                return f(self, *args, **kwargs)
-            except RpcError as e:
-                if e.rpc_code in [
-                    GRPC_STATUS_CODE_UNAVAILABLE,
-                    GRPC_STATUS_CODE_UNKNOWN,
-                ]:
-                    if remaining_retry <= 0:
-                        logger.error(
-                            "Failed to connect to GCS. Please check"
-                            " `gcs_server.out` for more details."
-                        )
-                        raise
-                    logger.debug(
-                        f"Failed to send request to gcs, reconnecting. Error {e}"
-                    )
-                    try:
-                        self._connect()
-                    except Exception:
-                        logger.error(f"Connecting to gcs failed. Error {e}")
-                    time.sleep(1)
-                    remaining_retry -= 1
-                    continue
-                raise
-
-    return wrapper
-
-
 cdef class GcsClient:
     """
     Client to the GCS server. Only contains synchronous methods. For async methods,
@@ -2766,10 +2730,8 @@ cdef class GcsClient:
 
     cdef InnerGcsClient inner
 
-    def __cinit__(self, address,
-                  nums_reconnect_retry=RayConfig.instance().nums_py_gcs_reconnect_retry(
-                  ),
-                  cluster_id: str = None):
+    def __cinit__(self, address: str,
+                  cluster_id: Optional[str] = None):
         # For timeout (DEADLINE_EXCEEDED): retries once with timeout_ms.
         #
         # For other RpcError (UNAVAILABLE, UNKNOWN): retries indefinitely until it

python/ray/dashboard/agent.py

@@ -77,7 +77,6 @@ def __init__(
         # Used by the agent and sub-modules.
         self.gcs_aio_client = GcsAioClient(
             address=self.gcs_address,
-            nums_reconnect_retry=ray._config.gcs_rpc_server_reconnect_timeout_s(),
             cluster_id=self.cluster_id_hex,
         )
 

python/ray/dashboard/head.py

@@ -257,11 +257,9 @@ async def run(self):
         gcs_address = self.gcs_address
 
         # Dashboard will handle connection failure automatically
-        self.gcs_client = GcsClient(
-            address=gcs_address, nums_reconnect_retry=0, cluster_id=self.cluster_id_hex
-        )
+        self.gcs_client = GcsClient(address=gcs_address, cluster_id=self.cluster_id_hex)
         self.gcs_aio_client = GcsAioClient(
-            address=gcs_address, nums_reconnect_retry=0, cluster_id=self.cluster_id_hex
+            address=gcs_address, cluster_id=self.cluster_id_hex
         )
         internal_kv._initialize_internal_kv(self.gcs_client)
 

python/ray/dashboard/modules/event/tests/test_generate_export_events.py

@@ -163,9 +163,7 @@ async def test_submission_job_export_events(call_ray_start, tmp_path):  # noqa:
     """
 
     address_info = ray.init(address=call_ray_start)
-    gcs_aio_client = GcsAioClient(
-        address=address_info["gcs_address"], nums_reconnect_retry=0
-    )
+    gcs_aio_client = GcsAioClient(address=address_info["gcs_address"])
     job_manager = JobManager(gcs_aio_client, tmp_path)
 
     # Submit a job.

python/ray/dashboard/modules/job/tests/conftest.py

@@ -20,9 +20,7 @@ def create_ray_cluster(_tracing_startup_hook=None):
 
 def create_job_manager(ray_cluster, tmp_path):
     address_info = ray_cluster
-    gcs_aio_client = GcsAioClient(
-        address=address_info["gcs_address"], nums_reconnect_retry=0
-    )
+    gcs_aio_client = GcsAioClient(address=address_info["gcs_address"])
     return JobManager(gcs_aio_client, tmp_path)
 
 

python/ray/dashboard/modules/job/tests/test_job_manager.py

@@ -58,9 +58,7 @@ async def test_get_scheduling_strategy(
 ):
     monkeypatch.setenv(RAY_JOB_ALLOW_DRIVER_ON_WORKER_NODES_ENV_VAR, "0")
     address_info = ray.init(address=call_ray_start)
-    gcs_aio_client = GcsAioClient(
-        address=address_info["gcs_address"], nums_reconnect_retry=0
-    )
+    gcs_aio_client = GcsAioClient(address=address_info["gcs_address"])
 
     job_manager = JobManager(gcs_aio_client, tmp_path)
 
@@ -104,9 +102,7 @@ async def test_submit_no_ray_address(call_ray_start, tmp_path):  # noqa: F811
     """Test that a job script with an unspecified Ray address works."""
 
     address_info = ray.init(address=call_ray_start)
-    gcs_aio_client = GcsAioClient(
-        address=address_info["gcs_address"], nums_reconnect_retry=0
-    )
+    gcs_aio_client = GcsAioClient(address=address_info["gcs_address"])
     job_manager = JobManager(gcs_aio_client, tmp_path)
 
     init_ray_no_address_script = """
@@ -141,9 +137,7 @@ async def test_submit_no_ray_address(call_ray_start, tmp_path):  # noqa: F811
 async def test_get_all_job_info(call_ray_start, tmp_path):  # noqa: F811
     """Test that JobInfo is correctly populated in the GCS get_all_job_info API."""
     address_info = ray.init(address=call_ray_start)
-    gcs_aio_client = GcsAioClient(
-        address=address_info["gcs_address"], nums_reconnect_retry=0
-    )
+    gcs_aio_client = GcsAioClient(address=address_info["gcs_address"])
     job_manager = JobManager(gcs_aio_client, tmp_path)
 
     # Submit a job.
@@ -190,9 +184,7 @@ async def test_get_all_job_info_with_is_running_tasks(call_ray_start):  # noqa:
     """Test the is_running_tasks bit in the GCS get_all_job_info API."""
 
     address_info = ray.init(address=call_ray_start)
-    gcs_aio_client = GcsAioClient(
-        address=address_info["gcs_address"], nums_reconnect_retry=0
-    )
+    gcs_aio_client = GcsAioClient(address=address_info["gcs_address"])
 
     @ray.remote
     def sleep_forever():
@@ -271,9 +263,7 @@ class Actor:
 async def test_job_supervisor_log_json(call_ray_start, tmp_path):  # noqa: F811
     """Test JobSupervisor logs are structured JSON logs"""
     address_info = ray.init(address=call_ray_start)
-    gcs_aio_client = GcsAioClient(
-        address=address_info["gcs_address"], nums_reconnect_retry=0
-    )
+    gcs_aio_client = GcsAioClient(address=address_info["gcs_address"])
     job_manager = JobManager(gcs_aio_client, tmp_path)
     job_id = await job_manager.submit_job(
         entrypoint="echo hello 1", submission_id="job_1"
@@ -304,9 +294,7 @@ async def test_job_supervisor_logs_saved(
 ):
     """Test JobSupervisor logs are saved to jobs/supervisor-{submission_id}.log"""
     address_info = ray.init(address=call_ray_start)
-    gcs_aio_client = GcsAioClient(
-        address=address_info["gcs_address"], nums_reconnect_retry=0
-    )
+    gcs_aio_client = GcsAioClient(address=address_info["gcs_address"])
     job_manager = JobManager(gcs_aio_client, tmp_path)
     job_id = await job_manager.submit_job(
         entrypoint="echo hello 1", submission_id="job_1"
@@ -342,9 +330,7 @@ async def test_runtime_env_setup_logged_to_job_driver_logs(
 ):
     """Test runtime env setup messages are logged to jobs driver log"""
     address_info = ray.init(address=call_ray_start)
-    gcs_aio_client = GcsAioClient(
-        address=address_info["gcs_address"], nums_reconnect_retry=0
-    )
+    gcs_aio_client = GcsAioClient(address=address_info["gcs_address"])
     job_manager = JobManager(gcs_aio_client, tmp_path)
 
     job_id = await job_manager.submit_job(

python/ray/dashboard/utils.py

@@ -169,7 +169,6 @@ def gcs_client(self):
         if self._gcs_client is None:
             self._gcs_client = GcsClient(
                 address=self._config.gcs_address,
-                nums_reconnect_retry=0,
                 cluster_id=self._config.cluster_id_hex,
             )
         return self._gcs_client
@@ -179,7 +178,6 @@ def gcs_aio_client(self):
         if self._gcs_aio_client is None:
             self._gcs_aio_client = GcsAioClient(
                 address=self._config.gcs_address,
-                nums_reconnect_retry=0,
                 cluster_id=self._config.cluster_id_hex,
             )
             if not internal_kv._internal_kv_initialized():
@@ -835,7 +833,7 @@ def ray_address_to_api_server_url(address: Optional[str]) -> str:
     """
 
     address = services.canonicalize_bootstrap_address_or_die(address)
-    gcs_client = GcsClient(address=address, nums_reconnect_retry=0)
+    gcs_client = GcsClient(address=address)
 
     ray.experimental.internal_kv._initialize_internal_kv(gcs_client)
     api_server_url = ray._private.utils.internal_kv_get_with_retry(

python/ray/tests/test_gcs_fault_tolerance.py

@@ -437,7 +437,7 @@ def ready(self):
 
 def test_gcs_client_reconnect(ray_start_regular_with_external_redis):
     gcs_address = ray._private.worker.global_worker.gcs_client.address
-    gcs_client = ray._raylet.GcsClient(address=gcs_address, nums_reconnect_retry=20)
+    gcs_client = ray._raylet.GcsClient(address=gcs_address)
 
     gcs_client.internal_kv_put(b"a", b"b", True, None)
     assert gcs_client.internal_kv_get(b"a", None) == b"b"
@@ -467,9 +467,7 @@ def test_gcs_aio_client_reconnect(ray_start_regular_with_external_redis):
     passed = [False]
 
     async def async_kv_get():
-        gcs_aio_client = gcs_utils.GcsAioClient(
-            address=gcs_address, nums_reconnect_retry=20
-        )
+        gcs_aio_client = gcs_utils.GcsAioClient(address=gcs_address)
         assert await gcs_aio_client.internal_kv_get(b"a", None) == b"b"
         return True