[4/n] IPv6 support: Add IPv6 support for sockets (#56147)

Signed-off-by: Yicheng-Lu-llll <luyc58576@gmail.com>
Signed-off-by: Yicheng-Lu-llll <51814063+Yicheng-Lu-llll@users.noreply.github.com>
Signed-off-by: Jiajun Yao <jeromeyjj@gmail.com>
Co-authored-by: Jiajun Yao <jeromeyjj@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,6 @@
 from fairseq_cli.train import main
 
 import ray
-from ray._common.network_utils import build_address
 
 _original_save_checkpoint = fairseq.checkpoint_utils.save_checkpoint
 
@@ -113,7 +112,7 @@ def run_fault_tolerant_loop():
         # fairseq distributed training.
         ip = ray.get(workers[0].get_node_ip.remote())
         port = ray.get(workers[0].find_free_port.remote())
-        address = f"tcp://{build_address(ip, port)}"
+        address = f"tcp://{ip}:{port}"
 
         # Start the remote processes, and check whether their are any process
         # fails. If so, restart all the processes.

python/ray/_common/network_utils.py

@@ -1,6 +1,13 @@
+import socket
+from functools import lru_cache
 from typing import Optional, Tuple, Union
 
-from ray._raylet import build_address as _build_address, parse_address as _parse_address
+from ray._raylet import (
+    build_address as _build_address,
+    is_ipv6 as _is_ipv6,
+    node_ip_address_from_perspective as _node_ip_address_from_perspective,
+    parse_address as _parse_address,
+)
 
 
 def parse_address(address: str) -> Optional[Tuple[str, str]]:
@@ -28,6 +35,54 @@ def build_address(host: str, port: Union[int, str]) -> str:
     return _build_address(host, port)
 
 
+def node_ip_address_from_perspective(address: Optional[str] = None) -> 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.
+
+    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 is_ipv6(host: str) -> bool:
+    """Check if a host is resolved to IPv6.
+
+    Args:
+        host: The IP or domain name to check (must be without port).
+
+    Returns:
+        True if the host is resolved to IPv6, False if IPv4.
+    """
+    return _is_ipv6(host)
+
+
+@lru_cache(maxsize=1)
+def get_localhost_ip() -> str:
+    """Get localhost loopback ip with IPv4/IPv6 support.
+
+    Returns:
+        The localhost loopback IP.
+    """
+    # Try IPv4 first, then IPv6 localhost resolution
+    for family in [socket.AF_INET, socket.AF_INET6]:
+        try:
+            dns_result = socket.getaddrinfo(
+                "localhost", None, family, socket.SOCK_STREAM
+            )
+            return dns_result[0][4][0]
+        except Exception:
+            continue
+
+    # Final fallback to IPv4 loopback
+    return "127.0.0.1"
+
+
 def is_localhost(host: str) -> bool:
     """Check if the given host string represents a localhost address.
 

python/ray/_private/node.py

@@ -22,7 +22,12 @@
 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,
+    get_localhost_ip,
+    is_ipv6,
+    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
@@ -140,7 +145,7 @@ def __init__(
         )
 
         self._resource_and_label_spec = None
-        self._localhost = socket.gethostbyname("localhost")
+        self._localhost = get_localhost_ip()
         self._ray_params = ray_params
         self._config = ray_params._system_config or {}
 
@@ -882,7 +887,10 @@ 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 = socket.socket(
+            socket.AF_INET6 if is_ipv6(self._node_ip_address) else socket.AF_INET,
+            socket.SOCK_STREAM,
+        )
         s.bind(("", 0))
         port = s.getsockname()[1]
 
@@ -895,7 +903,10 @@ 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 = socket.socket(
+                socket.AF_INET6 if is_ipv6(self._node_ip_address) else socket.AF_INET,
+                socket.SOCK_STREAM,
+            )
             try:
                 new_s.bind(("", new_port))
             except OSError:

python/ray/_private/services.py

@@ -21,7 +21,13 @@
 # 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,
+    get_localhost_ip,
+    is_ipv6,
+    node_ip_address_from_perspective,
+    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
@@ -618,52 +624,21 @@ 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=None):
     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"
+        return get_localhost_ip()
+
     return node_ip_address_from_perspective(address)
 
 
@@ -1225,7 +1200,10 @@ def start_api_server(
             port = ray_constants.DEFAULT_DASHBOARD_PORT
         else:
             port_retries = 0
-            port_test_socket = socket.socket()
+            port_test_socket = socket.socket(
+                socket.AF_INET6 if is_ipv6(host) else socket.AF_INET,
+                socket.SOCK_STREAM,
+            )
             port_test_socket.setsockopt(
                 socket.SOL_SOCKET,
                 socket.SO_REUSEADDR,

python/ray/_private/tls_utils.py

@@ -2,6 +2,11 @@
 import os
 import socket
 
+from ray._common.network_utils import (
+    get_localhost_ip,
+    node_ip_address_from_perspective,
+)
+
 
 def generate_self_signed_tls_certs():
     """Create self-signed key/cert pair for testing.
@@ -29,21 +34,13 @@ def generate_self_signed_tls_certs():
     ).decode()
 
     ray_interal = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "ray-internal")])
-    # This is the same logic used by the GCS server to acquire a
-    # private/interal IP address to listen on. If we just use localhost +
-    # 127.0.0.1 then we won't be able to connect to the GCS and will get
-    # an error like "No match found for server name: 192.168.X.Y"
-    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    s.connect(("8.8.8.8", 80))
-    private_ip_address = s.getsockname()[0]
-    s.close()
     altnames = x509.SubjectAlternativeName(
         [
             x509.DNSName(
                 socket.gethostbyname(socket.gethostname())
-            ),  # Probably 127.0.0.1
-            x509.DNSName("127.0.0.1"),
-            x509.DNSName(private_ip_address),  # 192.168.*.*
+            ),  # Probably 127.0.0.1 or ::1
+            x509.DNSName(get_localhost_ip()),
+            x509.DNSName(node_ip_address_from_perspective()),
             x509.DNSName("localhost"),
         ]
     )