Adapt worker and nanny to ipv6

Author: csfldf

distributed/nanny.py

@@ -27,7 +27,7 @@
 from distributed import preloading
 from distributed._async_taskgroup import AsyncTaskGroupClosedError
 from distributed.comm import get_address_host
-from distributed.comm.addressing import address_from_user_args
+from distributed.comm.addressing import address_from_user_args, unparse_host_port
 from distributed.compatibility import asyncio_run
 from distributed.config import get_loop_factory
 from distributed.core import (
@@ -48,7 +48,7 @@
 from distributed.protocol.serialize import _is_dumpable
 from distributed.security import Security
 from distributed.utils import (
-    get_ip,
+    get_ip_adaptive,
     get_mp_context,
     json_load_robust,
     log_errors,
@@ -276,7 +276,9 @@ def __init__(  # type: ignore[no-untyped-def]
             and not interface
             and not self.scheduler_addr.startswith("inproc://")
         ):
-            host = get_ip(get_address_host(self.scheduler.address))
+            host = unparse_host_port(
+                get_ip_adaptive(get_address_host(self.scheduler.address))
+            )
 
         self._start_port = port
         self._start_host = host

distributed/node.py

@@ -176,7 +176,13 @@ def start_http_server(
         bound_addresses = get_tcp_server_addresses(self.http_server)
 
         # If more than one address is configured we just use the first here
-        self.http_server.address, self.http_server.port = bound_addresses[0]
+        # Socket addresses representation: https://docs.python.org/3/library/socket.html#socket-families
+        if len(bound_addresses[0]) == 2:  # (host, port) for ipv4
+            self.http_server.address, self.http_server.port = bound_addresses[0]
+        elif len(bound_addresses[0]) == 4:  # (host, port, flowinfo, scope_id) for ipv6
+            self.http_server.address, self.http_server.port = bound_addresses[0][:2]
+        else:
+            raise RuntimeError("Unexpected bound address format")
         self.services["dashboard"] = self.http_server
 
         # Warn on port changes

distributed/utils.py

@@ -226,6 +226,16 @@ def get_ipv6(host="2001:4860:4860::8888", port=80):
     return _get_ip(host, port, family=socket.AF_INET6)
 
 
+def get_ip_adaptive(host="8.8.8.8", port=80):
+    """
+    Get ipv4 or ipv6 adaptively depending on the host to reach.
+    """
+    if ":" in host:
+        return get_ipv6(host, port)
+    else:
+        return get_ip(host, port)
+
+
 def get_ip_interface(ifname):
     """
     Get the local IPv4 address of a network interface.

distributed/worker.py

@@ -96,7 +96,7 @@
 from distributed.threadpoolexecutor import secede as tpe_secede
 from distributed.utils import (
     TimeoutError,
-    get_ip,
+    get_ip_adaptive,
     has_arg,
     in_async_call,
     iscoroutinefunction,
@@ -1405,7 +1405,7 @@ async def start_unsafe(self):
             kwargs = self.security.get_listen_args("worker")
             if self._protocol in ("tcp", "tls"):
                 kwargs = kwargs.copy()
-                kwargs["default_host"] = get_ip(
+                kwargs["default_host"] = get_ip_adaptive(
                     get_address_host(self.scheduler.address)
                 )
             try: