[Core] Consolidate find_free_port to network_utils (ray-project#58304)

This PR is a follow-up to
ray-project#56147 (comment)

The `find_free_port()` function was duplicated across multiple locations
in the codebase. This PR consolidates all implementations into a single
canonical location.

---------

Signed-off-by: Yicheng-Lu-llll <luyc58576@gmail.com>
Signed-off-by: peterxcli <peterxcli@gmail.com>

Author: Yicheng-Lu-llll

doc/source/ray-core/examples/lm/ray_train.py

@@ -80,7 +80,6 @@ def find_free_port(self):
         """Finds a free port on the current node."""
         with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
             s.bind(("", 0))
-            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             return s.getsockname()[1]
 
 

python/ray/_common/network_utils.py

@@ -1,4 +1,5 @@
 import socket
+from contextlib import closing
 from functools import lru_cache
 from typing import Optional, Tuple, Union
 
@@ -93,3 +94,18 @@ def is_localhost(host: str) -> bool:
         True if the host is a localhost address, False otherwise.
     """
     return host in ("localhost", "127.0.0.1", "::1")
+
+
+def find_free_port(family: socket.AddressFamily = socket.AF_INET) -> int:
+    """Find a free port on the local machine.
+
+    Args:
+        family: The socket address family (AF_INET for IPv4, AF_INET6 for IPv6).
+            Defaults to AF_INET.
+
+    Returns:
+        An available port number.
+    """
+    with closing(socket.socket(family, socket.SOCK_STREAM)) as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]

python/ray/_private/test_utils.py

@@ -1850,14 +1850,6 @@ def job_hook(**kwargs):
     sys.exit(0)
 
 
-def find_free_port() -> int:
-    sock = socket.socket()
-    sock.bind(("", 0))
-    port = sock.getsockname()[1]
-    sock.close()
-    return port
-
-
 def wandb_setup_api_key_hook():
     """
     Example external hook to set up W&B API key in

python/ray/air/_internal/util.py

@@ -2,9 +2,7 @@
 import logging
 import os
 import queue
-import socket
 import threading
-from contextlib import closing
 from typing import Optional
 
 import numpy as np
@@ -14,13 +12,6 @@
 logger = logging.getLogger(__name__)
 
 
-def find_free_port(family=socket.AF_INET):
-    with closing(socket.socket(family, socket.SOCK_STREAM)) as s:
-        s.bind(("", 0))
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        return s.getsockname()[1]
-
-
 def is_nan(value):
     return np.isnan(value)
 

python/ray/dashboard/modules/aggregator/tests/test_aggregator_agent.py

@@ -8,11 +8,9 @@
 from google.protobuf.timestamp_pb2 import Timestamp
 
 import ray.dashboard.consts as dashboard_consts
+from ray._common.network_utils import find_free_port
 from ray._private import ray_constants
-from ray._private.test_utils import (
-    find_free_port,
-    wait_for_condition,
-)
+from ray._private.test_utils import wait_for_condition
 from ray._private.utils import init_grpc_channel
 from ray._raylet import GcsClient
 from ray.core.generated.common_pb2 import (

python/ray/dashboard/modules/reporter/tests/test_healthz.py

@@ -4,8 +4,8 @@
 import requests
 
 import ray._private.ray_constants as ray_constants
+from ray._common.network_utils import find_free_port
 from ray._common.test_utils import wait_for_condition
-from ray._private.test_utils import find_free_port
 from ray.tests.conftest import *  # noqa: F401 F403
 
 

python/ray/data/tests/mock_server.py

@@ -21,7 +21,6 @@ def _is_port_available(host, port):
     """Check if a port is available for use."""
     try:
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             s.bind((host, port))
             return True
     except OSError:
@@ -44,7 +43,6 @@ def _find_available_port(host, preferred_port, max_attempts=10):
     # If all else fails, let the OS pick a port
     try:
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             s.bind((host, 0))  # Let OS pick port
             _, port = s.getsockname()
             return port

python/ray/experimental/collective/util.py

@@ -1,9 +1,8 @@
 import socket
-from contextlib import closing
 from typing import TYPE_CHECKING, Tuple
 
 import ray
-from ray._common.network_utils import is_ipv6
+from ray._common.network_utils import find_free_port, is_ipv6
 from ray.experimental.collective.collective_tensor_transport import (
     CollectiveTensorTransport,
 )
@@ -65,13 +64,5 @@ def device_match_transport(device: "torch.device", tensor_transport: Backend) ->
 def get_address_and_port() -> Tuple[str, int]:
     """Returns the IP address and a free port on this node."""
     addr = ray.util.get_node_ip_address()
-    with closing(
-        socket.socket(
-            socket.AF_INET6 if is_ipv6(addr) else socket.AF_INET, socket.SOCK_STREAM
-        )
-    ) as s:
-        s.bind(("", 0))
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        port = s.getsockname()[1]
-
+    port = find_free_port(socket.AF_INET6 if is_ipv6(addr) else socket.AF_INET)
     return addr, port

python/ray/tests/conftest.py

@@ -22,14 +22,13 @@
 
 import ray
 import ray._private.ray_constants as ray_constants
-from ray._common.network_utils import build_address
+from ray._common.network_utils import build_address, find_free_port
 from ray._common.test_utils import wait_for_condition
 from ray._private.conftest_utils import set_override_dashboard_url  # noqa: F401
 from ray._private.runtime_env import virtualenv_utils
 from ray._private.test_utils import (
     RayletKiller,
     external_redis_test_enabled,
-    find_free_port,
     get_and_run_resource_killer,
     get_redis_cli,
     init_error_pubsub,

python/ray/tests/test_gcs_utils.py

@@ -11,11 +11,10 @@
 import ray
 import ray._private.gcs_utils as gcs_utils
 import ray._private.ray_constants as ray_constants
-from ray._common.network_utils import parse_address
+from ray._common.network_utils import find_free_port, parse_address
 from ray._common.test_utils import async_wait_for_condition
 from ray._private.test_utils import (
     external_redis_test_enabled,
-    find_free_port,
     generate_system_config_map,
 )
 from ray._raylet import GcsClient, NodeID