From 723ad8074d55193ed01106b64623e7ca4d11877f Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Wed, 22 Jul 2026 11:53:43 +0200 Subject: [PATCH 01/25] Add UDS support --- distributed/comm/__init__.py | 3 +- distributed/comm/addressing.py | 6 +- distributed/comm/core.py | 7 +- distributed/comm/tcp.py | 8 +- distributed/comm/tests/test_comms.py | 9 +++ distributed/comm/tests/test_uds.py | 81 ++++++++++++++++++++ distributed/comm/uds.py | 109 +++++++++++++++++++++++++++ distributed/comm/utils.py | 10 +++ 8 files changed, 227 insertions(+), 6 deletions(-) create mode 100644 distributed/comm/tests/test_uds.py create mode 100644 distributed/comm/uds.py diff --git a/distributed/comm/__init__.py b/distributed/comm/__init__.py index 0b9801c5cd3..d3f16eb669c 100644 --- a/distributed/comm/__init__.py +++ b/distributed/comm/__init__.py @@ -17,10 +17,11 @@ def _register_transports(): - from distributed.comm import inproc, tcp, ws + from distributed.comm import inproc, tcp, uds, ws backends["tcp"] = tcp.TCPBackend() backends["tls"] = tcp.TLSBackend() + backends["uds"] = uds.UDSBackend() try: # If `distributed-ucxx` is installed, it takes over the protocol="ucx" support diff --git a/distributed/comm/addressing.py b/distributed/comm/addressing.py index f5e1660da13..8bb381d526b 100644 --- a/distributed/comm/addressing.py +++ b/distributed/comm/addressing.py @@ -14,6 +14,8 @@ def parse_address(addr: str, strict: bool = False) -> tuple[str, str]: >>> parse_address('tcp://127.0.0.1') ('tcp', '127.0.0.1') + >>> parse_address('uds:///tmp/socket') + ('uds', '/tmp/socket') If strict is set to true the address must have a scheme. """ @@ -37,7 +39,7 @@ def unparse_address(scheme: str, loc: str) -> str: Undo parse_address(). >>> unparse_address('tcp', '127.0.0.1') - 'tcp://127.0.0.1' + 'tcp://127.0.0.1' # Example remains unchanged. """ return f"{scheme}://{loc}" @@ -159,7 +161,7 @@ def get_local_address_for(addr: str) -> str: >>> get_local_address_for('tcp://8.8.8.8:1234') 'tcp://192.168.1.68' >>> get_local_address_for('tcp://127.0.0.1:1234') - 'tcp://127.0.0.1' + 'tcp://127.0.0.1' # Example remains unchanged. """ scheme, loc = parse_address(addr) backend = registry.get_backend(scheme) diff --git a/distributed/comm/core.py b/distributed/comm/core.py index be454bde03f..1f83a3c139e 100644 --- a/distributed/comm/core.py +++ b/distributed/comm/core.py @@ -13,7 +13,11 @@ from dask.utils import parse_timedelta from distributed.comm import registry -from distributed.comm.addressing import get_address_host, parse_address, resolve_address +from distributed.comm.addressing import ( + get_address_host, + parse_address, + resolve_address, +) from distributed.metrics import time from distributed.protocol.compression import get_compression_settings from distributed.protocol.pickle import HIGHEST_PROTOCOL @@ -321,6 +325,7 @@ async def connect( scheme, loc = parse_address(addr) backend = registry.get_backend(scheme) connector = backend.get_connector() + comm = None start = time() diff --git a/distributed/comm/tcp.py b/distributed/comm/tcp.py index 1b85ede2321..fb5c2690990 100644 --- a/distributed/comm/tcp.py +++ b/distributed/comm/tcp.py @@ -59,7 +59,7 @@ def set_tcp_timeout(comm): """ Set kernel-level TCP timeout on the stream. """ - if comm.closed(): + if comm.closed() or comm.socket.family is socket.AF_UNIX: return timeout = dask.config.get("distributed.comm.timeouts.tcp") @@ -124,7 +124,10 @@ def get_stream_address(comm): if comm.closed(): raise CommClosedError() - return unparse_host_port(*comm.socket.getsockname()[:2]) + if comm.socket.family is socket.AF_UNIX: + return comm.socket.getsockname() or comm.socket.getpeername() + else: + return unparse_host_port(*comm.socket.getsockname()[:2]) def convert_stream_closed_error(obj, exc): @@ -567,6 +570,7 @@ async def connect(self, address, deserialize=True, **connection_args): raise FatalCommClosedError() from err local_address = self.prefix + get_stream_address(stream) + comm = self.comm_class( stream, local_address, self.prefix + address, deserialize ) diff --git a/distributed/comm/tests/test_comms.py b/distributed/comm/tests/test_comms.py index 306e875ecee..b7fe5b0a946 100644 --- a/distributed/comm/tests/test_comms.py +++ b/distributed/comm/tests/test_comms.py @@ -102,6 +102,10 @@ def get_tcp_comm_pair(**kwargs): return get_comm_pair("tcp://", **kwargs) +def get_uds_comm_pair(**kwargs): + return get_comm_pair("uds://", **kwargs) + + def get_tls_comm_pair(**kwargs): kwargs.update(tls_kwargs) return get_comm_pair("tls://", **kwargs) @@ -171,6 +175,7 @@ def test_get_address_host(tcp): assert f("tcp://127.0.0.1:123") == "127.0.0.1" assert f("inproc://%s/%d/123" % (get_ip(), os.getpid())) == get_ip() + assert f("uds://%2Ftmp%2Fdask.sock") == "%2Ftmp%2Fdask.sock" def test_resolve_address(tcp): @@ -192,6 +197,8 @@ def test_resolve_address(tcp): assert f("tcp://localhost:456") == "tcp://127.0.0.1:456" assert f("tls://localhost:456") == "tls://127.0.0.1:456" + assert f("uds://%2Ftmp%2Fdask.sock") == "uds://localhost" + def test_get_local_address_for(tcp): f = get_local_address_for @@ -206,6 +213,8 @@ def test_get_local_address_for(tcp): assert inproc_res.startswith("inproc://") assert inproc_res != inproc_arg + assert f("uds://%2Ftmp%2Fdask.sock") == "uds://%2Ftmp%2Fdask.sock" + # # Test concrete transport APIs diff --git a/distributed/comm/tests/test_uds.py b/distributed/comm/tests/test_uds.py new file mode 100644 index 00000000000..5ef5930259d --- /dev/null +++ b/distributed/comm/tests/test_uds.py @@ -0,0 +1,81 @@ +import asyncio +import os + +import pytest + +from distributed.comm.addressing import parse_address, unparse_address +from distributed.comm.core import connect +from distributed.comm.registry import backends, get_backend +from distributed.comm.uds import UDSBackend, UDSListener +from distributed.utils_test import gen_test + + +@pytest.fixture(params=["tornado"]) +def uds(monkeypatch, request): + """Set the TCP backend to either tornado or asyncio""" + if request.param == "tornado": + import distributed.comm.uds as uds + else: + raise NotImplementedError() + monkeypatch.setitem(backends, "uds", UDSBackend()) + return uds + + +def test_registered(): + assert "uds" in backends + backend = get_backend("uds") + assert isinstance(backend, UDSBackend) + + +def test_parse_uds_address(): + addr = "uds:///tmp/dask-test.sock" + scheme, loc = parse_address(addr) + assert scheme == "uds" + assert loc == "/tmp/dask-test.sock" + assert unparse_address(scheme, loc) == addr + + +@gen_test() +async def test_uds_specific(uds): + """ + Test concrete UDS API. + """ + + async def handle_comm(comm): + assert comm.peer_address == (f"uds://{host}:0") + assert comm.extra_info == {} + msg = await comm.read() + msg["op"] = "pong" + await comm.write(msg) + await comm.close() + + listener = await UDSListener("localhost", handle_comm) + host, port = listener.get_host_port() + + assert host.endswith(".sock") + assert port == 0 # we fake port 0 when using UDS + + l = [] + + async def client_communicate(key, delay=0): + comm = await connect(listener.contact_address) + assert comm.peer_address == f"uds://{host}:0" + assert comm.extra_info == {} + await comm.write({"op": "ping", "data": key}) + if delay: + await asyncio.sleep(delay) + msg = await comm.read() + assert msg == {"op": "pong", "data": key} + l.append(key) + await comm.close() + + await client_communicate(key=1234) + + # Many clients at once + N = 100 + futures = [client_communicate(key=i, delay=0.05) for i in range(N)] + await asyncio.gather(*futures) + assert set(l) == {1234} | set(range(N)) + + listener.stop() + assert not os.path.exists(host) # assert socket deleted diff --git a/distributed/comm/uds.py b/distributed/comm/uds.py new file mode 100644 index 00000000000..9168c9b9920 --- /dev/null +++ b/distributed/comm/uds.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import logging +import os +import socket + +import tornado.netutil as netutil +from tornado.tcpclient import TCPClient +from tornado.tcpserver import TCPServer + +import dask + +from distributed.comm.registry import backends +from distributed.comm.tcp import ( + MAX_BUFFER_SIZE, + TCP, + BaseTCPBackend, + TCPConnector, + TCPListener, +) +from distributed.comm.utils import ( + get_uds_path, +) + +logger = logging.getLogger(__name__) + + +class UnixSocketResolver(netutil.Resolver): + """A resolver for Unix Domain Sockets. Always returns socket type and pathname.""" + + async def resolve(self, host: str, port: int, *args, **kwargs) -> tuple[int, str]: + return [(socket.AF_UNIX, host)] + + +class UDSListener(TCPListener): + prefix = "uds://" + comm_class = TCP + + def __init__( + self, + address, + *args, + **kwargs, + ): + path = get_uds_path(address) + if ":" not in path: + path = f"{path}:0" + super().__init__(path, *args, **kwargs) # fake port 0 + + def get_host_port(self): + """ + The listening address as a tuple. Port is always 0. + """ + self._check_started() + + if self.bound_address is None: + self.bound_address = self.tcp_server._sockets[0].getsockname() + return (self.bound_address, 0) # fake port 0 + + async def _handle_stream(self, stream, address): + return await super()._handle_stream(stream, (self.bound_address, 0)) + + async def start(self): + self.tcp_server = TCPServer(max_buffer_size=MAX_BUFFER_SIZE, **self.server_args) + self.tcp_server.handle_stream = self._handle_stream + # When shuffling data between workers, there can + # really be O(cluster size) connection requests + # on a single worker socket, make sure the backlog + # is large enough not to lose any. + backlog = int(dask.config.get("distributed.comm.socket-backlog")) + socket = netutil.bind_unix_socket( + self.ip, # self.ip is actually the path to the socket + mode=0o600, + backlog=backlog, + ) + self.tcp_server.add_socket(socket) + self.bound_address = self.ip # ip is path to unix socket + + def stop(self): + super().stop() + if os.path.exists(self.bound_address): + try: + os.remove(self.bound_address) + except OSError as e: + print(f"FAILED {e}") + logger.debug( + f"Attempted removal of socket at {self.bound_address} failed with error: {e}" + ) + + +class UDSConnector(TCPConnector): + client: ClassVar[TCPClient] = TCPClient(resolver=UnixSocketResolver()) + + prefix = "uds://" + comm_class = TCP + + +class UDSBackend(BaseTCPBackend): + _connector_class = UDSConnector + _listener_class = UDSListener + + def get_address_host(self, loc): + return loc.split(":")[0] + + def resolve_address(self, loc): + return loc + + +backends["uds"] = UDSBackend() diff --git a/distributed/comm/utils.py b/distributed/comm/utils.py index e91ddc2c839..30799c4064e 100644 --- a/distributed/comm/utils.py +++ b/distributed/comm/utils.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +import os import socket import dask @@ -124,3 +125,12 @@ def ensure_concrete_host(host, default_host=None): return default_host or get_ipv6() else: return host + + +def get_uds_path(address: str) -> bool: + if os.path.isabs(address): + return address + else: + import uuid + + return os.path.join("/tmp", f"{str(uuid.uuid4()).replace('-', '')}.sock") From 791466a481ea22befe0c7586d9165c630b2675e3 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Thu, 23 Jul 2026 16:46:56 +0200 Subject: [PATCH 02/25] Also run dashboard on unix domain socket --- distributed/node.py | 40 ++++++++++++++++++++++++++-------------- distributed/utils.py | 3 +++ 2 files changed, 29 insertions(+), 14 deletions(-) diff --git a/distributed/node.py b/distributed/node.py index 9f9a731f51a..76d2cc047e9 100644 --- a/distributed/node.py +++ b/distributed/node.py @@ -1,12 +1,14 @@ from __future__ import annotations import logging +import os import ssl import warnings import weakref from contextlib import suppress import tlz +import tornado.netutil as netutil from tornado.httpserver import HTTPServer import dask @@ -160,20 +162,30 @@ def start_http_server( change_port = False retries_left = 3 - while True: - try: - if not change_port: - self.http_server.listen(**http_address) - else: - self.http_server.listen(**tlz.merge(http_address, {"port": 0})) - break - except Exception: - change_port = True - retries_left = retries_left - 1 - if retries_left < 1: - raise - - bound_addresses = get_tcp_server_addresses(self.http_server) + + if os.path.isabs(http_address["address"]): + dashboard_socket = netutil.bind_unix_socket( + http_address["address"], mode=0o666 + ) + self.http_server.add_socket(dashboard_socket) + bound_addresses = [http_address["address"]] + else: + while True: + try: + if not change_port: + self.http_server.listen(**http_address) + else: + self.http_server.listen( + **tlz.merge(http_address, {"port": 0}) + ) + break + except Exception: + change_port = True + retries_left = retries_left - 1 + if retries_left < 1: + raise + + bound_addresses = get_tcp_server_addresses(self.http_server) # If more than one address is configured we just use the first here # Socket addresses representation: https://docs.python.org/3/library/socket.html#socket-families diff --git a/distributed/utils.py b/distributed/utils.py index 8fc57cd379f..caf748bc361 100644 --- a/distributed/utils.py +++ b/distributed/utils.py @@ -1524,6 +1524,9 @@ def clean_dashboard_address(addrs: AnyType, default_listen_ip: str = "") -> list [{'address': '', 'port': 8787}, {'address': '', 'port': 8887}] """ + if isinstance(addrs, str) and os.path.isabs(addrs): # unix socket + return [{"address": addrs, "port": 0}] + if default_listen_ip == "0.0.0.0": default_listen_ip = "" # for IPV6 From 455252db3060cb062890f2aa4b656e8cd93f64e0 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 09:39:19 +0200 Subject: [PATCH 03/25] Pass linter --- distributed/comm/uds.py | 5 ++++- distributed/comm/utils.py | 4 ++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/distributed/comm/uds.py b/distributed/comm/uds.py index 9168c9b9920..8587b136b21 100644 --- a/distributed/comm/uds.py +++ b/distributed/comm/uds.py @@ -3,6 +3,7 @@ import logging import os import socket +from typing import ClassVar import tornado.netutil as netutil from tornado.tcpclient import TCPClient @@ -28,7 +29,9 @@ class UnixSocketResolver(netutil.Resolver): """A resolver for Unix Domain Sockets. Always returns socket type and pathname.""" - async def resolve(self, host: str, port: int, *args, **kwargs) -> tuple[int, str]: + async def resolve( + self, host: str, port: int, family: socket.AddressFamily = socket.AF_UNSPEC + ) -> list[tuple[int, str]]: return [(socket.AF_UNIX, host)] diff --git a/distributed/comm/utils.py b/distributed/comm/utils.py index 30799c4064e..83e55896534 100644 --- a/distributed/comm/utils.py +++ b/distributed/comm/utils.py @@ -127,9 +127,9 @@ def ensure_concrete_host(host, default_host=None): return host -def get_uds_path(address: str) -> bool: +def get_uds_path(address: str) -> str: if os.path.isabs(address): - return address + return str(address) else: import uuid From 819c378bd5c494fc3fbef772f8f0683d8606c6ee Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 10:00:33 +0200 Subject: [PATCH 04/25] Move integration test to test_comms.py --- distributed/comm/tests/test_comms.py | 65 ++++++++++++++++++++++++--- distributed/comm/tests/test_uds.py | 66 +--------------------------- 2 files changed, 59 insertions(+), 72 deletions(-) diff --git a/distributed/comm/tests/test_comms.py b/distributed/comm/tests/test_comms.py index b7fe5b0a946..a0118422775 100644 --- a/distributed/comm/tests/test_comms.py +++ b/distributed/comm/tests/test_comms.py @@ -55,6 +55,17 @@ def tcp(monkeypatch, request): return tcp +@pytest.fixture(params=["tornado"]) +def uds(monkeypatch, request): + """Set the TCP backend to either tornado or asyncio""" + if request.param == "tornado": + import distributed.comm.uds as uds + else: + raise NotImplementedError() + monkeypatch.setitem(backends, "uds", uds.UDSBackend()) + return uds + + ca_file = get_cert("tls-ca-cert.pem") # The Subject field of our test certs @@ -102,10 +113,6 @@ def get_tcp_comm_pair(**kwargs): return get_comm_pair("tcp://", **kwargs) -def get_uds_comm_pair(**kwargs): - return get_comm_pair("uds://", **kwargs) - - def get_tls_comm_pair(**kwargs): kwargs.update(tls_kwargs) return get_comm_pair("tls://", **kwargs) @@ -197,7 +204,7 @@ def test_resolve_address(tcp): assert f("tcp://localhost:456") == "tcp://127.0.0.1:456" assert f("tls://localhost:456") == "tls://127.0.0.1:456" - assert f("uds://%2Ftmp%2Fdask.sock") == "uds://localhost" + assert f("uds://%2Ftmp%2Fdask.sock") == "uds://%2Ftmp%2Fdask.sock" def test_get_local_address_for(tcp): @@ -213,8 +220,6 @@ def test_get_local_address_for(tcp): assert inproc_res.startswith("inproc://") assert inproc_res != inproc_arg - assert f("uds://%2Ftmp%2Fdask.sock") == "uds://%2Ftmp%2Fdask.sock" - # # Test concrete transport APIs @@ -286,6 +291,52 @@ async def client_communicate(key, delay=0): assert set(l) == {1234} | set(range(N)) +@gen_test() +async def test_uds_specific(uds): + """ + Test concrete UDS API. + """ + + async def handle_comm(comm): + assert comm.peer_address == (f"uds://{host}:0") + assert comm.extra_info == {} + msg = await comm.read() + msg["op"] = "pong" + await comm.write(msg) + await comm.close() + + listener = await uds.UDSListener("localhost", handle_comm) + host, port = listener.get_host_port() + + assert host.endswith(".sock") + assert port == 0 # we fake port 0 when using UDS + + l = [] + + async def client_communicate(key, delay=0): + comm = await connect(listener.contact_address) + assert comm.peer_address == f"uds://{host}:0" + assert comm.extra_info == {} + await comm.write({"op": "ping", "data": key}) + if delay: + await asyncio.sleep(delay) + msg = await comm.read() + assert msg == {"op": "pong", "data": key} + l.append(key) + await comm.close() + + await client_communicate(key=1234) + + # Many clients at once + N = 100 + futures = [client_communicate(key=i, delay=0.05) for i in range(N)] + await asyncio.gather(*futures) + assert set(l) == {1234} | set(range(N)) + + listener.stop() + assert not os.path.exists(host) # assert socket deleted + + @pytest.mark.parametrize("sni", [None, "localhost"]) @gen_test() async def test_tls_specific(tcp, sni): diff --git a/distributed/comm/tests/test_uds.py b/distributed/comm/tests/test_uds.py index 5ef5930259d..026a93f546e 100644 --- a/distributed/comm/tests/test_uds.py +++ b/distributed/comm/tests/test_uds.py @@ -1,24 +1,6 @@ -import asyncio -import os - -import pytest - from distributed.comm.addressing import parse_address, unparse_address -from distributed.comm.core import connect from distributed.comm.registry import backends, get_backend -from distributed.comm.uds import UDSBackend, UDSListener -from distributed.utils_test import gen_test - - -@pytest.fixture(params=["tornado"]) -def uds(monkeypatch, request): - """Set the TCP backend to either tornado or asyncio""" - if request.param == "tornado": - import distributed.comm.uds as uds - else: - raise NotImplementedError() - monkeypatch.setitem(backends, "uds", UDSBackend()) - return uds +from distributed.comm.uds import UDSBackend def test_registered(): @@ -33,49 +15,3 @@ def test_parse_uds_address(): assert scheme == "uds" assert loc == "/tmp/dask-test.sock" assert unparse_address(scheme, loc) == addr - - -@gen_test() -async def test_uds_specific(uds): - """ - Test concrete UDS API. - """ - - async def handle_comm(comm): - assert comm.peer_address == (f"uds://{host}:0") - assert comm.extra_info == {} - msg = await comm.read() - msg["op"] = "pong" - await comm.write(msg) - await comm.close() - - listener = await UDSListener("localhost", handle_comm) - host, port = listener.get_host_port() - - assert host.endswith(".sock") - assert port == 0 # we fake port 0 when using UDS - - l = [] - - async def client_communicate(key, delay=0): - comm = await connect(listener.contact_address) - assert comm.peer_address == f"uds://{host}:0" - assert comm.extra_info == {} - await comm.write({"op": "ping", "data": key}) - if delay: - await asyncio.sleep(delay) - msg = await comm.read() - assert msg == {"op": "pong", "data": key} - l.append(key) - await comm.close() - - await client_communicate(key=1234) - - # Many clients at once - N = 100 - futures = [client_communicate(key=i, delay=0.05) for i in range(N)] - await asyncio.gather(*futures) - assert set(l) == {1234} | set(range(N)) - - listener.stop() - assert not os.path.exists(host) # assert socket deleted From 14b66709df429741c869c09eb1363ec652ca8ec7 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 10:08:48 +0200 Subject: [PATCH 05/25] uds:// -> unix:// --- distributed/comm/__init__.py | 2 +- distributed/comm/addressing.py | 4 ++-- distributed/comm/tests/test_comms.py | 8 ++++---- distributed/comm/tests/test_uds.py | 8 ++++---- distributed/comm/uds.py | 8 ++------ 5 files changed, 13 insertions(+), 17 deletions(-) diff --git a/distributed/comm/__init__.py b/distributed/comm/__init__.py index d3f16eb669c..1a2f5be6766 100644 --- a/distributed/comm/__init__.py +++ b/distributed/comm/__init__.py @@ -21,7 +21,7 @@ def _register_transports(): backends["tcp"] = tcp.TCPBackend() backends["tls"] = tcp.TLSBackend() - backends["uds"] = uds.UDSBackend() + backends["unix"] = uds.UDSBackend() try: # If `distributed-ucxx` is installed, it takes over the protocol="ucx" support diff --git a/distributed/comm/addressing.py b/distributed/comm/addressing.py index 8bb381d526b..59efc20c936 100644 --- a/distributed/comm/addressing.py +++ b/distributed/comm/addressing.py @@ -14,8 +14,8 @@ def parse_address(addr: str, strict: bool = False) -> tuple[str, str]: >>> parse_address('tcp://127.0.0.1') ('tcp', '127.0.0.1') - >>> parse_address('uds:///tmp/socket') - ('uds', '/tmp/socket') + >>> parse_address('unix:///tmp/socket') + ('unix', '/tmp/socket') If strict is set to true the address must have a scheme. """ diff --git a/distributed/comm/tests/test_comms.py b/distributed/comm/tests/test_comms.py index a0118422775..60ae2644d51 100644 --- a/distributed/comm/tests/test_comms.py +++ b/distributed/comm/tests/test_comms.py @@ -182,7 +182,7 @@ def test_get_address_host(tcp): assert f("tcp://127.0.0.1:123") == "127.0.0.1" assert f("inproc://%s/%d/123" % (get_ip(), os.getpid())) == get_ip() - assert f("uds://%2Ftmp%2Fdask.sock") == "%2Ftmp%2Fdask.sock" + assert f("unix://%2Ftmp%2Fdask.sock") == "%2Ftmp%2Fdask.sock" def test_resolve_address(tcp): @@ -204,7 +204,7 @@ def test_resolve_address(tcp): assert f("tcp://localhost:456") == "tcp://127.0.0.1:456" assert f("tls://localhost:456") == "tls://127.0.0.1:456" - assert f("uds://%2Ftmp%2Fdask.sock") == "uds://%2Ftmp%2Fdask.sock" + assert f("unix://%2Ftmp%2Fdask.sock") == "unix://%2Ftmp%2Fdask.sock" def test_get_local_address_for(tcp): @@ -298,7 +298,7 @@ async def test_uds_specific(uds): """ async def handle_comm(comm): - assert comm.peer_address == (f"uds://{host}:0") + assert comm.peer_address == (f"unix://{host}:0") assert comm.extra_info == {} msg = await comm.read() msg["op"] = "pong" @@ -315,7 +315,7 @@ async def handle_comm(comm): async def client_communicate(key, delay=0): comm = await connect(listener.contact_address) - assert comm.peer_address == f"uds://{host}:0" + assert comm.peer_address == f"unix://{host}:0" assert comm.extra_info == {} await comm.write({"op": "ping", "data": key}) if delay: diff --git a/distributed/comm/tests/test_uds.py b/distributed/comm/tests/test_uds.py index 026a93f546e..0c277683ad6 100644 --- a/distributed/comm/tests/test_uds.py +++ b/distributed/comm/tests/test_uds.py @@ -4,14 +4,14 @@ def test_registered(): - assert "uds" in backends - backend = get_backend("uds") + assert "unix" in backends + backend = get_backend("unix") assert isinstance(backend, UDSBackend) def test_parse_uds_address(): - addr = "uds:///tmp/dask-test.sock" + addr = "unix:///tmp/dask-test.sock" scheme, loc = parse_address(addr) - assert scheme == "uds" + assert scheme == "unix" assert loc == "/tmp/dask-test.sock" assert unparse_address(scheme, loc) == addr diff --git a/distributed/comm/uds.py b/distributed/comm/uds.py index 8587b136b21..bf4a16cfd9d 100644 --- a/distributed/comm/uds.py +++ b/distributed/comm/uds.py @@ -11,7 +11,6 @@ import dask -from distributed.comm.registry import backends from distributed.comm.tcp import ( MAX_BUFFER_SIZE, TCP, @@ -36,7 +35,7 @@ async def resolve( class UDSListener(TCPListener): - prefix = "uds://" + prefix = "unix://" comm_class = TCP def __init__( @@ -94,7 +93,7 @@ def stop(self): class UDSConnector(TCPConnector): client: ClassVar[TCPClient] = TCPClient(resolver=UnixSocketResolver()) - prefix = "uds://" + prefix = "unix://" comm_class = TCP @@ -107,6 +106,3 @@ def get_address_host(self, loc): def resolve_address(self, loc): return loc - - -backends["uds"] = UDSBackend() From 5bb2eea6d8f070bff65c09391ff19a77be90d493 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 10:20:48 +0200 Subject: [PATCH 06/25] Add some docs --- distributed/comm/uds.py | 10 ++++++++-- docs/source/communications.rst | 2 ++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/distributed/comm/uds.py b/distributed/comm/uds.py index bf4a16cfd9d..07eebf6fda4 100644 --- a/distributed/comm/uds.py +++ b/distributed/comm/uds.py @@ -26,7 +26,7 @@ class UnixSocketResolver(netutil.Resolver): - """A resolver for Unix Domain Sockets. Always returns socket type and pathname.""" + """A resolver for Unix Domain Sockets. This is used by tornado to lookup hostnames. For UDS, this should always return socket type and pathname (instead of a real DNS lookup).""" async def resolve( self, host: str, port: int, family: socket.AddressFamily = socket.AF_UNSPEC @@ -35,6 +35,8 @@ async def resolve( class UDSListener(TCPListener): + """A Listener for Unix Domain Sockets, based on the TCPListener class. Ensures the address is an absolute path instead of a hostname:port string.""" + prefix = "unix://" comm_class = TCP @@ -60,6 +62,9 @@ def get_host_port(self): return (self.bound_address, 0) # fake port 0 async def _handle_stream(self, stream, address): + """ + We override the super class's _handle_stream to pass in 0 as a fake port (it will be ignored anyway). + """ return await super()._handle_stream(stream, (self.bound_address, 0)) async def start(self): @@ -84,7 +89,6 @@ def stop(self): try: os.remove(self.bound_address) except OSError as e: - print(f"FAILED {e}") logger.debug( f"Attempted removal of socket at {self.bound_address} failed with error: {e}" ) @@ -98,6 +102,8 @@ class UDSConnector(TCPConnector): class UDSBackend(BaseTCPBackend): + """A Backend for Unix Domain Sockets. It overrides the TCP class's functions for parsing addresses, since UDS does not require port numbers.""" + _connector_class = UDSConnector _listener_class = UDSListener diff --git a/docs/source/communications.rst b/docs/source/communications.rst index b9406e6b3c5..3298536f25a 100644 --- a/docs/source/communications.rst +++ b/docs/source/communications.rst @@ -39,6 +39,8 @@ source tree: communication between endpoints as long as they are situated in the same process. +* ``unix`` is a transport using Unix Domain Sockets. Unix sockets bypass the system's network layer and are therefore more efficient than TCP sockets. Moreover, they do not use ports but instead listen at filesystem paths, and can be secured using normal file permissions. This makes them a safe choice in multi-user environments, without needing TLS. (Not available on Windows.) + Some URIs may be valid for listening but not for connecting. For example, the URI ``tcp://`` will listen on all IPv4 and IPv6 addresses and on an arbitrary port, but you cannot connect to that address. From ebdd90d2c8a703a27a1cca12cebb7cb5f0064f4f Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 10:28:51 +0200 Subject: [PATCH 07/25] Skip tests and don't register UDS backend on Windows --- distributed/comm/__init__.py | 4 +++- distributed/comm/tests/test_comms.py | 3 ++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/distributed/comm/__init__.py b/distributed/comm/__init__.py index 1a2f5be6766..ce5d7ac3baa 100644 --- a/distributed/comm/__init__.py +++ b/distributed/comm/__init__.py @@ -14,6 +14,7 @@ from distributed.comm.core import Comm, CommClosedError, connect, listen from distributed.comm.registry import backends from distributed.comm.utils import get_tcp_server_address, get_tcp_server_addresses +from distributed.compatibility import WINDOWS def _register_transports(): @@ -21,7 +22,8 @@ def _register_transports(): backends["tcp"] = tcp.TCPBackend() backends["tls"] = tcp.TLSBackend() - backends["unix"] = uds.UDSBackend() + if not WINDOWS: + backends["unix"] = uds.UDSBackend() try: # If `distributed-ucxx` is installed, it takes over the protocol="ucx" support diff --git a/distributed/comm/tests/test_comms.py b/distributed/comm/tests/test_comms.py index 60ae2644d51..41fb953f031 100644 --- a/distributed/comm/tests/test_comms.py +++ b/distributed/comm/tests/test_comms.py @@ -25,7 +25,7 @@ ) from distributed.comm.registry import backends, get_backend from distributed.comm.tcp import get_stream_address -from distributed.compatibility import asyncio_run +from distributed.compatibility import WINDOWS, asyncio_run from distributed.config import get_loop_factory from distributed.metrics import time from distributed.protocol import Serialized, deserialize, serialize, to_serialize @@ -291,6 +291,7 @@ async def client_communicate(key, delay=0): assert set(l) == {1234} | set(range(N)) +@pytest.mark.skipif(WINDOWS, reason="No unix sockets on Windows") @gen_test() async def test_uds_specific(uds): """ From de2a7ec656727ea0feba2dd3b6b5618e91b18366 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 11:20:14 +0200 Subject: [PATCH 08/25] Don't check for unix socket type on Windows --- distributed/comm/tcp.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/distributed/comm/tcp.py b/distributed/comm/tcp.py index fb5c2690990..7a91f3b5ac7 100644 --- a/distributed/comm/tcp.py +++ b/distributed/comm/tcp.py @@ -37,6 +37,7 @@ get_tcp_server_address, to_frames, ) +from distributed.compatibility import WINDOWS from distributed.protocol.utils import host_array, pack_frames_prelude, unpack_frames from distributed.system import MEMORY_LIMIT from distributed.utils import ensure_ip, ensure_memoryview, get_ip, nbytes @@ -59,7 +60,7 @@ def set_tcp_timeout(comm): """ Set kernel-level TCP timeout on the stream. """ - if comm.closed() or comm.socket.family is socket.AF_UNIX: + if comm.closed() or (not WINDOWS and comm.socket.family is socket.AF_UNIX): return timeout = dask.config.get("distributed.comm.timeouts.tcp") @@ -124,7 +125,7 @@ def get_stream_address(comm): if comm.closed(): raise CommClosedError() - if comm.socket.family is socket.AF_UNIX: + if not WINDOWS and comm.socket.family is socket.AF_UNIX: return comm.socket.getsockname() or comm.socket.getpeername() else: return unparse_host_port(*comm.socket.getsockname()[:2]) From 959bd1f953fbf24f3d6d33e4896647b3ae7440cf Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 11:23:47 +0200 Subject: [PATCH 09/25] Move get_uds_path() to distributed.utils Make get_uds_path() respect XDG_RUNTIME_DIR and dask.config['temporary-directory'] --- distributed/comm/uds.py | 2 +- distributed/comm/utils.py | 10 ---------- distributed/utils.py | 25 +++++++++++++++++++++++++ 3 files changed, 26 insertions(+), 11 deletions(-) diff --git a/distributed/comm/uds.py b/distributed/comm/uds.py index 07eebf6fda4..2018a55d56c 100644 --- a/distributed/comm/uds.py +++ b/distributed/comm/uds.py @@ -18,7 +18,7 @@ TCPConnector, TCPListener, ) -from distributed.comm.utils import ( +from distributed.utils import ( get_uds_path, ) diff --git a/distributed/comm/utils.py b/distributed/comm/utils.py index 83e55896534..e91ddc2c839 100644 --- a/distributed/comm/utils.py +++ b/distributed/comm/utils.py @@ -1,7 +1,6 @@ from __future__ import annotations import logging -import os import socket import dask @@ -125,12 +124,3 @@ def ensure_concrete_host(host, default_host=None): return default_host or get_ipv6() else: return host - - -def get_uds_path(address: str) -> str: - if os.path.isabs(address): - return str(address) - else: - import uuid - - return os.path.join("/tmp", f"{str(uuid.uuid4()).replace('-', '')}.sock") diff --git a/distributed/utils.py b/distributed/utils.py index caf748bc361..a91e9fd8522 100644 --- a/distributed/utils.py +++ b/distributed/utils.py @@ -1897,3 +1897,28 @@ def url_escape(url, *args, **kwargs): Escape a URL path segment. Cache results for better performance. """ return escape.url_escape(url, *args, **kwargs) + + +def get_uds_path(address: str | None) -> str: + """ + Take an address for a Unix Domain Socket and return an absolute path. + If address is already an absolute path (the user passed in a specific path), return it. + In all other cases, generate a random filename in one of these locations: + 1. $XDG_RUNTIME_DIR/dask (if set; will be created) + 2. dask.config["temporary-directory"] (if set; will be created) + 3. tempfile.gettempdir() + """ + if os.path.isabs(str(address)): + return str(address) + else: + xdg_dir = os.environ.get("XDG_RUNTIME_DIR", False) + if xdg_dir: + base_path = f"{xdg_dir}/dask" + # os.makedirs(base_path, exist_ok=True) + else: + base_path = dask.config.get("temporary-directory", tempfile.gettempdir()) + # os.makedirs(base_path, exist_ok=True) + + import secrets # use token_hex to generate a random filename + + return os.path.join(base_path, f"{secrets.token_hex()}.sock") From 5c8451ba63eb80549cdf1d0dec99e11003b27f48 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 11:24:47 +0200 Subject: [PATCH 10/25] Dashboard: allow passing in 'unix://' as dashboard address for random filename --- distributed/deploy/local.py | 2 ++ distributed/utils.py | 8 ++++++-- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/distributed/deploy/local.py b/distributed/deploy/local.py index e090f6649e0..a0900092221 100644 --- a/distributed/deploy/local.py +++ b/distributed/deploy/local.py @@ -60,6 +60,8 @@ class LocalCluster(SpecCluster): 'localhost:8787' or '0.0.0.0:8787'. Defaults to ':8787'. Set to ``None`` to disable the dashboard. Use ':0' for a random port. + Set to an absolute filesystem path to listen on a Unix Domain Socket at that path. + Set to 'unix://' to listen on a Unix socket with a random name, in the location configured by dask.config['temporary-directory'] or a default temporary directory. When specifying only a port like ':8787', the dashboard will bind to the given interface from the ``host`` parameter. If ``host`` is empty, binding will occur on all interfaces '0.0.0.0'. To avoid firewall issues when deploying locally, set ``host`` to 'localhost'. diff --git a/distributed/utils.py b/distributed/utils.py index a91e9fd8522..a51983875cd 100644 --- a/distributed/utils.py +++ b/distributed/utils.py @@ -1522,10 +1522,14 @@ def clean_dashboard_address(addrs: AnyType, default_listen_ip: str = "") -> list [{'address': '', 'port': 8787}, {'address': '', 'port': 8887}] >>> clean_dashboard_address(":8787,:8887") [{'address': '', 'port': 8787}, {'address': '', 'port': 8887}] + >>> clean_dashboard_address("/tmp/dashboard.sock") + [{'address': '/tmp/dashboard.sock', 'port': 0}] + >>> clean_dashboard_address("unix://") + [{'address': '/tmp/.sock', 'port': 0}] """ - if isinstance(addrs, str) and os.path.isabs(addrs): # unix socket - return [{"address": addrs, "port": 0}] + if isinstance(addrs, str) and (addrs.startswith("unix://") or os.path.isabs(addrs)): + return [{"address": get_uds_path(addrs), "port": 0}] if default_listen_ip == "0.0.0.0": default_listen_ip = "" # for IPV6 From bd52b6ed4cd88504a9826fe2ecc42ec995c1eb09 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 11:26:06 +0200 Subject: [PATCH 11/25] Set dashboard unix socket permissions to 600 --- distributed/node.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/distributed/node.py b/distributed/node.py index 76d2cc047e9..2e74c973c34 100644 --- a/distributed/node.py +++ b/distributed/node.py @@ -165,7 +165,7 @@ def start_http_server( if os.path.isabs(http_address["address"]): dashboard_socket = netutil.bind_unix_socket( - http_address["address"], mode=0o666 + http_address["address"], mode=0o600 ) self.http_server.add_socket(dashboard_socket) bound_addresses = [http_address["address"]] From 34f1ba2fceefafff643c3d6214f79a4dc6b1c88a Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 12:33:46 +0200 Subject: [PATCH 12/25] Fix get_uds_path --- distributed/utils.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/distributed/utils.py b/distributed/utils.py index a51983875cd..e110c70ebb3 100644 --- a/distributed/utils.py +++ b/distributed/utils.py @@ -52,7 +52,7 @@ import tblib.pickling_support from tornado import escape -from distributed.compatibility import asyncio_run +from distributed.compatibility import MACOS, asyncio_run from distributed.config import get_loop_factory try: @@ -1918,10 +1918,15 @@ def get_uds_path(address: str | None) -> str: xdg_dir = os.environ.get("XDG_RUNTIME_DIR", False) if xdg_dir: base_path = f"{xdg_dir}/dask" - # os.makedirs(base_path, exist_ok=True) else: - base_path = dask.config.get("temporary-directory", tempfile.gettempdir()) - # os.makedirs(base_path, exist_ok=True) + base_path = dask.config.get("temporary-directory") + if not base_path and not MACOS: + base_path = tempfile.gettempdir() + elif MACOS: # MacOS throws `OSError: AF_UNIX path too long` with tempfile.gettempdir() + print("MACOS") + base_path = f"/tmp/dask-{os.environ.get('USER')}" + + os.makedirs(base_path, exist_ok=True) import secrets # use token_hex to generate a random filename From 0db0f6ae39a69b31b208f324e99cff18464c782f Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 12:35:37 +0200 Subject: [PATCH 13/25] Add unix to --protocol help --- distributed/cli/dask_scheduler.py | 2 +- distributed/cli/dask_worker.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/distributed/cli/dask_scheduler.py b/distributed/cli/dask_scheduler.py index 7302027666c..dabeea5dbc9 100755 --- a/distributed/cli/dask_scheduler.py +++ b/distributed/cli/dask_scheduler.py @@ -37,7 +37,7 @@ help="Preferred network interface like 'eth0' or 'ib0'", ) @click.option( - "--protocol", type=str, default=None, help="Protocol like tcp, tls, or ucx" + "--protocol", type=str, default=None, help="Protocol like tcp, tls, unix, or ucx" ) @click.option( "--tls-ca-file", diff --git a/distributed/cli/dask_worker.py b/distributed/cli/dask_worker.py index 9c89569876d..c32d8338d81 100755 --- a/distributed/cli/dask_worker.py +++ b/distributed/cli/dask_worker.py @@ -116,7 +116,7 @@ "--interface", type=str, default=None, help="Network interface like 'eth0' or 'ib0'" ) @click.option( - "--protocol", type=str, default=None, help="Protocol like tcp, tls, or ucx" + "--protocol", type=str, default=None, help="Protocol like tcp, tls, unix, or ucx" ) @click.option("--nthreads", type=int, default=0, help="Number of threads per process.") @click.option( From 805dc71cc666967c2429a38c62e1b11a3e18de80 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 12:48:15 +0200 Subject: [PATCH 14/25] fix uds path --- distributed/utils.py | 1 - 1 file changed, 1 deletion(-) diff --git a/distributed/utils.py b/distributed/utils.py index e110c70ebb3..0b1b36aacee 100644 --- a/distributed/utils.py +++ b/distributed/utils.py @@ -1923,7 +1923,6 @@ def get_uds_path(address: str | None) -> str: if not base_path and not MACOS: base_path = tempfile.gettempdir() elif MACOS: # MacOS throws `OSError: AF_UNIX path too long` with tempfile.gettempdir() - print("MACOS") base_path = f"/tmp/dask-{os.environ.get('USER')}" os.makedirs(base_path, exist_ok=True) From c2ce79b6dd079fc9c17d0fbc7b6a7eb386f73da2 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 13:37:44 +0200 Subject: [PATCH 15/25] Remove unnecessarily added comments --- distributed/comm/addressing.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/distributed/comm/addressing.py b/distributed/comm/addressing.py index 59efc20c936..9f506a02142 100644 --- a/distributed/comm/addressing.py +++ b/distributed/comm/addressing.py @@ -39,7 +39,7 @@ def unparse_address(scheme: str, loc: str) -> str: Undo parse_address(). >>> unparse_address('tcp', '127.0.0.1') - 'tcp://127.0.0.1' # Example remains unchanged. + 'tcp://127.0.0.1' """ return f"{scheme}://{loc}" @@ -161,7 +161,7 @@ def get_local_address_for(addr: str) -> str: >>> get_local_address_for('tcp://8.8.8.8:1234') 'tcp://192.168.1.68' >>> get_local_address_for('tcp://127.0.0.1:1234') - 'tcp://127.0.0.1' # Example remains unchanged. + 'tcp://127.0.0.1' """ scheme, loc = parse_address(addr) backend = registry.get_backend(scheme) From 66fb708bd3784d05996ea58abc2e4e6bbcca2d14 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 28 Jul 2026 13:38:13 +0200 Subject: [PATCH 16/25] Add test and support for starting scheduler on a random unix socket --- distributed/cli/tests/test_dask_scheduler.py | 20 ++++++++++++++++++++ distributed/node.py | 4 ++-- 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/distributed/cli/tests/test_dask_scheduler.py b/distributed/cli/tests/test_dask_scheduler.py index dc5b7b14a4b..edd11641b85 100644 --- a/distributed/cli/tests/test_dask_scheduler.py +++ b/distributed/cli/tests/test_dask_scheduler.py @@ -352,6 +352,26 @@ def test_dashboard_port_zero(loop): with Client(f"tcp://127.0.0.1:{port}", loop=loop) as c: assert get_dashboard_port(c) > 0 +@pytest.mark.skipif(WINDOWS, reason="POSIX only") +def test_dashboard_unix_socket(loop): + pytest.importorskip("bokeh") + port = open_port() + with popen( + [ + sys.executable, + "-m", + "dask", + "scheduler", + "--host", + f"127.0.0.1:{port}", + "--dashboard-address", + "unix://", + ], + ): + with Client(f"tcp://127.0.0.1:{port}", loop=loop) as c: + assert get_dashboard_port(c) == 0 + assert c.dashboard_link.startswith("unix://") + assert c.dashboard_link.endswith(".sock") PRELOAD_TEXT = """ _scheduler_info = {} diff --git a/distributed/node.py b/distributed/node.py index 2e74c973c34..9ac8a5405c0 100644 --- a/distributed/node.py +++ b/distributed/node.py @@ -163,12 +163,12 @@ def start_http_server( change_port = False retries_left = 3 - if os.path.isabs(http_address["address"]): + if os.path.isabs(http_address["address"]): # unix socket dashboard_socket = netutil.bind_unix_socket( http_address["address"], mode=0o600 ) self.http_server.add_socket(dashboard_socket) - bound_addresses = [http_address["address"]] + bound_addresses = [(f"unix://{http_address["address"]}", 0)] else: while True: try: From d8756994bafd634d2423da4debd2d313e4e9db55 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Wed, 29 Jul 2026 15:37:37 +0200 Subject: [PATCH 17/25] refactor format_dashboard_link logic: allow setting unix socket path in service definition --- distributed/cli/tests/test_dask_scheduler.py | 27 ++++++++++++-- distributed/client.py | 12 ++----- distributed/deploy/cluster.py | 4 +-- distributed/node.py | 4 +-- distributed/objects.py | 4 +-- distributed/scheduler.py | 16 ++++++++- distributed/tests/test_utils.py | 2 ++ distributed/utils.py | 37 ++++++++++++++++---- 8 files changed, 79 insertions(+), 27 deletions(-) diff --git a/distributed/cli/tests/test_dask_scheduler.py b/distributed/cli/tests/test_dask_scheduler.py index edd11641b85..ced9f339041 100644 --- a/distributed/cli/tests/test_dask_scheduler.py +++ b/distributed/cli/tests/test_dask_scheduler.py @@ -4,6 +4,7 @@ import os import shutil import signal +import socket import subprocess import sys import tempfile @@ -352,10 +353,23 @@ def test_dashboard_port_zero(loop): with Client(f"tcp://127.0.0.1:{port}", loop=loop) as c: assert get_dashboard_port(c) > 0 + @pytest.mark.skipif(WINDOWS, reason="POSIX only") def test_dashboard_unix_socket(loop): pytest.importorskip("bokeh") port = open_port() + + def _connect_unix_socket(path): + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + s.connect(path) + except Exception: + return False + else: + return True + finally: + s.close() + with popen( [ sys.executable, @@ -369,9 +383,16 @@ def test_dashboard_unix_socket(loop): ], ): with Client(f"tcp://127.0.0.1:{port}", loop=loop) as c: - assert get_dashboard_port(c) == 0 - assert c.dashboard_link.startswith("unix://") - assert c.dashboard_link.endswith(".sock") + proto, rest = c.dashboard_link.split("://") + assert proto == "http+unix" + assert rest.endswith(".sock/status") + assert ":" not in rest # no port + + path = rest.split("/status")[0] + assert os.path.isabs(path) + assert os.path.exists(path) + assert _connect_unix_socket(path) + PRELOAD_TEXT = """ _scheduler_info = {} diff --git a/distributed/client.py b/distributed/client.py index 5f14cb4911b..8ae44daad4d 100644 --- a/distributed/client.py +++ b/distributed/client.py @@ -1337,16 +1337,10 @@ def dashboard_link(self): scheduler, info = self._get_scheduler_info(n_workers=0) if scheduler is None: return None - else: - protocol, rest = scheduler.address.split("://") - - port = info["services"]["dashboard"] - if protocol == "inproc": - host = "localhost" - else: - host = rest.split(":")[0] - return format_dashboard_link(host, port) + return format_dashboard_link( + scheduler.address, info["services"]["dashboard"] + ) def _get_scheduler_info(self, n_workers): from distributed.scheduler import Scheduler diff --git a/distributed/deploy/cluster.py b/distributed/deploy/cluster.py index 8d80c963e98..05bd946fe43 100644 --- a/distributed/deploy/cluster.py +++ b/distributed/deploy/cluster.py @@ -123,7 +123,6 @@ def name(self, name): async def _start(self): comm = await self.scheduler_comm.live_comm() comm.name = "Cluster worker status" - await comm.write({"op": "subscribe_worker_status"}) self.scheduler_info = SchedulerInfo(await comm.read()) self._watch_worker_status_comm = comm self._watch_worker_status_task = asyncio.ensure_future( @@ -361,8 +360,7 @@ def dashboard_link(self): except KeyError: return "" else: - host = self.scheduler_address.split("://")[1].split("/")[0].split(":")[0] - return format_dashboard_link(host, port) + return format_dashboard_link(self.scheduler_address, port) def _scaling_status(self): if self._adaptive and self._adaptive.periodic_callback: diff --git a/distributed/node.py b/distributed/node.py index 9ac8a5405c0..98c0b593a4b 100644 --- a/distributed/node.py +++ b/distributed/node.py @@ -163,12 +163,12 @@ def start_http_server( change_port = False retries_left = 3 - if os.path.isabs(http_address["address"]): # unix socket + if os.path.isabs(http_address["address"]): # unix socket dashboard_socket = netutil.bind_unix_socket( http_address["address"], mode=0o600 ) self.http_server.add_socket(dashboard_socket) - bound_addresses = [(f"unix://{http_address["address"]}", 0)] + bound_addresses = [(f"unix://{http_address['address']}", 0)] else: while True: try: diff --git a/distributed/objects.py b/distributed/objects.py index 53b7b91565a..eddc386fbe2 100644 --- a/distributed/objects.py +++ b/distributed/objects.py @@ -7,8 +7,6 @@ from dask.widgets import get_environment, get_template -from distributed.utils import format_dashboard_link - class HasWhat(dict): """A dictionary of all workers and which keys that worker has.""" @@ -35,7 +33,7 @@ def _format_dashboard_address(server): if "host" in server else urlparse(server["address"]).hostname ) - return format_dashboard_link(host, server["services"]["dashboard"]) + return host, server["services"]["dashboard"] except KeyError: return None diff --git a/distributed/scheduler.py b/distributed/scheduler.py index 8f4ba87da3f..85c2a9df16c 100644 --- a/distributed/scheduler.py +++ b/distributed/scheduler.py @@ -12,6 +12,7 @@ import os import pickle import random +import socket import textwrap import uuid import warnings @@ -22,6 +23,7 @@ Callable, Collection, Container, + Generator, Hashable, Iterable, Iterator, @@ -94,6 +96,7 @@ unparse_host_port, ) from distributed.comm.addressing import addresses_from_user_args +from distributed.compatibility import WINDOWS from distributed.core import ( ErrorMessage, OKMessage, @@ -4279,13 +4282,24 @@ def _repr_html_(self) -> str: def identity(self, n_workers: int = -1) -> dict[str, Any]: """Basic information about ourselves and our cluster""" + + def _map_services_to_uri(services: dict) -> Generator[tuple[str, int | str]]: + """the services dict contains k/v-pairs where v is either an int port number, of a unix socket url""" + for k, v in services.items(): + s = list(v._sockets.values())[0] + if not WINDOWS and s.family is socket.AF_UNIX: + yield k, f"unix://{s.getsockname()}" + else: + yield k, v.port + if n_workers == -1: n_workers = len(self.workers) + d = { "type": type(self).__name__, "id": str(self.id), "address": self.address, - "services": {key: v.port for (key, v) in self.services.items()}, + "services": {k: v for k, v in _map_services_to_uri(self.services)}, "started": self.time_started, "n_workers": len(self.workers), "total_threads": self.total_nthreads, diff --git a/distributed/tests/test_utils.py b/distributed/tests/test_utils.py index 1ebeafe2992..ba53566055a 100644 --- a/distributed/tests/test_utils.py +++ b/distributed/tests/test_utils.py @@ -630,6 +630,8 @@ def test_format_dashboard_link(): assert "host" in format_dashboard_link("host", 1234) assert "1234" in format_dashboard_link("host", 1234) + assert format_dashboard_link("localhost", "/tmp/dashboard.sock") == "http+unix:///tmp/dashboard.sock/status" + try: os.environ["host"] = "hello" assert "hello" not in format_dashboard_link("host", 1234) diff --git a/distributed/utils.py b/distributed/utils.py index 0b1b36aacee..e09ecf7d5b9 100644 --- a/distributed/utils.py +++ b/distributed/utils.py @@ -1267,12 +1267,37 @@ def warn_on_duration(duration: str | float | timedelta, msg: str) -> Generator[N warnings.warn(msg.format(duration=diff), stacklevel=2) -def format_dashboard_link(host, port): - template = dask.config.get("distributed.dashboard.link") - if dask.config.get("distributed.scheduler.dashboard.tls.cert"): - scheme = "https" - else: - scheme = "http" +def format_dashboard_link(process_address: str, port_or_uri: str | int) -> str: + """Return a formatted link to the dashboard based on: + `process_address`: the address of the process (e.g. Scheduler) that is calling this. + `port_or_uri`: value of an entry in a Client, Cluster, or Scheduler's `services` list. Can be either a port, in which case the hostname is assumed to be the same as the hostname for `process_address`. Or can be an absolute path for a service that listens on a Unix Domain Socket. + """ + if isinstance(port_or_uri, str) and ( + port_or_uri.startswith("unix://") or os.path.isabs(port_or_uri) + ): # unix socket + host = port_or_uri[7:] if port_or_uri.startswith("unix://") else port_or_uri + port = None + scheme = "http+unix" + template = "{scheme}://{host}/status" + else: # no unix socket + port = port_or_uri + try: + protocol, rest = process_address.split("://") + except ValueError: # no protocol prefix given + protocol, rest = "", process_address + + if protocol == "inproc": + host = "localhost" + else: + host = rest.split(":")[0] + + if dask.config.get("distributed.scheduler.dashboard.tls.cert"): + scheme = "https" + else: + scheme = "http" + + template = dask.config.get("distributed.dashboard.link") + return template.format( **toolz.merge(os.environ, dict(scheme=scheme, host=host, port=port)) ) From e5fc50dc69fcc89f4d2f27a0da4351dcbe631c9d Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Wed, 29 Jul 2026 17:24:01 +0200 Subject: [PATCH 18/25] Add test for get_uds_path() --- distributed/tests/test_utils.py | 54 ++++++++++++++++++++++++++++++++- distributed/utils.py | 27 +++++++++++------ 2 files changed, 70 insertions(+), 11 deletions(-) diff --git a/distributed/tests/test_utils.py b/distributed/tests/test_utils.py index ba53566055a..9ef5218cd09 100644 --- a/distributed/tests/test_utils.py +++ b/distributed/tests/test_utils.py @@ -8,6 +8,7 @@ import multiprocessing import os import queue +import shutil import socket import sys import traceback @@ -42,6 +43,7 @@ get_ip_interface, get_mp_context, get_traceback, + get_uds_path, is_kernel, is_valid_xml, iscoroutinefunction, @@ -630,7 +632,10 @@ def test_format_dashboard_link(): assert "host" in format_dashboard_link("host", 1234) assert "1234" in format_dashboard_link("host", 1234) - assert format_dashboard_link("localhost", "/tmp/dashboard.sock") == "http+unix:///tmp/dashboard.sock/status" + assert ( + format_dashboard_link("localhost", "/tmp/dashboard.sock") + == "http+unix:///tmp/dashboard.sock/status" + ) try: os.environ["host"] = "hello" @@ -1088,3 +1093,50 @@ def test_tuple_comparable_eq(obj1, obj2, expected): def test_tuple_comparable_error(): with pytest.raises(ValueError): TupleComparable("string") + + +@pytest.mark.parametrize( + "env_var", + [ + "XDG_RUNTIME_DIR", + "DASK_TEMPORARY_DIRECTORY", + None, + ], +) +@pytest.mark.parametrize( + "arg,expectation", + [ + ["", None], + ["localhost", None], + ["unix://", None], + ["unix:///dask.sock", "/dask.sock"], + ["/dask.sock", "/dask.sock"], + ], +) +def test_get_uds_path(request, tmp_path, env_var, arg, expectation, monkeypatch): + if env_var: + base_path = str(tmp_path) + monkeypatch.setenv(env_var, base_path) + if ( + env_var == "DASK_TEMPORARY_DIRECTORY" + ): # hack: setting this env var doesn't have effect because the dask config is already loaded at this point + monkeypatch.setitem(dask.config.config, "temporary-directory", base_path) + elif MACOS: # on MACOS we don't use tempfile.gettempdir + base_path = f"/tmp/dask-{os.environ.get('USER')}" + else: # a path will be created using tempfile.gettempdir + base_path = f"/tmp/pytest/dask/{request.node.name}" + import tempfile + + monkeypatch.setattr(tempfile, "gettempdir", lambda: base_path) + + try: + result = get_uds_path(arg.replace("", base_path)) + + if expectation: # we have an explicit absolute path to check for + assert result == expectation.replace("", base_path) + else: + assert result.startswith(f"{base_path}/") + assert result.endswith(".sock") + assert os.path.exists(os.path.dirname(result)) + finally: + shutil.rmtree(base_path, ignore_errors=True) diff --git a/distributed/utils.py b/distributed/utils.py index e09ecf7d5b9..02e895351da 100644 --- a/distributed/utils.py +++ b/distributed/utils.py @@ -1928,29 +1928,36 @@ def url_escape(url, *args, **kwargs): return escape.url_escape(url, *args, **kwargs) -def get_uds_path(address: str | None) -> str: +def get_uds_path(address: str | None, default_dir: str = "") -> str: """ Take an address for a Unix Domain Socket and return an absolute path. - If address is already an absolute path (the user passed in a specific path), return it. + If address is already an absolute path (possibly prefixed by unix://), return it. Underlying directories will not be created. In all other cases, generate a random filename in one of these locations: + 1. default_dir 1. $XDG_RUNTIME_DIR/dask (if set; will be created) 2. dask.config["temporary-directory"] (if set; will be created) - 3. tempfile.gettempdir() + 3. tempfile.gettempdir() (will be created) """ - if os.path.isabs(str(address)): - return str(address) + addr = str(address) + if addr.startswith("unix://"): + addr = addr[7:] + + if os.path.isabs(addr): + return addr else: xdg_dir = os.environ.get("XDG_RUNTIME_DIR", False) if xdg_dir: base_path = f"{xdg_dir}/dask" else: base_path = dask.config.get("temporary-directory") - if not base_path and not MACOS: - base_path = tempfile.gettempdir() - elif MACOS: # MacOS throws `OSError: AF_UNIX path too long` with tempfile.gettempdir() - base_path = f"/tmp/dask-{os.environ.get('USER')}" + if not base_path: + if MACOS: + # MacOS throws `OSError: AF_UNIX path too long` with tempfile.gettempdir(), so use a hardcoded path under /tmp/ + base_path = f"/tmp/dask-{os.environ.get('USER')}" + else: + base_path = tempfile.gettempdir() - os.makedirs(base_path, exist_ok=True) + os.makedirs(base_path, mode=0o700, exist_ok=True) import secrets # use token_hex to generate a random filename From d2e191a6cf340e5f5d4bf1e226c2201f45bdc7a7 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Wed, 29 Jul 2026 18:10:46 +0200 Subject: [PATCH 19/25] Restore wrongly deleted line --- distributed/deploy/cluster.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/distributed/deploy/cluster.py b/distributed/deploy/cluster.py index 05bd946fe43..8d80c963e98 100644 --- a/distributed/deploy/cluster.py +++ b/distributed/deploy/cluster.py @@ -123,6 +123,7 @@ def name(self, name): async def _start(self): comm = await self.scheduler_comm.live_comm() comm.name = "Cluster worker status" + await comm.write({"op": "subscribe_worker_status"}) self.scheduler_info = SchedulerInfo(await comm.read()) self._watch_worker_status_comm = comm self._watch_worker_status_task = asyncio.ensure_future( @@ -360,7 +361,8 @@ def dashboard_link(self): except KeyError: return "" else: - return format_dashboard_link(self.scheduler_address, port) + host = self.scheduler_address.split("://")[1].split("/")[0].split(":")[0] + return format_dashboard_link(host, port) def _scaling_status(self): if self._adaptive and self._adaptive.periodic_callback: From 42a25619759f17f6182ce5f0430561895aa9d910 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Wed, 29 Jul 2026 18:16:44 +0200 Subject: [PATCH 20/25] Remove dashboard socket before binding. This is easier than removing on stop. It technically introduces a race condition, but so would removing on stop, I think. --- distributed/node.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/distributed/node.py b/distributed/node.py index 98c0b593a4b..500223ee96c 100644 --- a/distributed/node.py +++ b/distributed/node.py @@ -164,6 +164,10 @@ def start_http_server( retries_left = 3 if os.path.isabs(http_address["address"]): # unix socket + try: # remove any old sockets + os.remove(http_address["address"]) + except OSError: + pass dashboard_socket = netutil.bind_unix_socket( http_address["address"], mode=0o600 ) From 0bf2f6e7b596e182a3025ae54e7da07b80aa8858 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Thu, 30 Jul 2026 10:39:58 +0200 Subject: [PATCH 21/25] Fix test_get_uds_path on ubuntu. On Linux we need to unset XDG_RUNTIME_DIR when we're not testing it. --- distributed/tests/test_utils.py | 33 +++++++++++++++++++++------------ distributed/utils.py | 4 ++-- 2 files changed, 23 insertions(+), 14 deletions(-) diff --git a/distributed/tests/test_utils.py b/distributed/tests/test_utils.py index 9ef5218cd09..2e122ea3531 100644 --- a/distributed/tests/test_utils.py +++ b/distributed/tests/test_utils.py @@ -1114,28 +1114,37 @@ def test_tuple_comparable_error(): ], ) def test_get_uds_path(request, tmp_path, env_var, arg, expectation, monkeypatch): + if env_var != "XDG_RUNTIME_DIR": + # on Ubuntu this env var is always set, so we need to unset it to test the other options + monkeypatch.setenv("XDG_RUNTIME_DIR", "") + if env_var: base_path = str(tmp_path) monkeypatch.setenv(env_var, base_path) - if ( - env_var == "DASK_TEMPORARY_DIRECTORY" - ): # hack: setting this env var doesn't have effect because the dask config is already loaded at this point - monkeypatch.setitem(dask.config.config, "temporary-directory", base_path) - elif MACOS: # on MACOS we don't use tempfile.gettempdir + elif MACOS: + # on MACOS we don't use tempfile.gettempdir base_path = f"/tmp/dask-{os.environ.get('USER')}" - else: # a path will be created using tempfile.gettempdir - base_path = f"/tmp/pytest/dask/{request.node.name}" + else: + # a path will be created using tempfile.gettempdir + base_path = f"/tmp/pytest/{request.node.name}" import tempfile monkeypatch.setattr(tempfile, "gettempdir", lambda: base_path) - try: - result = get_uds_path(arg.replace("", base_path)) + if env_var == "DASK_TEMPORARY_DIRECTORY": + # hack: setting this env var doesn't have effect within this test because the dask config is already loaded at this point + monkeypatch.setitem(dask.config.config, "temporary-directory", base_path) - if expectation: # we have an explicit absolute path to check for - assert result == expectation.replace("", base_path) + arg = arg.replace("", base_path) + expectation = expectation.replace("", base_path) if expectation else "" + + try: + result = get_uds_path(arg) + if os.path.isabs(str(expectation)): + assert result == expectation else: - assert result.startswith(f"{base_path}/") + # we don't know what the socket filename is, but it should live under result_base_path + assert result.startswith(f"{base_path}/dask_run") assert result.endswith(".sock") assert os.path.exists(os.path.dirname(result)) finally: diff --git a/distributed/utils.py b/distributed/utils.py index 02e895351da..c3fb2590b39 100644 --- a/distributed/utils.py +++ b/distributed/utils.py @@ -1941,13 +1941,12 @@ def get_uds_path(address: str | None, default_dir: str = "") -> str: addr = str(address) if addr.startswith("unix://"): addr = addr[7:] - if os.path.isabs(addr): return addr else: xdg_dir = os.environ.get("XDG_RUNTIME_DIR", False) if xdg_dir: - base_path = f"{xdg_dir}/dask" + base_path = f"{xdg_dir}" else: base_path = dask.config.get("temporary-directory") if not base_path: @@ -1957,6 +1956,7 @@ def get_uds_path(address: str | None, default_dir: str = "") -> str: else: base_path = tempfile.gettempdir() + base_path = f"{base_path}/dask_run" os.makedirs(base_path, mode=0o700, exist_ok=True) import secrets # use token_hex to generate a random filename From a5431bc59be4dbab2421b9100209080dd6d182ec Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Thu, 30 Jul 2026 14:23:32 +0200 Subject: [PATCH 22/25] Fix resolver tests --- distributed/comm/tests/test_comms.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/distributed/comm/tests/test_comms.py b/distributed/comm/tests/test_comms.py index 41fb953f031..096d861a36f 100644 --- a/distributed/comm/tests/test_comms.py +++ b/distributed/comm/tests/test_comms.py @@ -182,7 +182,8 @@ def test_get_address_host(tcp): assert f("tcp://127.0.0.1:123") == "127.0.0.1" assert f("inproc://%s/%d/123" % (get_ip(), os.getpid())) == get_ip() - assert f("unix://%2Ftmp%2Fdask.sock") == "%2Ftmp%2Fdask.sock" + if not WINDOWS: + assert f("unix:///tmp/dask.sock") == "/tmp/dask.sock" def test_resolve_address(tcp): @@ -204,7 +205,8 @@ def test_resolve_address(tcp): assert f("tcp://localhost:456") == "tcp://127.0.0.1:456" assert f("tls://localhost:456") == "tls://127.0.0.1:456" - assert f("unix://%2Ftmp%2Fdask.sock") == "unix://%2Ftmp%2Fdask.sock" + if not WINDOWS: + assert f("unix:///tmp/dask.sock") == "unix:///tmp/dask.sock" def test_get_local_address_for(tcp): From 558101d1c60a320e123880523b3bf083d9e87e16 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Thu, 30 Jul 2026 14:22:10 +0200 Subject: [PATCH 23/25] Fix get_uds_path method signature --- distributed/utils.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/distributed/utils.py b/distributed/utils.py index c3fb2590b39..2779094fe79 100644 --- a/distributed/utils.py +++ b/distributed/utils.py @@ -1928,13 +1928,12 @@ def url_escape(url, *args, **kwargs): return escape.url_escape(url, *args, **kwargs) -def get_uds_path(address: str | None, default_dir: str = "") -> str: +def get_uds_path(address: str | None = None) -> str: """ Take an address for a Unix Domain Socket and return an absolute path. If address is already an absolute path (possibly prefixed by unix://), return it. Underlying directories will not be created. - In all other cases, generate a random filename in one of these locations: - 1. default_dir - 1. $XDG_RUNTIME_DIR/dask (if set; will be created) + In all other cases, generate a random filename in 'dask_run' under one of these locations: + 1. $XDG_RUNTIME_DIR (set; will be created) 2. dask.config["temporary-directory"] (if set; will be created) 3. tempfile.gettempdir() (will be created) """ From 04a14d6609350f72e2867f39769e8295c15cb764 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Thu, 30 Jul 2026 14:50:02 +0200 Subject: [PATCH 24/25] Fix UDS backend: ensure we return a filesystem path on host lookup --- distributed/comm/uds.py | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/distributed/comm/uds.py b/distributed/comm/uds.py index 2018a55d56c..c25e12edbc8 100644 --- a/distributed/comm/uds.py +++ b/distributed/comm/uds.py @@ -11,10 +11,10 @@ import dask +from distributed.comm.registry import Backend from distributed.comm.tcp import ( MAX_BUFFER_SIZE, TCP, - BaseTCPBackend, TCPConnector, TCPListener, ) @@ -101,14 +101,30 @@ class UDSConnector(TCPConnector): comm_class = TCP -class UDSBackend(BaseTCPBackend): +class UDSBackend(Backend): """A Backend for Unix Domain Sockets. It overrides the TCP class's functions for parsing addresses, since UDS does not require port numbers.""" _connector_class = UDSConnector _listener_class = UDSListener + def get_connector(self): + return self._connector_class() + + def get_listener(self, loc, handle_comm, deserialize, **connection_args): + return self._listener_class(loc, handle_comm, deserialize, **connection_args) + def get_address_host(self, loc): - return loc.split(":")[0] + path = loc.split("unix://")[-1] + if os.path.isabs(path): + return path + else: + # something like `unix://127.0.0.1:0` was passed in + # this happens when a cluster sets protocl to 'unix', but doesn't explicitly override the default host and port + # in this case, return a new uds socket path + return get_uds_path(path) def resolve_address(self, loc): return loc + + def get_local_address_for(self, loc): + return loc From 588d58f9e3b7ad8d5d5ff4f6a4a778c7d2833674 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Thu, 30 Jul 2026 15:05:07 +0200 Subject: [PATCH 25/25] Skip all UDS tests on Windows --- distributed/comm/tests/test_uds.py | 6 ++++++ distributed/tests/test_utils.py | 10 ++++++---- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/distributed/comm/tests/test_uds.py b/distributed/comm/tests/test_uds.py index 0c277683ad6..a2c5b29e3a1 100644 --- a/distributed/comm/tests/test_uds.py +++ b/distributed/comm/tests/test_uds.py @@ -1,14 +1,20 @@ +import pytest + +from distributed.compatibility import WINDOWS + from distributed.comm.addressing import parse_address, unparse_address from distributed.comm.registry import backends, get_backend from distributed.comm.uds import UDSBackend +@pytest.mark.skipif(WINDOWS, reason="No unix sockets on Windows") def test_registered(): assert "unix" in backends backend = get_backend("unix") assert isinstance(backend, UDSBackend) +@pytest.mark.skipif(WINDOWS, reason="No unix sockets on Windows") def test_parse_uds_address(): addr = "unix:///tmp/dask-test.sock" scheme, loc = parse_address(addr) diff --git a/distributed/tests/test_utils.py b/distributed/tests/test_utils.py index 2e122ea3531..5810145e709 100644 --- a/distributed/tests/test_utils.py +++ b/distributed/tests/test_utils.py @@ -632,10 +632,11 @@ def test_format_dashboard_link(): assert "host" in format_dashboard_link("host", 1234) assert "1234" in format_dashboard_link("host", 1234) - assert ( - format_dashboard_link("localhost", "/tmp/dashboard.sock") - == "http+unix:///tmp/dashboard.sock/status" - ) + if not WINDOWS: + assert ( + format_dashboard_link("localhost", "/tmp/dashboard.sock") + == "http+unix:///tmp/dashboard.sock/status" + ) try: os.environ["host"] = "hello" @@ -1113,6 +1114,7 @@ def test_tuple_comparable_error(): ["/dask.sock", "/dask.sock"], ], ) +@pytest.mark.skipif(WINDOWS, reason="No unix sockets on Windows") def test_get_uds_path(request, tmp_path, env_var, arg, expectation, monkeypatch): if env_var != "XDG_RUNTIME_DIR": # on Ubuntu this env var is always set, so we need to unset it to test the other options