From ac297f50bf0bfca34c494bd93da916b55d457db2 Mon Sep 17 00:00:00 2001 From: DPeterK Date: Fri, 4 Jun 2021 13:02:45 +0100 Subject: [PATCH 01/28] Add local env cluster --- distributed/deploy/local_env.py | 292 ++++++++++++++++++++++++++++++++ 1 file changed, 292 insertions(+) create mode 100644 distributed/deploy/local_env.py diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py new file mode 100644 index 00000000000..4c579ff3dd3 --- /dev/null +++ b/distributed/deploy/local_env.py @@ -0,0 +1,292 @@ +import asyncio +import logging +import sys +import weakref +from typing import List, Union + +import dask +import dask.config + +from ..core import Status +from ..scheduler import Scheduler as _Scheduler +from ..utils import cli_keywords +from ..worker import Worker as _Worker +from .spec import ProcessInterface, SpecCluster + +logger = logging.getLogger(__name__) + + +class Process(ProcessInterface): + """A superclass for SSH Workers and Nannies + + See Also + -------- + Worker + Scheduler + """ + + def __init__(self, **kwargs): + self.proc = None + super().__init__(**kwargs) + + async def start(self): + weakref.finalize( + self, self.proc.kill + ) + await super().start() + + async def close(self): + self.proc.kill() + await super().close() + + def __repr__(self): + return "" % (type(self).__name__, self.status) + + +class Worker(Process): + """A Remote Dask Worker run by a specified Python executable + + Parameters + ---------- + scheduler: str + The address of the scheduler + python_exe: str + Full path to Python executable to run this worker + connect_options: dict + kwargs to be passed to asyncio subprocess connections + kwargs: dict + These will be passed through the dask-worker CLI to the + dask.distributed.Worker class + worker_module: str + The python module to run to start the worker + name: str + Optionally specify a name for this worker + """ + + def __init__( + self, + scheduler: str, + python_exe: str, + connect_options: dict, + kwargs: dict, + worker_module="distributed.cli.dask_worker", + name=None, + ): + super().__init__() + + self.scheduler = scheduler + self.python_exe = python_exe + self.connect_options = connect_options + self.kwargs = kwargs + self.worker_module = worker_module + self.name = name + + async def start(self): + set_env = await _set_env_helper(self.connect_options) + + cmd = " ".join( + [ + set_env, + self.python_exe, + "-m", + self.worker_module, + self.scheduler, + "--name", + str(self.name), + ] + + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module) + ) + self.proc = await asyncio.create_subprocess_shell( + cmd, + stderr=asyncio.subprocess.PIPE, + **self.connect_options + ) + + # We watch stderr in order to get the address, then we return + while True: + _, line = await self.proc.communicate() + if not line.decode().strip(): + raise Exception("Worker failed to start") + else: + line = line.decode() + logger.info(line.strip()) + if "worker at" in line: + self.address = line.split("worker at:")[1].strip() + self.status = Status.running + break + logger.debug("%s", line) + await super().start() + + +class Scheduler(Process): + """A Remote Dask Scheduler run by a specified Python executable + + Parameters + ---------- + python_exe: str + Full path to Python executable to run this scheduler + connect_options: dict + kwargs to be passed to asyncio subprocess connections + kwargs: dict + These will be passed through the dask-scheduler CLI to the + dask.distributed.Scheduler class + """ + + def __init__(self, python_exe: str, connect_options: dict, kwargs: dict): + super().__init__() + + self.python_exe = python_exe + self.kwargs = kwargs + self.connect_options = connect_options + + async def start(self): + logger.debug("Created Scheduler") + + set_env = await _set_env_helper(self.connect_options) + + cmd = " ".join( + [ + set_env, + self.python_exe, + "-m", + "distributed.cli.dask_scheduler" + ] + + cli_keywords(self.kwargs, cls=_Scheduler) + ) + self.proc = await asyncio.create_subprocess_shell( + cmd, + stderr=asyncio.subprocess.PIPE, + **self.connect_options + ) + + # We watch stderr in order to get the address, then we return + while True: + _, line = await self.proc.communicate() + if not line.decode().strip(): + raise Exception("Worker failed to start") + else: + line = line.decode() + logger.info(line.strip()) + if "Scheduler at" in line: + self.address = line.split("Scheduler at:")[1].strip() + break + logger.debug("%s", line) + await super().start() + + +async def _set_env_helper(connect_options: dict): + proc = await asyncio.create_subprocess_shell("uname", **connect_options) + await proc.communicate() + if proc.returncode == 0: + set_env = 'env DASK_INTERNAL_INHERIT_CONFIG="{}"'.format( + dask.config.serialize(dask.config.global_config) + ) + else: + proc = await asyncio.create_subprocess_shell( + "cmd /c ver", + **connect_options + ) + await proc.communicate() + if proc.returncode == 0: + set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format( + dask.config.serialize(dask.config.global_config) + ) + else: + raise Exception( + "Scheduler failed to set DASK_INTERNAL_INHERIT_CONFIG variable " + ) + return set_env + +def LocalEnvCluster( + python_exe: str, + connect_options: Union[List[dict], dict] = {}, + worker_options: dict = {}, + scheduler_options: dict = {}, + worker_module: str = "distributed.cli.dask_worker", + **kwargs, +): + """Deploy a Dask cluster that utilises a different Python executable + + The SSHCluster function deploys a Dask Scheduler and Workers for you on a + set of machine addresses that you provide. The first address will be used + for the scheduler while the rest will be used for the workers (feel free to + repeat the first hostname if you want to have the scheduler and worker + co-habitate one machine.) + + You may configure the scheduler and workers by passing + ``scheduler_options`` and ``worker_options`` dictionary keywords. See the + ``dask.distributed.Scheduler`` and ``dask.distributed.Worker`` classes for + details on the available options, but the defaults should work in most + situations. + + You may configure your use of SSH itself using the ``connect_options`` + keyword, which passes values to the ``asyncssh.connect`` function. For + more information on these see the documentation for the ``asyncssh`` + library https://asyncssh.readthedocs.io . + + Parameters + ---------- + python_exe : str + Full path to another Python executable, for example from another + conda env or Python venv. + connect_options : dict or list of dict, optional + Keywords to pass through to :func:`asyncssh.connect`. + This could include things such as ``port``, ``username``, ``password`` + or ``known_hosts``. See docs for :func:`asyncssh.connect` and + :class:`asyncssh.SSHClientConnectionOptions` for full information. + If a list it must have the same length as ``hosts``. + worker_options : dict, optional + Keywords to pass on to workers. + scheduler_options : dict, optional + Keywords to pass on to scheduler. + worker_module : str, optional + Python module to call to start the worker. + + Examples + -------- + >>> from dask.distributed import Client, SSHCluster + >>> cluster = SSHCluster( + ... ["localhost", "localhost", "localhost", "localhost"], + ... connect_options={"known_hosts": None}, + ... worker_options={"nthreads": 2}, + ... scheduler_options={"port": 0, "dashboard_address": ":8797"} + ... ) + >>> client = Client(cluster) + + An example using a different worker module, in particular the + ``dask-cuda-worker`` command from the ``dask-cuda`` project. + + >>> from dask.distributed import Client, SSHCluster + >>> cluster = SSHCluster( + ... ["localhost", "hostwithgpus", "anothergpuhost"], + ... connect_options={"known_hosts": None}, + ... scheduler_options={"port": 0, "dashboard_address": ":8797"}, + ... worker_module='dask_cuda.dask_cuda_worker') + >>> client = Client(cluster) + + See Also + -------- + dask.distributed.Scheduler + dask.distributed.Worker + asyncssh.connect + """ + scheduler = { + "cls": Scheduler, + "options": { + "python_exe": python_exe, + "connect_options": connect_options, + "kwargs": scheduler_options, + }, + } + workers = { + 0: { + "cls": Worker, + "options": { + "python_exe": python_exe, + "connect_options": connect_options, + "kwargs": worker_options, + "worker_module": worker_module, + }, + } + } + return SpecCluster(workers, scheduler, name="LocalEnvCluster", **kwargs) From a12dfa2f79b7162b842ea0f7237d106c1168c9c8 Mon Sep 17 00:00:00 2001 From: DPeterK Date: Wed, 16 Jun 2021 14:33:09 +0100 Subject: [PATCH 02/28] Different approach for getting scheduler/worker addresses --- distributed/__init__.py | 1 + distributed/deploy/__init__.py | 1 + distributed/deploy/local_env.py | 31 +++++++++++++++++++++---------- 3 files changed, 23 insertions(+), 10 deletions(-) diff --git a/distributed/__init__.py b/distributed/__init__.py index b624db8ad1c..2725aa96e56 100644 --- a/distributed/__init__.py +++ b/distributed/__init__.py @@ -36,6 +36,7 @@ from distributed.deploy import ( Adaptive, LocalCluster, + LocalEnvCluster, SpecCluster, SSHCluster, SubprocessCluster, diff --git a/distributed/deploy/__init__.py b/distributed/deploy/__init__.py index ffbeb6a8c64..3e3dbbd6208 100644 --- a/distributed/deploy/__init__.py +++ b/distributed/deploy/__init__.py @@ -3,6 +3,7 @@ from distributed.deploy.adaptive import Adaptive from distributed.deploy.cluster import Cluster from distributed.deploy.local import LocalCluster +from distributed.deploy.local_env import LocalEnvCluster from distributed.deploy.spec import ProcessInterface, SpecCluster from distributed.deploy.ssh import SSHCluster from distributed.deploy.subprocess import SubprocessCluster diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index 4c579ff3dd3..aff7dd815a7 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -1,6 +1,5 @@ import asyncio import logging -import sys import weakref from typing import List, Union @@ -104,12 +103,12 @@ async def start(self): # We watch stderr in order to get the address, then we return while True: - _, line = await self.proc.communicate() - if not line.decode().strip(): + line = await self.proc.stderr.readline() + if not line.decode('ascii').strip(): raise Exception("Worker failed to start") else: - line = line.decode() - logger.info(line.strip()) + line = line.decode('ascii').strip() + logger.info(line) if "worker at" in line: self.address = line.split("worker at:")[1].strip() self.status = Status.running @@ -161,12 +160,12 @@ async def start(self): # We watch stderr in order to get the address, then we return while True: - _, line = await self.proc.communicate() - if not line.decode().strip(): - raise Exception("Worker failed to start") + line = await self.proc.stderr.readline() + if not line.decode('ascii').strip(): + raise Exception("Scheduler failed to start") else: - line = line.decode() - logger.info(line.strip()) + line = line.decode('ascii').strip() + logger.info(line) if "Scheduler at" in line: self.address = line.split("Scheduler at:")[1].strip() break @@ -175,6 +174,18 @@ async def start(self): async def _set_env_helper(connect_options: dict): + """Helper function to locate existing dask internal config for the remote + scheduler and workers to inherit when started. + + Parameters + ---------- + connect_options : dict + Connection options to pass to the async subprocess shell. + + Returns + ------- + Dask config to inherit, if any. + """ proc = await asyncio.create_subprocess_shell("uname", **connect_options) await proc.communicate() if proc.returncode == 0: From c7a3ded359348f2c7ddc805b6570e9704645e6d2 Mon Sep 17 00:00:00 2001 From: DPeterK Date: Thu, 1 Jul 2021 12:29:48 +0100 Subject: [PATCH 03/28] Update cluster docstring --- distributed/deploy/local_env.py | 46 +++++++++++++-------------------- 1 file changed, 18 insertions(+), 28 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index aff7dd815a7..e76285fe653 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -218,11 +218,12 @@ def LocalEnvCluster( ): """Deploy a Dask cluster that utilises a different Python executable - The SSHCluster function deploys a Dask Scheduler and Workers for you on a - set of machine addresses that you provide. The first address will be used - for the scheduler while the rest will be used for the workers (feel free to - repeat the first hostname if you want to have the scheduler and worker - co-habitate one machine.) + The LocalEnvCluster function deploys a Dask Scheduler and Workers for you on + your local machine, running in the Python environment specified by + `python_exe`. This allows you to run a Dask cluster in a different Python + environment to the host Python environment; particularly useful when + combined with `dask-labextension` as you can run a Dask Scheduler and + Workers in a different Python env to the env hosting JupyterLab. You may configure the scheduler and workers by passing ``scheduler_options`` and ``worker_options`` dictionary keywords. See the @@ -230,10 +231,11 @@ def LocalEnvCluster( details on the available options, but the defaults should work in most situations. - You may configure your use of SSH itself using the ``connect_options`` - keyword, which passes values to the ``asyncssh.connect`` function. For - more information on these see the documentation for the ``asyncssh`` - library https://asyncssh.readthedocs.io . + You may configure how to connect to the local Scheduler and Workers using + the ``connect_options`` keyword, which passes values to the + ``asyncio.create_subprocess_shell`` function. For more information on this + see the documentation on the + [asyncio library](https://docs.python.org/3/library/asyncio-subprocess.html#asyncio-subprocess). Parameters ---------- @@ -253,33 +255,21 @@ def LocalEnvCluster( worker_module : str, optional Python module to call to start the worker. - Examples - -------- - >>> from dask.distributed import Client, SSHCluster - >>> cluster = SSHCluster( - ... ["localhost", "localhost", "localhost", "localhost"], + Example + ------- + >>> from dask.distributed import Client, LocalEnvCluster + >>> cluster = LocalEnvCluster( + ... "/Users/user/miniconda3/envs/myenv/bin/python", ... connect_options={"known_hosts": None}, - ... worker_options={"nthreads": 2}, - ... scheduler_options={"port": 0, "dashboard_address": ":8797"} + ... worker_options={"nthreads": 1}, ... ) >>> client = Client(cluster) - An example using a different worker module, in particular the - ``dask-cuda-worker`` command from the ``dask-cuda`` project. - - >>> from dask.distributed import Client, SSHCluster - >>> cluster = SSHCluster( - ... ["localhost", "hostwithgpus", "anothergpuhost"], - ... connect_options={"known_hosts": None}, - ... scheduler_options={"port": 0, "dashboard_address": ":8797"}, - ... worker_module='dask_cuda.dask_cuda_worker') - >>> client = Client(cluster) - See Also -------- dask.distributed.Scheduler dask.distributed.Worker - asyncssh.connect + asyncio.create_subprocess_shell """ scheduler = { "cls": Scheduler, From ab2f3601e17c2cc8a992bacc1a08cc1e2af03171 Mon Sep 17 00:00:00 2001 From: DPeterK Date: Thu, 1 Jul 2021 13:32:40 +0100 Subject: [PATCH 04/28] Utility libraries checks --- distributed/deploy/local_env.py | 33 ++++++++++----------------------- 1 file changed, 10 insertions(+), 23 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index e76285fe653..e8d5b50607c 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -29,9 +29,7 @@ def __init__(self, **kwargs): super().__init__(**kwargs) async def start(self): - weakref.finalize( - self, self.proc.kill - ) + weakref.finalize(self, self.proc.kill) await super().start() async def close(self): @@ -96,18 +94,16 @@ async def start(self): + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module) ) self.proc = await asyncio.create_subprocess_shell( - cmd, - stderr=asyncio.subprocess.PIPE, - **self.connect_options + cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options ) # We watch stderr in order to get the address, then we return while True: line = await self.proc.stderr.readline() - if not line.decode('ascii').strip(): + if not line.decode("ascii").strip(): raise Exception("Worker failed to start") else: - line = line.decode('ascii').strip() + line = line.decode("ascii").strip() logger.info(line) if "worker at" in line: self.address = line.split("worker at:")[1].strip() @@ -144,27 +140,20 @@ async def start(self): set_env = await _set_env_helper(self.connect_options) cmd = " ".join( - [ - set_env, - self.python_exe, - "-m", - "distributed.cli.dask_scheduler" - ] + [set_env, self.python_exe, "-m", "distributed.cli.dask_scheduler"] + cli_keywords(self.kwargs, cls=_Scheduler) ) self.proc = await asyncio.create_subprocess_shell( - cmd, - stderr=asyncio.subprocess.PIPE, - **self.connect_options + cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options ) # We watch stderr in order to get the address, then we return while True: line = await self.proc.stderr.readline() - if not line.decode('ascii').strip(): + if not line.decode("ascii").strip(): raise Exception("Scheduler failed to start") else: - line = line.decode('ascii').strip() + line = line.decode("ascii").strip() logger.info(line) if "Scheduler at" in line: self.address = line.split("Scheduler at:")[1].strip() @@ -193,10 +182,7 @@ async def _set_env_helper(connect_options: dict): dask.config.serialize(dask.config.global_config) ) else: - proc = await asyncio.create_subprocess_shell( - "cmd /c ver", - **connect_options - ) + proc = await asyncio.create_subprocess_shell("cmd /c ver", **connect_options) await proc.communicate() if proc.returncode == 0: set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format( @@ -208,6 +194,7 @@ async def _set_env_helper(connect_options: dict): ) return set_env + def LocalEnvCluster( python_exe: str, connect_options: Union[List[dict], dict] = {}, From b11e9158a1322ac154316bbd3c79e83bd24ec365 Mon Sep 17 00:00:00 2001 From: DPeterK Date: Thu, 1 Jul 2021 13:45:48 +0100 Subject: [PATCH 05/28] Update docstring usage example --- distributed/deploy/local_env.py | 1 - 1 file changed, 1 deletion(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index e8d5b50607c..d30f5f9ca5d 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -247,7 +247,6 @@ def LocalEnvCluster( >>> from dask.distributed import Client, LocalEnvCluster >>> cluster = LocalEnvCluster( ... "/Users/user/miniconda3/envs/myenv/bin/python", - ... connect_options={"known_hosts": None}, ... worker_options={"nthreads": 1}, ... ) >>> client = Client(cluster) From 08d1dc20f637c3c69b99ba52e814fe53aaa00a0a Mon Sep 17 00:00:00 2001 From: DPeterK Date: Mon, 12 Jul 2021 09:33:46 +0100 Subject: [PATCH 06/28] Change name of python_executable variable --- distributed/deploy/local_env.py | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index d30f5f9ca5d..ad303794d41 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -47,7 +47,7 @@ class Worker(Process): ---------- scheduler: str The address of the scheduler - python_exe: str + python_executable: str Full path to Python executable to run this worker connect_options: dict kwargs to be passed to asyncio subprocess connections @@ -63,7 +63,7 @@ class Worker(Process): def __init__( self, scheduler: str, - python_exe: str, + python_executable: str, connect_options: dict, kwargs: dict, worker_module="distributed.cli.dask_worker", @@ -72,7 +72,7 @@ def __init__( super().__init__() self.scheduler = scheduler - self.python_exe = python_exe + self.python_executable = python_executable self.connect_options = connect_options self.kwargs = kwargs self.worker_module = worker_module @@ -84,7 +84,7 @@ async def start(self): cmd = " ".join( [ set_env, - self.python_exe, + self.python_executable, "-m", self.worker_module, self.scheduler, @@ -118,7 +118,7 @@ class Scheduler(Process): Parameters ---------- - python_exe: str + python_executable: str Full path to Python executable to run this scheduler connect_options: dict kwargs to be passed to asyncio subprocess connections @@ -127,10 +127,10 @@ class Scheduler(Process): dask.distributed.Scheduler class """ - def __init__(self, python_exe: str, connect_options: dict, kwargs: dict): + def __init__(self, python_executable: str, connect_options: dict, kwargs: dict): super().__init__() - self.python_exe = python_exe + self.python_executable = python_executable self.kwargs = kwargs self.connect_options = connect_options @@ -140,7 +140,7 @@ async def start(self): set_env = await _set_env_helper(self.connect_options) cmd = " ".join( - [set_env, self.python_exe, "-m", "distributed.cli.dask_scheduler"] + [set_env, self.python_executable, "-m", "distributed.cli.dask_scheduler"] + cli_keywords(self.kwargs, cls=_Scheduler) ) self.proc = await asyncio.create_subprocess_shell( @@ -196,7 +196,7 @@ async def _set_env_helper(connect_options: dict): def LocalEnvCluster( - python_exe: str, + python_executable: str, connect_options: Union[List[dict], dict] = {}, worker_options: dict = {}, scheduler_options: dict = {}, @@ -207,8 +207,8 @@ def LocalEnvCluster( The LocalEnvCluster function deploys a Dask Scheduler and Workers for you on your local machine, running in the Python environment specified by - `python_exe`. This allows you to run a Dask cluster in a different Python - environment to the host Python environment; particularly useful when + `python_executable`. This allows you to run a Dask cluster in a different + Python environment to the host Python environment; particularly useful when combined with `dask-labextension` as you can run a Dask Scheduler and Workers in a different Python env to the env hosting JupyterLab. @@ -226,8 +226,8 @@ def LocalEnvCluster( Parameters ---------- - python_exe : str - Full path to another Python executable, for example from another + python_executable : str + Full path to a Python executable, for example from another conda env or Python venv. connect_options : dict or list of dict, optional Keywords to pass through to :func:`asyncssh.connect`. @@ -260,7 +260,7 @@ def LocalEnvCluster( scheduler = { "cls": Scheduler, "options": { - "python_exe": python_exe, + "python_executable": python_executable, "connect_options": connect_options, "kwargs": scheduler_options, }, @@ -269,7 +269,7 @@ def LocalEnvCluster( 0: { "cls": Worker, "options": { - "python_exe": python_exe, + "python_executable": python_executable, "connect_options": connect_options, "kwargs": worker_options, "worker_module": worker_module, From b36e0abb57b4316575ca2a3ddbafd793c8b6dc7e Mon Sep 17 00:00:00 2001 From: DPeterK Date: Mon, 12 Jul 2021 13:15:13 +0100 Subject: [PATCH 07/28] Refactor common code into Process superclass --- distributed/deploy/local_env.py | 118 +++++++++++++++----------------- 1 file changed, 54 insertions(+), 64 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index ad303794d41..43a051e0b2d 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -1,6 +1,5 @@ import asyncio import logging -import weakref from typing import List, Union import dask @@ -16,7 +15,7 @@ class Process(ProcessInterface): - """A superclass for SSH Workers and Nannies + """A superclass for Workers and Nannies run by a specified Python executable See Also -------- @@ -29,7 +28,6 @@ def __init__(self, **kwargs): super().__init__(**kwargs) async def start(self): - weakref.finalize(self, self.proc.kill) await super().start() async def close(self): @@ -37,7 +35,53 @@ async def close(self): await super().close() def __repr__(self): - return "" % (type(self).__name__, self.status) + return "" % (type(self).__name__, self.status) + + async def _set_env_helper(self): + """Helper function to locate existing dask internal config for the remote + Scheduler and Workers to inherit when started. + + Returns + ------- + Dask config to inherit, if any. + """ + proc = await asyncio.create_subprocess_shell("uname", **self.connect_options) + await proc.communicate() + if proc.returncode == 0: + set_env = 'env DASK_INTERNAL_INHERIT_CONFIG="{}"'.format( + dask.config.serialize(dask.config.global_config) + ) + else: + proc = await asyncio.create_subprocess_shell( + "cmd /c ver", **self.connect_options + ) + await proc.communicate() + if proc.returncode == 0: + set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format( + dask.config.serialize(dask.config.global_config) + ) + else: + name = self.__class__.__name__ + emsg = f"{name} failed to set DASK_INTERNAL_INHERIT_CONFIG variable" + raise Exception(emsg) + return set_env + + async def _get_address(self, search_str): + # We watch stderr in order to get the address, then we return + name = self.__class__.__name__ + while True: + line = await self.proc.stderr.readline() + if not line.decode("ascii").strip(): + raise Exception(f"{name} failed to start") + else: + line = line.decode("ascii").strip() + logger.info(line) + if search_str in line: + self.address = line.split(f"{search_str}:")[1].strip() + if name == "Worker": + self.status = Status.running + break + logger.debug("%s", line) class Worker(Process): @@ -79,7 +123,7 @@ def __init__( self.name = name async def start(self): - set_env = await _set_env_helper(self.connect_options) + set_env = await self._set_env_helper() cmd = " ".join( [ @@ -97,19 +141,8 @@ async def start(self): cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options ) - # We watch stderr in order to get the address, then we return - while True: - line = await self.proc.stderr.readline() - if not line.decode("ascii").strip(): - raise Exception("Worker failed to start") - else: - line = line.decode("ascii").strip() - logger.info(line) - if "worker at" in line: - self.address = line.split("worker at:")[1].strip() - self.status = Status.running - break - logger.debug("%s", line) + search_string = "worker at" + await self._get_address(search_string) await super().start() @@ -137,7 +170,7 @@ def __init__(self, python_executable: str, connect_options: dict, kwargs: dict): async def start(self): logger.debug("Created Scheduler") - set_env = await _set_env_helper(self.connect_options) + set_env = await self._set_env_helper() cmd = " ".join( [set_env, self.python_executable, "-m", "distributed.cli.dask_scheduler"] @@ -147,54 +180,11 @@ async def start(self): cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options ) - # We watch stderr in order to get the address, then we return - while True: - line = await self.proc.stderr.readline() - if not line.decode("ascii").strip(): - raise Exception("Scheduler failed to start") - else: - line = line.decode("ascii").strip() - logger.info(line) - if "Scheduler at" in line: - self.address = line.split("Scheduler at:")[1].strip() - break - logger.debug("%s", line) + search_string = "Scheduler at" + await self._get_address(search_string) await super().start() -async def _set_env_helper(connect_options: dict): - """Helper function to locate existing dask internal config for the remote - scheduler and workers to inherit when started. - - Parameters - ---------- - connect_options : dict - Connection options to pass to the async subprocess shell. - - Returns - ------- - Dask config to inherit, if any. - """ - proc = await asyncio.create_subprocess_shell("uname", **connect_options) - await proc.communicate() - if proc.returncode == 0: - set_env = 'env DASK_INTERNAL_INHERIT_CONFIG="{}"'.format( - dask.config.serialize(dask.config.global_config) - ) - else: - proc = await asyncio.create_subprocess_shell("cmd /c ver", **connect_options) - await proc.communicate() - if proc.returncode == 0: - set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format( - dask.config.serialize(dask.config.global_config) - ) - else: - raise Exception( - "Scheduler failed to set DASK_INTERNAL_INHERIT_CONFIG variable " - ) - return set_env - - def LocalEnvCluster( python_executable: str, connect_options: Union[List[dict], dict] = {}, From 114038f5a17ca3dba0a44ebafac884c91cb32cbe Mon Sep 17 00:00:00 2001 From: DPeterK Date: Mon, 19 Jul 2021 16:55:32 +0100 Subject: [PATCH 08/28] Add n_workers option to local env cluster --- distributed/deploy/local_env.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index 43a051e0b2d..680ba6f1a18 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -187,6 +187,7 @@ async def start(self): def LocalEnvCluster( python_executable: str, + n_workers: int = 1, connect_options: Union[List[dict], dict] = {}, worker_options: dict = {}, scheduler_options: dict = {}, @@ -219,6 +220,8 @@ def LocalEnvCluster( python_executable : str Full path to a Python executable, for example from another conda env or Python venv. + n_workers : int + Number of workers to start connect_options : dict or list of dict, optional Keywords to pass through to :func:`asyncssh.connect`. This could include things such as ``port``, ``username``, ``password`` @@ -256,7 +259,7 @@ def LocalEnvCluster( }, } workers = { - 0: { + i: { "cls": Worker, "options": { "python_executable": python_executable, @@ -265,5 +268,6 @@ def LocalEnvCluster( "worker_module": worker_module, }, } + for i in range(n_workers) } return SpecCluster(workers, scheduler, name="LocalEnvCluster", **kwargs) From b44ae0a2a43189f3431278c027690bee125df129 Mon Sep 17 00:00:00 2001 From: DPeterK Date: Mon, 19 Jul 2021 16:57:52 +0100 Subject: [PATCH 09/28] Correct docstring for connect_options --- distributed/deploy/local_env.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index 680ba6f1a18..add67d295e0 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -222,12 +222,9 @@ def LocalEnvCluster( conda env or Python venv. n_workers : int Number of workers to start - connect_options : dict or list of dict, optional - Keywords to pass through to :func:`asyncssh.connect`. - This could include things such as ``port``, ``username``, ``password`` - or ``known_hosts``. See docs for :func:`asyncssh.connect` and - :class:`asyncssh.SSHClientConnectionOptions` for full information. - If a list it must have the same length as ``hosts``. + connect_options : dict, optional + Keywords to pass through to :func:`asyncio.create_subprocess_shell`. + See docs for :func:`asyncio.create_subprocess_shell` for full information. worker_options : dict, optional Keywords to pass on to workers. scheduler_options : dict, optional From e82e5eb942e2c9c9ac724e11483a063efdb46ca7 Mon Sep 17 00:00:00 2001 From: DPeterK Date: Mon, 2 Aug 2021 13:35:52 +0100 Subject: [PATCH 10/28] Add tests --- distributed/deploy/tests/test_local_env.py | 77 ++++++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 distributed/deploy/tests/test_local_env.py diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py new file mode 100644 index 00000000000..370452fbc01 --- /dev/null +++ b/distributed/deploy/tests/test_local_env.py @@ -0,0 +1,77 @@ +import pytest + +import sys + +import dask + +from distributed import Client +from distributed.core import Status +from distributed.deploy.local_env import LocalEnvCluster + + +@pytest.mark.asyncio +async def test_basic(): + async with LocalEnvCluster( + sys.executable, + scheduler_options={"port": 0, "idle_timeout": "5s"}, + worker_options={"death_timeout": "5s"}, + ) as cluster: + assert len(cluster.workers) == 1 + assert cluster.scheduler.python_executable == sys.executable + assert cluster.workers[0].python_executable == sys.executable + assert "LocalEnv" in repr(cluster) + assert cluster.status == Status.closed + + +@pytest.mark.asyncio +async def test_job_submission(): + async with LocalEnvCluster( + sys.executable, + scheduler_options={"port": 0, "idle_timeout": "5s"}, + worker_options={"death_timeout": "5s"}, + ) as cluster: + async with Client(cluster) as client: + result = await client.submit(lambda x: x + 1, 10) + assert result == 11 + + +@pytest.mark.asyncio +async def test_multiple_workers(): + n_workers = 2 + async with LocalEnvCluster( + sys.executable, + n_workers=n_workers, + scheduler_options={"port": 0, "idle_timeout": "5s"}, + worker_options={"death_timeout": "5s"}, + ) as cluster: + assert len(cluster.workers) == n_workers + + +@pytest.mark.asyncio +async def test_bad_executable(): + with pytest.raises(Exception): + async with LocalEnvCluster( + "/foo/bar/baz/python", + scheduler_options={"port": 0, "idle_timeout": "5s"}, + worker_options={"death_timeout": "5s"}, + ) as cluster: + assert cluster + cluster.close() + + +@pytest.mark.asyncio +async def test_set_env(): + value = 100 + + def f(): + return dask.config.get("foo") + + with dask.config.set(foo=value): + async with LocalEnvCluster( + sys.executable, + scheduler_options={"port": 0, "idle_timeout": "5s"}, + worker_options={"death_timeout": "5s"}, + ) as cluster: + async with Client(cluster) as client: + result = await client.submit(f) + assert result == value From 08adbee2e346b4a8c92060effa144e67eae0dcc3 Mon Sep 17 00:00:00 2001 From: DPeterK Date: Wed, 4 Aug 2021 15:28:20 +0100 Subject: [PATCH 11/28] Add asynchronous flag to tests --- distributed/deploy/tests/test_local_env.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index 370452fbc01..61b12466359 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -13,6 +13,7 @@ async def test_basic(): async with LocalEnvCluster( sys.executable, + asynchronous=True, scheduler_options={"port": 0, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: @@ -27,6 +28,7 @@ async def test_basic(): async def test_job_submission(): async with LocalEnvCluster( sys.executable, + asynchronous=True, scheduler_options={"port": 0, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: @@ -41,6 +43,7 @@ async def test_multiple_workers(): async with LocalEnvCluster( sys.executable, n_workers=n_workers, + asynchronous=True, scheduler_options={"port": 0, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: @@ -52,6 +55,7 @@ async def test_bad_executable(): with pytest.raises(Exception): async with LocalEnvCluster( "/foo/bar/baz/python", + asynchronous=True, scheduler_options={"port": 0, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: @@ -69,6 +73,7 @@ def f(): with dask.config.set(foo=value): async with LocalEnvCluster( sys.executable, + asynchronous=True, scheduler_options={"port": 0, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: From 4b045fe428482da7aed2b6e26c48f4600e1e850a Mon Sep 17 00:00:00 2001 From: DPeterK Date: Thu, 5 Aug 2021 10:58:58 +0100 Subject: [PATCH 12/28] Add discrete scheduler port for each test --- distributed/deploy/tests/test_local_env.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index 61b12466359..fa961b736e1 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -14,7 +14,7 @@ async def test_basic(): async with LocalEnvCluster( sys.executable, asynchronous=True, - scheduler_options={"port": 0, "idle_timeout": "5s"}, + scheduler_options={"port": 20000, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: assert len(cluster.workers) == 1 @@ -29,7 +29,7 @@ async def test_job_submission(): async with LocalEnvCluster( sys.executable, asynchronous=True, - scheduler_options={"port": 0, "idle_timeout": "5s"}, + scheduler_options={"port": 20001, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: async with Client(cluster) as client: @@ -44,7 +44,7 @@ async def test_multiple_workers(): sys.executable, n_workers=n_workers, asynchronous=True, - scheduler_options={"port": 0, "idle_timeout": "5s"}, + scheduler_options={"port": 20002, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: assert len(cluster.workers) == n_workers @@ -56,7 +56,7 @@ async def test_bad_executable(): async with LocalEnvCluster( "/foo/bar/baz/python", asynchronous=True, - scheduler_options={"port": 0, "idle_timeout": "5s"}, + scheduler_options={"port": 20003, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: assert cluster @@ -74,7 +74,7 @@ def f(): async with LocalEnvCluster( sys.executable, asynchronous=True, - scheduler_options={"port": 0, "idle_timeout": "5s"}, + scheduler_options={"port": 20004, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: async with Client(cluster) as client: From 5aeb0480189ec51a98d1d140cd3c317d82e991ec Mon Sep 17 00:00:00 2001 From: DPeterK Date: Thu, 5 Aug 2021 10:59:50 +0100 Subject: [PATCH 13/28] WIP: try using exec and not shell subprocesses --- distributed/deploy/local_env.py | 97 ++++++++++++++++++++++++--------- 1 file changed, 71 insertions(+), 26 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index add67d295e0..f638a6518f0 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -45,21 +45,24 @@ async def _set_env_helper(self): ------- Dask config to inherit, if any. """ + keyname = "DASK_INTERNAL_INHERIT_CONFIG" proc = await asyncio.create_subprocess_shell("uname", **self.connect_options) await proc.communicate() if proc.returncode == 0: - set_env = 'env DASK_INTERNAL_INHERIT_CONFIG="{}"'.format( - dask.config.serialize(dask.config.global_config) - ) + # set_env = 'env DASK_INTERNAL_INHERIT_CONFIG="{}"'.format( + # dask.config.serialize(dask.config.global_config) + # ) + set_env = {keyname: f"{dask.config.serialize(dask.config.global_config)}"} else: proc = await asyncio.create_subprocess_shell( "cmd /c ver", **self.connect_options ) await proc.communicate() if proc.returncode == 0: - set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format( - dask.config.serialize(dask.config.global_config) - ) + # set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format( + # dask.config.serialize(dask.config.global_config) + # ) + set_env = {keyname: f"{dask.config.serialize(dask.config.global_config)}"} else: name = self.__class__.__name__ emsg = f"{name} failed to set DASK_INTERNAL_INHERIT_CONFIG variable" @@ -125,26 +128,49 @@ def __init__( async def start(self): set_env = await self._set_env_helper() - cmd = " ".join( - [ - set_env, - self.python_executable, - "-m", - self.worker_module, - self.scheduler, - "--name", - str(self.name), - ] - + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module) - ) - self.proc = await asyncio.create_subprocess_shell( - cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options + cmds = [ + "-m", + self.worker_module, + self.scheduler, + "--name", + str(self.name), + ] + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module) + + self.proc = await asyncio.create_subprocess_exec( + self.python_executable, + *cmds, + stderr=asyncio.subprocess.PIPE, + env=set_env, + **self.connect_options ) search_string = "worker at" await self._get_address(search_string) await super().start() + # async def _start(self): + # set_env = await self._set_env_helper() + # + # cmd = " ".join( + # [ + # set_env, + # self.python_executable, + # "-m", + # self.worker_module, + # self.scheduler, + # "--name", + # str(self.name), + # ] + # + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module) + # ) + # self.proc = await asyncio.create_subprocess_shell( + # cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options + # ) + # + # search_string = "worker at" + # await self._get_address(search_string) + # await super().start() + class Scheduler(Process): """A Remote Dask Scheduler run by a specified Python executable @@ -172,18 +198,37 @@ async def start(self): set_env = await self._set_env_helper() - cmd = " ".join( - [set_env, self.python_executable, "-m", "distributed.cli.dask_scheduler"] - + cli_keywords(self.kwargs, cls=_Scheduler) - ) - self.proc = await asyncio.create_subprocess_shell( - cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options + cmds = ["-m", "distributed.cli.dask_scheduler"] + cli_keywords(self.kwargs, cls=_Scheduler) + + self.proc = await asyncio.create_subprocess_exec( + self.python_executable, + *cmds, + stderr=asyncio.subprocess.PIPE, + env=set_env, + **self.connect_options ) search_string = "Scheduler at" await self._get_address(search_string) await super().start() + # async def _start(self): + # logger.debug("Created Scheduler") + # + # set_env = await self._set_env_helper() + # + # cmd = " ".join( + # [set_env, self.python_executable, "-m", "distributed.cli.dask_scheduler"] + # + cli_keywords(self.kwargs, cls=_Scheduler) + # ) + # self.proc = await asyncio.create_subprocess_shell( + # cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options + # ) + # + # search_string = "Scheduler at" + # await self._get_address(search_string) + # await super().start() + def LocalEnvCluster( python_executable: str, From e4c6a8f1900057d1557f8d3d2d49e1c079586d6c Mon Sep 17 00:00:00 2001 From: DPeterK Date: Thu, 5 Aug 2021 13:48:12 +0100 Subject: [PATCH 14/28] Tests pass --- distributed/deploy/local_env.py | 97 ++++++---------------- distributed/deploy/tests/test_local_env.py | 26 +++--- 2 files changed, 39 insertions(+), 84 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index f638a6518f0..add67d295e0 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -45,24 +45,21 @@ async def _set_env_helper(self): ------- Dask config to inherit, if any. """ - keyname = "DASK_INTERNAL_INHERIT_CONFIG" proc = await asyncio.create_subprocess_shell("uname", **self.connect_options) await proc.communicate() if proc.returncode == 0: - # set_env = 'env DASK_INTERNAL_INHERIT_CONFIG="{}"'.format( - # dask.config.serialize(dask.config.global_config) - # ) - set_env = {keyname: f"{dask.config.serialize(dask.config.global_config)}"} + set_env = 'env DASK_INTERNAL_INHERIT_CONFIG="{}"'.format( + dask.config.serialize(dask.config.global_config) + ) else: proc = await asyncio.create_subprocess_shell( "cmd /c ver", **self.connect_options ) await proc.communicate() if proc.returncode == 0: - # set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format( - # dask.config.serialize(dask.config.global_config) - # ) - set_env = {keyname: f"{dask.config.serialize(dask.config.global_config)}"} + set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format( + dask.config.serialize(dask.config.global_config) + ) else: name = self.__class__.__name__ emsg = f"{name} failed to set DASK_INTERNAL_INHERIT_CONFIG variable" @@ -128,49 +125,26 @@ def __init__( async def start(self): set_env = await self._set_env_helper() - cmds = [ - "-m", - self.worker_module, - self.scheduler, - "--name", - str(self.name), - ] + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module) - - self.proc = await asyncio.create_subprocess_exec( - self.python_executable, - *cmds, - stderr=asyncio.subprocess.PIPE, - env=set_env, - **self.connect_options + cmd = " ".join( + [ + set_env, + self.python_executable, + "-m", + self.worker_module, + self.scheduler, + "--name", + str(self.name), + ] + + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module) + ) + self.proc = await asyncio.create_subprocess_shell( + cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options ) search_string = "worker at" await self._get_address(search_string) await super().start() - # async def _start(self): - # set_env = await self._set_env_helper() - # - # cmd = " ".join( - # [ - # set_env, - # self.python_executable, - # "-m", - # self.worker_module, - # self.scheduler, - # "--name", - # str(self.name), - # ] - # + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module) - # ) - # self.proc = await asyncio.create_subprocess_shell( - # cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options - # ) - # - # search_string = "worker at" - # await self._get_address(search_string) - # await super().start() - class Scheduler(Process): """A Remote Dask Scheduler run by a specified Python executable @@ -198,37 +172,18 @@ async def start(self): set_env = await self._set_env_helper() - cmds = ["-m", "distributed.cli.dask_scheduler"] + cli_keywords(self.kwargs, cls=_Scheduler) - - self.proc = await asyncio.create_subprocess_exec( - self.python_executable, - *cmds, - stderr=asyncio.subprocess.PIPE, - env=set_env, - **self.connect_options + cmd = " ".join( + [set_env, self.python_executable, "-m", "distributed.cli.dask_scheduler"] + + cli_keywords(self.kwargs, cls=_Scheduler) + ) + self.proc = await asyncio.create_subprocess_shell( + cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options ) search_string = "Scheduler at" await self._get_address(search_string) await super().start() - # async def _start(self): - # logger.debug("Created Scheduler") - # - # set_env = await self._set_env_helper() - # - # cmd = " ".join( - # [set_env, self.python_executable, "-m", "distributed.cli.dask_scheduler"] - # + cli_keywords(self.kwargs, cls=_Scheduler) - # ) - # self.proc = await asyncio.create_subprocess_shell( - # cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options - # ) - # - # search_string = "Scheduler at" - # await self._get_address(search_string) - # await super().start() - def LocalEnvCluster( python_executable: str, diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index fa961b736e1..e53502f9616 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -1,7 +1,7 @@ -import pytest - import sys +import pytest + import dask from distributed import Client @@ -32,7 +32,7 @@ async def test_job_submission(): scheduler_options={"port": 20001, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: - async with Client(cluster) as client: + async with Client(cluster, asynchronous=True) as client: result = await client.submit(lambda x: x + 1, 10) assert result == 11 @@ -41,11 +41,11 @@ async def test_job_submission(): async def test_multiple_workers(): n_workers = 2 async with LocalEnvCluster( - sys.executable, - n_workers=n_workers, - asynchronous=True, - scheduler_options={"port": 20002, "idle_timeout": "5s"}, - worker_options={"death_timeout": "5s"}, + sys.executable, + n_workers=n_workers, + asynchronous=True, + scheduler_options={"port": 20002, "idle_timeout": "5s"}, + worker_options={"death_timeout": "5s"}, ) as cluster: assert len(cluster.workers) == n_workers @@ -54,10 +54,10 @@ async def test_multiple_workers(): async def test_bad_executable(): with pytest.raises(Exception): async with LocalEnvCluster( - "/foo/bar/baz/python", - asynchronous=True, - scheduler_options={"port": 20003, "idle_timeout": "5s"}, - worker_options={"death_timeout": "5s"}, + "/foo/bar/baz/python", + asynchronous=True, + scheduler_options={"port": 20003, "idle_timeout": "5s"}, + worker_options={"death_timeout": "5s"}, ) as cluster: assert cluster cluster.close() @@ -77,6 +77,6 @@ def f(): scheduler_options={"port": 20004, "idle_timeout": "5s"}, worker_options={"death_timeout": "5s"}, ) as cluster: - async with Client(cluster) as client: + async with Client(cluster, asynchronous=True) as client: result = await client.submit(f) assert result == value From 4c1ce6daefa4abe0e78fa712f239e0d5c06f0a28 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Mon, 20 Jul 2026 16:45:42 +0200 Subject: [PATCH 15/28] Replace outdated asyncio mark with gen_test() --- distributed/deploy/tests/test_local_env.py | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index e53502f9616..d11c00bd7be 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -1,4 +1,5 @@ import sys +import asyncio import pytest @@ -7,9 +8,9 @@ from distributed import Client from distributed.core import Status from distributed.deploy.local_env import LocalEnvCluster +from distributed.utils_test import gen_test - -@pytest.mark.asyncio +@gen_test() async def test_basic(): async with LocalEnvCluster( sys.executable, @@ -23,8 +24,7 @@ async def test_basic(): assert "LocalEnv" in repr(cluster) assert cluster.status == Status.closed - -@pytest.mark.asyncio +@gen_test() async def test_job_submission(): async with LocalEnvCluster( sys.executable, @@ -36,8 +36,7 @@ async def test_job_submission(): result = await client.submit(lambda x: x + 1, 10) assert result == 11 - -@pytest.mark.asyncio +@gen_test() async def test_multiple_workers(): n_workers = 2 async with LocalEnvCluster( @@ -49,8 +48,7 @@ async def test_multiple_workers(): ) as cluster: assert len(cluster.workers) == n_workers - -@pytest.mark.asyncio +@gen_test() async def test_bad_executable(): with pytest.raises(Exception): async with LocalEnvCluster( @@ -62,8 +60,7 @@ async def test_bad_executable(): assert cluster cluster.close() - -@pytest.mark.asyncio +@gen_test() async def test_set_env(): value = 100 From 56acdb13fa400657478637f97ae194cc5ed5cf36 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Mon, 20 Jul 2026 17:23:16 +0200 Subject: [PATCH 16/28] Use random scheduler and dashboard ports --- distributed/deploy/tests/test_local_env.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index d11c00bd7be..4eb0a420edd 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -15,7 +15,7 @@ async def test_basic(): async with LocalEnvCluster( sys.executable, asynchronous=True, - scheduler_options={"port": 20000, "idle_timeout": "5s"}, + scheduler_options={"idle_timeout": "5s", "port": 0, "dashboard_address": ":0"}, worker_options={"death_timeout": "5s"}, ) as cluster: assert len(cluster.workers) == 1 @@ -29,7 +29,7 @@ async def test_job_submission(): async with LocalEnvCluster( sys.executable, asynchronous=True, - scheduler_options={"port": 20001, "idle_timeout": "5s"}, + scheduler_options={"idle_timeout": "5s", "port": 0, "dashboard_address": ":0"}, worker_options={"death_timeout": "5s"}, ) as cluster: async with Client(cluster, asynchronous=True) as client: @@ -43,7 +43,7 @@ async def test_multiple_workers(): sys.executable, n_workers=n_workers, asynchronous=True, - scheduler_options={"port": 20002, "idle_timeout": "5s"}, + scheduler_options={"idle_timeout": "5s", "port": 0, "dashboard_address": ":0"}, worker_options={"death_timeout": "5s"}, ) as cluster: assert len(cluster.workers) == n_workers @@ -54,7 +54,7 @@ async def test_bad_executable(): async with LocalEnvCluster( "/foo/bar/baz/python", asynchronous=True, - scheduler_options={"port": 20003, "idle_timeout": "5s"}, + scheduler_options={"idle_timeout": "5s", "port": 0, "dashboard_address": ":0"}, worker_options={"death_timeout": "5s"}, ) as cluster: assert cluster @@ -71,7 +71,7 @@ def f(): async with LocalEnvCluster( sys.executable, asynchronous=True, - scheduler_options={"port": 20004, "idle_timeout": "5s"}, + scheduler_options={"idle_timeout": "5s", "port": 0, "dashboard_address": ":0"}, worker_options={"death_timeout": "5s"}, ) as cluster: async with Client(cluster, asynchronous=True) as client: From 211d694ace8ad50ec4416bc2f0a60c721f47023b Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Mon, 20 Jul 2026 17:36:00 +0200 Subject: [PATCH 17/28] Lint --- distributed/deploy/local_env.py | 32 ++++++++++++---------- distributed/deploy/tests/test_local_env.py | 2 +- 2 files changed, 18 insertions(+), 16 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index add67d295e0..4fe389fe855 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -1,14 +1,14 @@ import asyncio import logging -from typing import List, Union import dask import dask.config -from ..core import Status -from ..scheduler import Scheduler as _Scheduler -from ..utils import cli_keywords -from ..worker import Worker as _Worker +from distributed.core import Status +from distributed.scheduler import Scheduler as _Scheduler +from distributed.utils import cli_keywords +from distributed.worker import Worker as _Worker + from .spec import ProcessInterface, SpecCluster logger = logging.getLogger(__name__) @@ -35,7 +35,7 @@ async def close(self): await super().close() def __repr__(self): - return "" % (type(self).__name__, self.status) + return f"" async def _set_env_helper(self): """Helper function to locate existing dask internal config for the remote @@ -48,18 +48,14 @@ async def _set_env_helper(self): proc = await asyncio.create_subprocess_shell("uname", **self.connect_options) await proc.communicate() if proc.returncode == 0: - set_env = 'env DASK_INTERNAL_INHERIT_CONFIG="{}"'.format( - dask.config.serialize(dask.config.global_config) - ) + set_env = f'env DASK_INTERNAL_INHERIT_CONFIG="{dask.config.serialize(dask.config.global_config)}"' else: proc = await asyncio.create_subprocess_shell( "cmd /c ver", **self.connect_options ) await proc.communicate() if proc.returncode == 0: - set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format( - dask.config.serialize(dask.config.global_config) - ) + set_env = f"set DASK_INTERNAL_INHERIT_CONFIG={dask.config.serialize(dask.config.global_config)} &&" else: name = self.__class__.__name__ emsg = f"{name} failed to set DASK_INTERNAL_INHERIT_CONFIG variable" @@ -188,9 +184,9 @@ async def start(self): def LocalEnvCluster( python_executable: str, n_workers: int = 1, - connect_options: Union[List[dict], dict] = {}, - worker_options: dict = {}, - scheduler_options: dict = {}, + connect_options: list[dict] | dict | None = None, + worker_options: dict | None = None, + scheduler_options: dict | None = None, worker_module: str = "distributed.cli.dask_worker", **kwargs, ): @@ -247,6 +243,12 @@ def LocalEnvCluster( dask.distributed.Worker asyncio.create_subprocess_shell """ + if scheduler_options is None: + scheduler_options = {} + if worker_options is None: + worker_options = {} + if connect_options is None: + connect_options = {} scheduler = { "cls": Scheduler, "options": { diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index 4eb0a420edd..9b1511e8421 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -1,5 +1,4 @@ import sys -import asyncio import pytest @@ -10,6 +9,7 @@ from distributed.deploy.local_env import LocalEnvCluster from distributed.utils_test import gen_test + @gen_test() async def test_basic(): async with LocalEnvCluster( From 62f301f808a23c7e40067b7b8cc469d9ef16cc17 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Mon, 20 Jul 2026 17:42:17 +0200 Subject: [PATCH 18/28] Fix test expectation --- distributed/deploy/local_env.py | 2 +- distributed/deploy/tests/test_local_env.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index 4fe389fe855..4b98bba6c3a 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -68,7 +68,7 @@ async def _get_address(self, search_str): while True: line = await self.proc.stderr.readline() if not line.decode("ascii").strip(): - raise Exception(f"{name} failed to start") + raise RuntimeError(f"{name} failed to start") else: line = line.decode("ascii").strip() logger.info(line) diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index 9b1511e8421..e9341e6777a 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -50,7 +50,7 @@ async def test_multiple_workers(): @gen_test() async def test_bad_executable(): - with pytest.raises(Exception): + with pytest.raises(RuntimeError): async with LocalEnvCluster( "/foo/bar/baz/python", asynchronous=True, From 7946d7649f0d2fe8964f04ad13751e26e1c6a0c9 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Mon, 20 Jul 2026 17:45:00 +0200 Subject: [PATCH 19/28] More linting --- distributed/deploy/tests/test_local_env.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index e9341e6777a..7b38abfbfb7 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -24,6 +24,7 @@ async def test_basic(): assert "LocalEnv" in repr(cluster) assert cluster.status == Status.closed + @gen_test() async def test_job_submission(): async with LocalEnvCluster( @@ -36,6 +37,7 @@ async def test_job_submission(): result = await client.submit(lambda x: x + 1, 10) assert result == 11 + @gen_test() async def test_multiple_workers(): n_workers = 2 @@ -48,6 +50,7 @@ async def test_multiple_workers(): ) as cluster: assert len(cluster.workers) == n_workers + @gen_test() async def test_bad_executable(): with pytest.raises(RuntimeError): @@ -60,6 +63,7 @@ async def test_bad_executable(): assert cluster cluster.close() + @gen_test() async def test_set_env(): value = 100 From c9af892796570d5ab53edefb1842ef9a4aa1338e Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Mon, 20 Jul 2026 17:53:22 +0200 Subject: [PATCH 20/28] Attempt to fix LocalEnvCluster kwargs --- distributed/deploy/local_env.py | 21 ++++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index 4b98bba6c3a..123a592697b 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -1,5 +1,6 @@ import asyncio import logging +from typing import Any as AnyType import dask import dask.config @@ -31,7 +32,10 @@ async def start(self): await super().start() async def close(self): - self.proc.kill() + try: + self.proc.kill() + except ProcessLookupError: + pass await super().close() def __repr__(self): @@ -104,17 +108,17 @@ def __init__( self, scheduler: str, python_executable: str, - connect_options: dict, - kwargs: dict, - worker_module="distributed.cli.dask_worker", + connect_options: dict[str, AnyType] | None = None, + worker_module: str = "distributed.cli.dask_worker", name=None, + kwargs: dict[str, AnyType] | None = None, ): super().__init__() self.scheduler = scheduler self.python_executable = python_executable self.connect_options = connect_options - self.kwargs = kwargs + self.kwargs = kwargs or {} self.worker_module = worker_module self.name = name @@ -133,6 +137,7 @@ async def start(self): ] + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module) ) + self.proc = await asyncio.create_subprocess_shell( cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options ) @@ -156,11 +161,12 @@ class Scheduler(Process): dask.distributed.Scheduler class """ - def __init__(self, python_executable: str, connect_options: dict, kwargs: dict): + def __init__(self, python_executable: str, connect_options: dict, kwargs: dict[str, AnyType] | None = None, +): super().__init__() self.python_executable = python_executable - self.kwargs = kwargs + self.kwargs = kwargs or {} self.connect_options = connect_options async def start(self): @@ -172,6 +178,7 @@ async def start(self): [set_env, self.python_executable, "-m", "distributed.cli.dask_scheduler"] + cli_keywords(self.kwargs, cls=_Scheduler) ) + self.proc = await asyncio.create_subprocess_shell( cmd, stderr=asyncio.subprocess.PIPE, **self.connect_options ) From f381dbac80ea77c5852cdbe69443a7b11c9ad72c Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 21 Jul 2026 08:59:00 +0200 Subject: [PATCH 21/28] Ignore asyncio warnings about unclosed transports --- pyproject.toml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 8876a833ceb..ac535599abf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -138,9 +138,12 @@ filterwarnings = [ '''ignore:unclosed event loop <_(Unix|Windows)SelectorEventLoop.*:ResourceWarning''', '''ignore:unclosed file <_io.BufferedWriter.*:ResourceWarning''', '''ignore:unclosed file <_io.TextIOWrapper.*:ResourceWarning''', + '''ignore:unclosed file <_io.FileIO.*:ResourceWarning''', '''ignore:unclosed transport <_SelectorSocketTransport.*:ResourceWarning''', '''ignore:unclosed transport .*sem_unlink.*FileNotFoundError:pytest.PytestUnraisableExceptionWarning''', '''ignore:(?s)Exception ignored in. Date: Tue, 21 Jul 2026 08:59:41 +0200 Subject: [PATCH 22/28] ruff --- distributed/deploy/local_env.py | 8 ++++++-- distributed/deploy/tests/test_local_env.py | 12 ++++++++++-- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index 123a592697b..8ce11b4ea7f 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -161,8 +161,12 @@ class Scheduler(Process): dask.distributed.Scheduler class """ - def __init__(self, python_executable: str, connect_options: dict, kwargs: dict[str, AnyType] | None = None, -): + def __init__( + self, + python_executable: str, + connect_options: dict, + kwargs: dict[str, AnyType] | None = None, + ): super().__init__() self.python_executable = python_executable diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index 7b38abfbfb7..93033c128cf 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -57,7 +57,11 @@ async def test_bad_executable(): async with LocalEnvCluster( "/foo/bar/baz/python", asynchronous=True, - scheduler_options={"idle_timeout": "5s", "port": 0, "dashboard_address": ":0"}, + scheduler_options={ + "idle_timeout": "5s", + "port": 0, + "dashboard_address": ":0", + }, worker_options={"death_timeout": "5s"}, ) as cluster: assert cluster @@ -75,7 +79,11 @@ def f(): async with LocalEnvCluster( sys.executable, asynchronous=True, - scheduler_options={"idle_timeout": "5s", "port": 0, "dashboard_address": ":0"}, + scheduler_options={ + "idle_timeout": "5s", + "port": 0, + "dashboard_address": ":0", + }, worker_options={"death_timeout": "5s"}, ) as cluster: async with Client(cluster, asynchronous=True) as client: From a47073130be035c78d1357c413051c33e954c3dc Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 21 Jul 2026 09:06:54 +0200 Subject: [PATCH 23/28] Add type annotations --- distributed/deploy/local_env.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index 8ce11b4ea7f..14511e446ad 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -110,7 +110,7 @@ def __init__( python_executable: str, connect_options: dict[str, AnyType] | None = None, worker_module: str = "distributed.cli.dask_worker", - name=None, + name: None = None, kwargs: dict[str, AnyType] | None = None, ): super().__init__() @@ -199,8 +199,8 @@ def LocalEnvCluster( worker_options: dict | None = None, scheduler_options: dict | None = None, worker_module: str = "distributed.cli.dask_worker", - **kwargs, -): + **kwargs: AnyType, +) -> SpecCluster: """Deploy a Dask cluster that utilises a different Python executable The LocalEnvCluster function deploys a Dask Scheduler and Workers for you on From 3865fc878bd01c108e8882b7c46d7b17c282e52a Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 21 Jul 2026 09:15:15 +0200 Subject: [PATCH 24/28] Refactor None handling for kwargs and connect_options --- distributed/deploy/local_env.py | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index 14511e446ad..bcae52decfd 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -117,8 +117,8 @@ def __init__( self.scheduler = scheduler self.python_executable = python_executable - self.connect_options = connect_options - self.kwargs = kwargs or {} + self.connect_options = connect_options if connect_options is not None else {} + self.kwargs = kwargs if kwargs is not None else {} self.worker_module = worker_module self.name = name @@ -170,8 +170,8 @@ def __init__( super().__init__() self.python_executable = python_executable - self.kwargs = kwargs or {} - self.connect_options = connect_options + self.kwargs = kwargs if kwargs is not None else {} + self.connect_options = connect_options if connect_options is not None else {} async def start(self): logger.debug("Created Scheduler") @@ -254,12 +254,7 @@ def LocalEnvCluster( dask.distributed.Worker asyncio.create_subprocess_shell """ - if scheduler_options is None: - scheduler_options = {} - if worker_options is None: - worker_options = {} - if connect_options is None: - connect_options = {} + scheduler = { "cls": Scheduler, "options": { From fa0e193f7fde0e34dd5b7a80293276335a1eb2f8 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Tue, 21 Jul 2026 11:40:06 +0200 Subject: [PATCH 25/28] Try ignoring PytestUnraisableExceptionWarning on Windows --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index ac535599abf..c23649e2433 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -142,6 +142,7 @@ filterwarnings = [ '''ignore:unclosed transport <_SelectorSocketTransport.*:ResourceWarning''', '''ignore:unclosed transport Date: Tue, 21 Jul 2026 15:32:56 +0200 Subject: [PATCH 26/28] No support for Windows --- distributed/deploy/local_env.py | 15 ++++++++++----- distributed/deploy/tests/test_local_env.py | 2 ++ 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index bcae52decfd..58b49e9ff2e 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -5,6 +5,7 @@ import dask import dask.config +from distributed.compatibility import WINDOWS from distributed.core import Status from distributed.scheduler import Scheduler as _Scheduler from distributed.utils import cli_keywords @@ -15,7 +16,7 @@ logger = logging.getLogger(__name__) -class Process(ProcessInterface): +class LocalEnvProcess(ProcessInterface): """A superclass for Workers and Nannies run by a specified Python executable See Also @@ -25,6 +26,10 @@ class Process(ProcessInterface): """ def __init__(self, **kwargs): + if WINDOWS: + raise RuntimeError( + "LocalEnvProcess relies on subprocesses, which are not supported on Windows." + ) self.proc = None super().__init__(**kwargs) @@ -84,7 +89,7 @@ async def _get_address(self, search_str): logger.debug("%s", line) -class Worker(Process): +class LocalEnvWorker(LocalEnvProcess): """A Remote Dask Worker run by a specified Python executable Parameters @@ -147,7 +152,7 @@ async def start(self): await super().start() -class Scheduler(Process): +class LocalEnvScheduler(LocalEnvProcess): """A Remote Dask Scheduler run by a specified Python executable Parameters @@ -256,7 +261,7 @@ def LocalEnvCluster( """ scheduler = { - "cls": Scheduler, + "cls": LocalEnvScheduler, "options": { "python_executable": python_executable, "connect_options": connect_options, @@ -265,7 +270,7 @@ def LocalEnvCluster( } workers = { i: { - "cls": Worker, + "cls": LocalEnvWorker, "options": { "python_executable": python_executable, "connect_options": connect_options, diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index 93033c128cf..a144006b513 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -5,11 +5,13 @@ import dask from distributed import Client +from distributed.compatibility import WINDOWS from distributed.core import Status from distributed.deploy.local_env import LocalEnvCluster from distributed.utils_test import gen_test +@pytest.mark.skipif(WINDOWS, reason="distributed#7434") @gen_test() async def test_basic(): async with LocalEnvCluster( From 85897cd289e4d097601842e831012570397cb948 Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Wed, 22 Jul 2026 11:10:40 +0200 Subject: [PATCH 27/28] No need for Windows workaround, change exception to RuntimeError --- distributed/deploy/local_env.py | 13 ++----------- 1 file changed, 2 insertions(+), 11 deletions(-) diff --git a/distributed/deploy/local_env.py b/distributed/deploy/local_env.py index 58b49e9ff2e..2538b8a5026 100644 --- a/distributed/deploy/local_env.py +++ b/distributed/deploy/local_env.py @@ -58,17 +58,8 @@ async def _set_env_helper(self): await proc.communicate() if proc.returncode == 0: set_env = f'env DASK_INTERNAL_INHERIT_CONFIG="{dask.config.serialize(dask.config.global_config)}"' - else: - proc = await asyncio.create_subprocess_shell( - "cmd /c ver", **self.connect_options - ) - await proc.communicate() - if proc.returncode == 0: - set_env = f"set DASK_INTERNAL_INHERIT_CONFIG={dask.config.serialize(dask.config.global_config)} &&" - else: - name = self.__class__.__name__ - emsg = f"{name} failed to set DASK_INTERNAL_INHERIT_CONFIG variable" - raise Exception(emsg) + else: # pragma: nocover + raise RuntimeError(f"{self.__class__.__name__} failed to set DASK_INTERNAL_INHERIT_CONFIG variable") return set_env async def _get_address(self, search_str): From 53dc38b5d8df514bd83e4cb80bdc53e830ab29dc Mon Sep 17 00:00:00 2001 From: Dawa Ometto Date: Wed, 22 Jul 2026 11:13:48 +0200 Subject: [PATCH 28/28] Actually skip all localenv tests on windows --- distributed/deploy/tests/test_local_env.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/distributed/deploy/tests/test_local_env.py b/distributed/deploy/tests/test_local_env.py index a144006b513..53f015f7f4b 100644 --- a/distributed/deploy/tests/test_local_env.py +++ b/distributed/deploy/tests/test_local_env.py @@ -26,7 +26,7 @@ async def test_basic(): assert "LocalEnv" in repr(cluster) assert cluster.status == Status.closed - +@pytest.mark.skipif(WINDOWS, reason="distributed#7434") @gen_test() async def test_job_submission(): async with LocalEnvCluster( @@ -39,7 +39,7 @@ async def test_job_submission(): result = await client.submit(lambda x: x + 1, 10) assert result == 11 - +@pytest.mark.skipif(WINDOWS, reason="distributed#7434") @gen_test() async def test_multiple_workers(): n_workers = 2 @@ -52,7 +52,7 @@ async def test_multiple_workers(): ) as cluster: assert len(cluster.workers) == n_workers - +@pytest.mark.skipif(WINDOWS, reason="distributed#7434") @gen_test() async def test_bad_executable(): with pytest.raises(RuntimeError): @@ -69,7 +69,7 @@ async def test_bad_executable(): assert cluster cluster.close() - +@pytest.mark.skipif(WINDOWS, reason="distributed#7434") @gen_test() async def test_set_env(): value = 100