From 80e528c869633539eb8d7b404fb4b916142d0441 Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Fri, 4 Jun 2021 13:02:45 +0100
Subject: [PATCH 01/14] 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 "<SSH %s: status=%s>" % (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 0a227a080628d94f210974718cb09470bf752af4 Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Wed, 16 Jun 2021 14:33:09 +0100
Subject: [PATCH 02/14] Different approach for getting scheduler/worker
 addresses

---
 distributed/__init__.py         |  8 +++++++-
 distributed/deploy/__init__.py  |  1 +
 distributed/deploy/local_env.py | 31 +++++++++++++++++++++----------
 3 files changed, 29 insertions(+), 11 deletions(-)

diff --git a/distributed/__init__.py b/distributed/__init__.py
index 9f7a8d6f4bc..c8ab91451d3 100644
--- a/distributed/__init__.py
+++ b/distributed/__init__.py
@@ -20,7 +20,13 @@
     wait,
 )
 from .core import Status, connect, rpc
-from .deploy import Adaptive, LocalCluster, SpecCluster, SSHCluster
+from .deploy import (
+    Adaptive,
+    LocalCluster,
+    LocalEnvCluster,
+    SpecCluster,
+    SSHCluster,
+)
 from .diagnostics.plugin import (
     Environ,
     NannyPlugin,
diff --git a/distributed/deploy/__init__.py b/distributed/deploy/__init__.py
index 1518942dc4c..e182fee85b3 100644
--- a/distributed/deploy/__init__.py
+++ b/distributed/deploy/__init__.py
@@ -3,6 +3,7 @@
 from .adaptive import Adaptive
 from .cluster import Cluster
 from .local import LocalCluster
+from .local_env import LocalEnvCluster
 from .spec import ProcessInterface, SpecCluster
 from .ssh import SSHCluster
 
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 40d9cd30bc517672f66fe152da3e4a026bab5aaf Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Thu, 1 Jul 2021 12:29:48 +0100
Subject: [PATCH 03/14] 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 a5c08769dc259540ce02d9552a15b736bf34c67a Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Thu, 1 Jul 2021 13:32:40 +0100
Subject: [PATCH 04/14] 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 93ba77010d9e3b547cee30bd04c59cda85d6bbdd Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Thu, 1 Jul 2021 13:45:48 +0100
Subject: [PATCH 05/14] 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 f3768ca366c967bf9530e67e775a1817ea9958b0 Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Mon, 12 Jul 2021 09:33:46 +0100
Subject: [PATCH 06/14] 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 7038903204ade9faf70836270661de1719ce9a60 Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Mon, 12 Jul 2021 13:15:13 +0100
Subject: [PATCH 07/14] 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 "<SSH %s: status=%s>" % (type(self).__name__, self.status)
+        return "<LocalEnv %s: status=%s>" % (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 325ee8030298c4abd685eb3c682bb312476a712d Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Mon, 19 Jul 2021 16:55:32 +0100
Subject: [PATCH 08/14] 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 1828c81d870536e97485579c842d579bde07a3df Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Mon, 19 Jul 2021 16:57:52 +0100
Subject: [PATCH 09/14] 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 6b5002ba5254a1d01c769b9cb8e2ffc56524285b Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Mon, 2 Aug 2021 13:35:52 +0100
Subject: [PATCH 10/14] 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 0eadd56bcd3ca589a3820071fd0a41b9b1ec4523 Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Wed, 4 Aug 2021 15:28:20 +0100
Subject: [PATCH 11/14] 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 0428b7a3fa509a92128b12adbf43b279c153cb2c Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Thu, 5 Aug 2021 10:58:58 +0100
Subject: [PATCH 12/14] 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 bffcecede2ab97afbe1e8a7ce54d964d0e2c97ac Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Thu, 5 Aug 2021 10:59:50 +0100
Subject: [PATCH 13/14] 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 dbdccea9905e2afceaed64850e50686673c3057b Mon Sep 17 00:00:00 2001
From: DPeterK <peter.killick@informaticslab.co.uk>
Date: Thu, 5 Aug 2021 13:48:12 +0100
Subject: [PATCH 14/14] Tests pass

---
 distributed/__init__.py                    |  8 +-
 distributed/deploy/local_env.py            | 97 ++++++----------------
 distributed/deploy/tests/test_local_env.py | 26 +++---
 3 files changed, 40 insertions(+), 91 deletions(-)

diff --git a/distributed/__init__.py b/distributed/__init__.py
index c8ab91451d3..6b2124440f0 100644
--- a/distributed/__init__.py
+++ b/distributed/__init__.py
@@ -20,13 +20,7 @@
     wait,
 )
 from .core import Status, connect, rpc
-from .deploy import (
-    Adaptive,
-    LocalCluster,
-    LocalEnvCluster,
-    SpecCluster,
-    SSHCluster,
-)
+from .deploy import Adaptive, LocalCluster, LocalEnvCluster, SpecCluster, SSHCluster
 from .diagnostics.plugin import (
     Environ,
     NannyPlugin,
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