add IPv6 support for sockets

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

Author: Yicheng-Lu-llll

cpp/src/ray/runtime/native_ray_runtime.cc

@@ -20,6 +20,7 @@
 #include "./object/object_store.h"
 #include "./task/native_task_submitter.h"
 #include "ray/common/ray_config.h"
+#include "ray/util/network_util.h"
 
 namespace ray {
 namespace internal {
@@ -31,7 +32,7 @@ NativeRayRuntime::NativeRayRuntime() {
 
   auto bootstrap_address = ConfigInternal::Instance().bootstrap_ip;
   if (bootstrap_address.empty()) {
-    bootstrap_address = GetNodeIpAddress();
+    bootstrap_address = ray::GetNodeIpAddressFromPerspective();
   }
   global_state_accessor_ = ProcessHelper::GetInstance().CreateGlobalStateAccessor(
       bootstrap_address, ConfigInternal::Instance().bootstrap_port);

cpp/src/ray/test/cluster/cluster_mode_test.cc

@@ -71,7 +71,7 @@ TEST(RayClusterModeTest, FullTest) {
     auto port = absl::GetFlag<int32_t>(FLAGS_redis_port);
     std::string username = absl::GetFlag<std::string>(FLAGS_redis_username);
     std::string password = absl::GetFlag<std::string>(FLAGS_redis_password);
-    std::string local_ip = ray::internal::GetNodeIpAddress();
+    std::string local_ip = ray::GetNodeIpAddressFromPerspective();
     ray::internal::ProcessHelper::GetInstance().StartRayNode(
         local_ip, port, username, password);
     config.address = ray::BuildAddress(local_ip, port);

cpp/src/ray/util/process_helper.cc

@@ -83,7 +83,7 @@ void ProcessHelper::RayStart(CoreWorkerOptions::TaskExecutionCallback callback)
 
   if (ConfigInternal::Instance().worker_type == WorkerType::DRIVER &&
       bootstrap_ip.empty()) {
-    bootstrap_ip = GetNodeIpAddress();
+    bootstrap_ip = ray::GetNodeIpAddressFromPerspective();
     StartRayNode(bootstrap_ip,
                  bootstrap_port,
                  ConfigInternal::Instance().redis_username,
@@ -95,9 +95,9 @@ void ProcessHelper::RayStart(CoreWorkerOptions::TaskExecutionCallback callback)
   std::string node_ip = ConfigInternal::Instance().node_ip_address;
   if (node_ip.empty()) {
     if (!bootstrap_ip.empty()) {
-      node_ip = GetNodeIpAddress(bootstrap_address);
+      node_ip = ray::GetNodeIpAddressFromPerspective(bootstrap_address);
     } else {
-      node_ip = GetNodeIpAddress();
+      node_ip = ray::GetNodeIpAddressFromPerspective();
     }
   }
 

cpp/src/ray/util/util.cc

@@ -24,27 +24,6 @@
 namespace ray {
 namespace internal {
 
-std::string GetNodeIpAddress(const std::string &address) {
-  auto parts = ParseAddress(address);
-  RAY_CHECK(parts.has_value());
-  try {
-    boost::asio::io_service netService;
-    boost::asio::ip::udp::resolver resolver(netService);
-    boost::asio::ip::udp::resolver::query query(
-        boost::asio::ip::udp::v4(), (*parts)[0], (*parts)[1]);
-    boost::asio::ip::udp::resolver::iterator endpoints = resolver.resolve(query);
-    boost::asio::ip::udp::endpoint ep = *endpoints;
-    boost::asio::ip::udp::socket socket(netService);
-    socket.connect(ep);
-    boost::asio::ip::address addr = socket.local_endpoint().address();
-    return addr.to_string();
-  } catch (std::exception &e) {
-    RAY_LOG(FATAL) << "Could not get the node IP address with socket. Exception: "
-                   << e.what();
-    return "";
-  }
-}
-
 std::string getLibraryPathEnv() {
   auto path_env_p = std::getenv(kLibraryPathEnvName);
   if (path_env_p != nullptr && strlen(path_env_p) != 0) {

cpp/src/ray/util/util.h

@@ -18,20 +18,6 @@
 namespace ray {
 namespace internal {
 
-/// IP address by which the local node can be reached *from* the `address`.
-///
-/// The behavior should be the same as `node_ip_address_from_perspective` from Ray Python
-/// code. See
-/// https://stackoverflow.com/questions/2674314/get-local-ip-address-using-boost-asio.
-///
-/// TODO(kfstorm): Make this function shared code and migrate Python & Java to use this
-/// function.
-///
-/// \param address The IP address and port of any known live service on the network
-/// you care about.
-/// \return The IP address by which the local node can be reached from the address.
-std::string GetNodeIpAddress(const std::string &address = "8.8.8.8:53");
-
 std::string getLibraryPathEnv();
 
 }  // namespace internal

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

@@ -11,7 +11,7 @@
 from fairseq_cli.train import main
 
 import ray
-from ray._common.network_utils import build_address
+from ray._common.network_utils import build_address, create_socket
 
 _original_save_checkpoint = fairseq.checkpoint_utils.save_checkpoint
 
@@ -79,7 +79,7 @@ def get_node_ip(self):
 
     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:
+        with closing(create_socket(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,7 +1,13 @@
 from typing import Optional, Tuple, Union
+import socket
 
 from ray._raylet import build_address as _build_address
 from ray._raylet import parse_address as _parse_address
+from ray._raylet import (
+    node_ip_address_from_perspective as _node_ip_address_from_perspective,
+)
+from ray._raylet import node_ip_from_hostname as _node_ip_from_hostname
+from ray._raylet import is_ipv6_ip as _is_ipv6_ip
 
 
 def parse_address(address: str) -> Optional[Tuple[str, str]]:
@@ -29,6 +35,47 @@ def build_address(host: str, port: Union[int, str]) -> str:
     return _build_address(host, port)
 
 
+def node_ip_address_from_perspective(address: str = "") -> str:
+    """IP address by which the local node can be reached *from* the `address`.
+
+    If no address is given, defaults to public DNS servers for detection. For
+    performance, the result is cached when using the default address (empty string).
+    When a specific address is provided, detection is performed fresh every time.
+
+    Args:
+        address: The IP address and port of any known live service on the
+            network you care about.
+
+    Returns:
+        The IP address by which the local node can be reached from the address.
+    """
+    return _node_ip_address_from_perspective(address)
+
+
+def node_ip_from_hostname() -> str:
+    """Get node IP address from hostname resolution without creating external connections.
+
+    This method uses hostname resolution to determine the local node's IP address,
+    avoiding socket creation that could trigger firewall popups on macOS/Windows.
+
+    Returns:
+        The IP address resolved from hostname, or empty string if resolution fails.
+    """
+    return _node_ip_from_hostname()
+
+
+def is_ipv6_ip(ip: str) -> bool:
+    """Check if an IP string is IPv6 format.
+
+    Args:
+        ip: The IP address string to check (must be pure IP, no port).
+
+    Returns:
+        True if the IP is IPv6, False if IPv4.
+    """
+    return _is_ipv6_ip(ip)
+
+
 def is_localhost(host: str) -> bool:
     """Check if the given host string represents a localhost address.
 
@@ -39,3 +86,29 @@ 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 create_socket(socket_type: int = socket.SOCK_STREAM) -> socket.socket:
+    """Create a Python socket object with the appropriate family based on the node IP.
+
+    This function automatically gets the node IP address and creates a socket
+    with the correct family (AF_INET for IPv4, AF_INET6 for IPv6).
+
+    Args:
+        socket_type: The socket type (socket.SOCK_STREAM, socket.SOCK_DGRAM, etc.).
+
+    Returns:
+        A Python socket.socket object configured for the node's IP family.
+
+    Example:
+        # Create a TCP socket for the current node
+        sock = create_socket()
+
+        # Create a UDP socket for the current node
+        sock = create_socket(socket.SOCK_DGRAM)
+    """
+    node_ip = node_ip_address_from_perspective()
+    family = socket.AF_INET6 if is_ipv6_ip(node_ip) else socket.AF_INET
+
+    # Create socket directly with Python socket API
+    return socket.socket(family, socket_type)

python/ray/_private/node.py

@@ -22,7 +22,7 @@
 import ray
 import ray._private.ray_constants as ray_constants
 import ray._private.services
-from ray._common.network_utils import build_address, parse_address
+from ray._common.network_utils import build_address, create_socket, parse_address
 from ray._common.ray_constants import LOGGING_ROTATE_BACKUP_COUNT, LOGGING_ROTATE_BYTES
 from ray._common.utils import try_to_create_directory
 from ray._private.resource_and_label_spec import ResourceAndLabelSpec
@@ -880,7 +880,7 @@ def _get_unused_port(self, allocated_ports=None):
         if allocated_ports is None:
             allocated_ports = set()
 
-        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s = create_socket(socket.SOCK_STREAM)
         s.bind(("", 0))
         port = s.getsockname()[1]
 
@@ -893,7 +893,7 @@ def _get_unused_port(self, allocated_ports=None):
                 # This port is allocated for other usage already,
                 # so we shouldn't use it even if it's not in use right now.
                 continue
-            new_s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            new_s = create_socket(socket.SOCK_STREAM)
             try:
                 new_s.bind(("", new_port))
             except OSError:

python/ray/_private/services.py

@@ -21,7 +21,14 @@
 # Ray modules
 import ray
 import ray._private.ray_constants as ray_constants
-from ray._common.network_utils import build_address, parse_address
+from ray._common.network_utils import (
+    build_address,
+    create_socket,
+    is_ipv6_ip,
+    node_ip_address_from_perspective,
+    node_ip_from_hostname,
+    parse_address,
+)
 from ray._private.ray_constants import RAY_NODE_IP_FILENAME
 from ray._private.resource_isolation_config import ResourceIsolationConfig
 from ray._raylet import GcsClient, GcsClientOptions
@@ -615,52 +622,22 @@ def resolve_ip_for_localhost(host: str):
         return host
 
 
-def node_ip_address_from_perspective(address: str):
-    """IP address by which the local node can be reached *from* the `address`.
-
-    Args:
-        address: The IP address and port of any known live service on the
-            network you care about.
-
-    Returns:
-        The IP address by which the local node can be reached from the address.
-    """
-    ip_address, port = parse_address(address)
-    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    try:
-        # This command will raise an exception if there is no internet
-        # connection.
-        s.connect((ip_address, int(port)))
-        node_ip_address = s.getsockname()[0]
-    except OSError as e:
-        node_ip_address = "127.0.0.1"
-        # [Errno 101] Network is unreachable
-        if e.errno == errno.ENETUNREACH:
-            try:
-                # try get node ip address from host name
-                host_name = socket.getfqdn(socket.gethostname())
-                node_ip_address = socket.gethostbyname(host_name)
-            except Exception:
-                pass
-    finally:
-        s.close()
-
-    return node_ip_address
-
-
 # NOTE: This API should not be used when you obtain the
 # IP address when ray.init is not called because
 # it cannot find the IP address if it is specified by
 # ray start --node-ip-address. You should instead use
 # get_cached_node_ip_address.
-def get_node_ip_address(address="8.8.8.8:53"):
+def get_node_ip_address(address=""):
     if ray._private.worker._global_node is not None:
         return ray._private.worker._global_node.node_ip_address
+
     if not ray_constants.ENABLE_RAY_CLUSTER:
         # Use loopback IP as the local IP address to prevent bothersome
         # firewall popups on OSX and Windows.
         # https://github.com/ray-project/ray/issues/18730.
-        return "127.0.0.1"
+        hostname_ip = node_ip_from_hostname()
+        return "::1" if hostname_ip and is_ipv6_ip(hostname_ip) else "127.0.0.1"
+
     return node_ip_address_from_perspective(address)
 
 
@@ -1222,7 +1199,7 @@ def start_api_server(
             port = ray_constants.DEFAULT_DASHBOARD_PORT
         else:
             port_retries = 0
-            port_test_socket = socket.socket()
+            port_test_socket = create_socket()
             port_test_socket.setsockopt(
                 socket.SOL_SOCKET,
                 socket.SO_REUSEADDR,

python/ray/_private/test_utils.py

@@ -30,7 +30,7 @@
 import ray._private.services
 import ray._private.services as services
 import ray._private.utils
-from ray._common.network_utils import build_address, parse_address
+from ray._common.network_utils import build_address, create_socket, parse_address
 from ray._common.test_utils import wait_for_condition
 from ray._common.utils import get_or_create_event_loop
 from ray._private import (
@@ -776,7 +776,7 @@ def wait_until_server_available(address, timeout_ms=5000, retry_interval_ms=100)
     time_elapsed = 0
     start = time.time()
     while time_elapsed <= timeout_ms:
-        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s = create_socket(socket.SOCK_STREAM)
         s.settimeout(1)
         try:
             s.connect((ip, port))
@@ -1753,7 +1753,7 @@ def job_hook(**kwargs):
 
 
 def find_free_port() -> int:
-    sock = socket.socket()
+    sock = create_socket()
     sock.bind(("", 0))
     port = sock.getsockname()[1]
     sock.close()
@@ -1884,7 +1884,7 @@ def get_current_unused_port():
         A port number that is not currently in use. (Note that this port
         might become used by the time you try to bind to it.)
     """
-    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock = create_socket(socket.SOCK_STREAM)
 
     # Bind the socket to a local address with a random port number
     sock.bind(("localhost", 0))