We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
remote_path is ignored in SSHCluster
Current State:

In the doc: https://docs.dask.org/en/latest/setup/ssh.html#cmdoption-dask-ssh-remote-python
It is not specified that one can use `remote_python` in SSHCluster, although both `worker_options` and `scheduler_options` support it:
- https://github.com/dask/distributed/blob/master/distributed/deploy/ssh.py#L87
- https://github.com/dask/distributed/blob/master/distributed/deploy/ssh.py#L163

Furthermore, these parameters are ignored when passed, dask always using `sys.executable` in both remote scheduler and workers.

Expected behavior:
I would like to pass `remote_python` to both `worker_options` and `scheduler_options`.

Implementation details:

- https://github.com/dask/distributed/blob/master/distributed/deploy/ssh.py#L113
- https://github.com/dask/distributed/blob/master/distributed/deploy/ssh.py#L189

Both calls should replaced with `self.remote_python or sys.executable`, as it is in [old_ssh.py](https://github.com/dask/distributed/blob/master/distributed/deploy/old_ssh.py#L295)


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `distributed/deploy/ssh.py`
Content:
```
1 import logging
2 import sys
3 from typing import List
4 import warnings
5 import weakref
6 
7 import dask
8 
9 from .spec import SpecCluster, ProcessInterface
10 from ..utils import cli_keywords
11 from ..scheduler import Scheduler as _Scheduler
12 from ..worker import Worker as _Worker
13 from ..utils import serialize_for_cli
14 
15 logger = logging.getLogger(__name__)
16 
17 
18 class Process(ProcessInterface):
19     """ A superclass for SSH Workers and Nannies
20 
21     See Also
22     --------
23     Worker
24     Scheduler
25     """
26 
27     def __init__(self, **kwargs):
28         self.connection = None
29         self.proc = None
30         super().__init__(**kwargs)
31 
32     async def start(self):
33         assert self.connection
34         weakref.finalize(
35             self, self.proc.kill
36         )  # https://github.com/ronf/asyncssh/issues/112
37         await super().start()
38 
39     async def close(self):
40         self.proc.kill()  # https://github.com/ronf/asyncssh/issues/112
41         self.connection.close()
42         await super().close()
43 
44     def __repr__(self):
45         return "<SSH %s: status=%s>" % (type(self).__name__, self.status)
46 
47 
48 class Worker(Process):
49     """ A Remote Dask Worker controled by SSH
50 
51     Parameters
52     ----------
53     scheduler: str
54         The address of the scheduler
55     address: str
56         The hostname where we should run this worker
57     worker_module: str
58         The python module to run to start the worker.
59     connect_options: dict
60         kwargs to be passed to asyncssh connections
61     remote_python: str
62         Path to Python on remote node to run this worker.
63     kwargs: dict
64         These will be passed through the dask-worker CLI to the
65         dask.distributed.Worker class
66     """
67 
68     def __init__(
69         self,
70         scheduler: str,
71         address: str,
72         connect_options: dict,
73         kwargs: dict,
74         worker_module="distributed.cli.dask_worker",
75         remote_python=None,
76         loop=None,
77         name=None,
78     ):
79         super().__init__()
80 
81         self.address = address
82         self.scheduler = scheduler
83         self.worker_module = worker_module
84         self.connect_options = connect_options
85         self.kwargs = kwargs
86         self.name = name
87         self.remote_python = remote_python
88 
89     async def start(self):
90         import asyncssh  # import now to avoid adding to module startup time
91 
92         self.connection = await asyncssh.connect(self.address, **self.connect_options)
93 
94         result = await self.connection.run("uname")
95         if result.exit_status == 0:
96             set_env = 'env DASK_INTERNAL_INHERIT_CONFIG="{}"'.format(
97                 serialize_for_cli(dask.config.global_config)
98             )
99         else:
100             result = await self.connection.run("cmd /c ver")
101             if result.exit_status == 0:
102                 set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format(
103                     serialize_for_cli(dask.config.global_config)
104                 )
105             else:
106                 raise Exception(
107                     "Worker failed to set DASK_INTERNAL_INHERIT_CONFIG variable "
108                 )
109 
110         cmd = " ".join(
111             [
112                 set_env,
113                 sys.executable,
114                 "-m",
115                 self.worker_module,
116                 self.scheduler,
117                 "--name",
118                 str(self.name),
119             ]
120             + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module)
121         )
122 
123         self.proc = await self.connection.create_process(cmd)
124 
125         # We watch stderr in order to get the address, then we return
126         while True:
127             line = await self.proc.stderr.readline()
128             if not line.strip():
129                 raise Exception("Worker failed to start")
130             logger.info(line.strip())
131             if "worker at" in line:
132                 self.address = line.split("worker at:")[1].strip()
133                 self.status = "running"
134                 break
135         logger.debug("%s", line)
136         await super().start()
137 
138 
139 class Scheduler(Process):
140     """ A Remote Dask Scheduler controlled by SSH
141 
142     Parameters
143     ----------
144     address: str
145         The hostname where we should run this worker
146     connect_options: dict
147         kwargs to be passed to asyncssh connections
148     remote_python: str
149         Path to Python on remote node to run this scheduler.
150     kwargs: dict
151         These will be passed through the dask-scheduler CLI to the
152         dask.distributed.Scheduler class
153     """
154 
155     def __init__(
156         self, address: str, connect_options: dict, kwargs: dict, remote_python=None
157     ):
158         super().__init__()
159 
160         self.address = address
161         self.kwargs = kwargs
162         self.connect_options = connect_options
163         self.remote_python = remote_python
164 
165     async def start(self):
166         import asyncssh  # import now to avoid adding to module startup time
167 
168         logger.debug("Created Scheduler Connection")
169 
170         self.connection = await asyncssh.connect(self.address, **self.connect_options)
171 
172         result = await self.connection.run("uname")
173         if result.exit_status == 0:
174             set_env = 'env DASK_INTERNAL_INHERIT_CONFIG="{}"'.format(
175                 serialize_for_cli(dask.config.global_config)
176             )
177         else:
178             result = await self.connection.run("cmd /c ver")
179             if result.exit_status == 0:
180                 set_env = "set DASK_INTERNAL_INHERIT_CONFIG={} &&".format(
181                     serialize_for_cli(dask.config.global_config)
182                 )
183             else:
184                 raise Exception(
185                     "Scheduler failed to set DASK_INTERNAL_INHERIT_CONFIG variable "
186                 )
187 
188         cmd = " ".join(
189             [set_env, sys.executable, "-m", "distributed.cli.dask_scheduler",]
190             + cli_keywords(self.kwargs, cls=_Scheduler)
191         )
192         self.proc = await self.connection.create_process(cmd)
193 
194         # We watch stderr in order to get the address, then we return
195         while True:
196             line = await self.proc.stderr.readline()
197             if not line.strip():
198                 raise Exception("Worker failed to start")
199             logger.info(line.strip())
200             if "Scheduler at" in line:
201                 self.address = line.split("Scheduler at:")[1].strip()
202                 break
203         logger.debug("%s", line)
204         await super().start()
205 
206 
207 old_cluster_kwargs = {
208     "scheduler_addr",
209     "scheduler_port",
210     "worker_addrs",
211     "nthreads",
212     "nprocs",
213     "ssh_username",
214     "ssh_port",
215     "ssh_private_key",
216     "nohost",
217     "logdir",
218     "remote_python",
219     "memory_limit",
220     "worker_port",
221     "nanny_port",
222     "remote_dask_worker",
223 }
224 
225 
226 def SSHCluster(
227     hosts: List[str] = None,
228     connect_options: dict = {},
229     worker_options: dict = {},
230     scheduler_options: dict = {},
231     worker_module: str = "distributed.cli.dask_worker",
232     remote_python: str = None,
233     **kwargs
234 ):
235     """ Deploy a Dask cluster using SSH
236 
237     The SSHCluster function deploys a Dask Scheduler and Workers for you on a
238     set of machine addresses that you provide.  The first address will be used
239     for the scheduler while the rest will be used for the workers (feel free to
240     repeat the first hostname if you want to have the scheduler and worker
241     co-habitate one machine.)
242 
243     You may configure the scheduler and workers by passing
244     ``scheduler_options`` and ``worker_options`` dictionary keywords.  See the
245     ``dask.distributed.Scheduler`` and ``dask.distributed.Worker`` classes for
246     details on the available options, but the defaults should work in most
247     situations.
248 
249     You may configure your use of SSH itself using the ``connect_options``
250     keyword, which passes values to the ``asyncssh.connect`` function.  For
251     more information on these see the documentation for the ``asyncssh``
252     library https://asyncssh.readthedocs.io .
253 
254     Parameters
255     ----------
256     hosts: List[str]
257         List of hostnames or addresses on which to launch our cluster.
258         The first will be used for the scheduler and the rest for workers.
259     connect_options: dict, optional
260         Keywords to pass through to ``asyncssh.connect``.
261     worker_options: dict, optional
262         Keywords to pass on to workers.
263     scheduler_options: dict, optional
264         Keywords to pass on to scheduler.
265     worker_module: str, optional
266         Python module to call to start the worker.
267     remote_python: str, optional
268         Path to Python on remote nodes.
269 
270     Examples
271     --------
272     >>> from dask.distributed import Client, SSHCluster
273     >>> cluster = SSHCluster(
274     ...     ["localhost", "localhost", "localhost", "localhost"],
275     ...     connect_options={"known_hosts": None},
276     ...     worker_options={"nthreads": 2},
277     ...     scheduler_options={"port": 0, "dashboard_address": ":8797"}
278     ... )
279     >>> client = Client(cluster)
280 
281     An example using a different worker module, in particular the
282     ``dask-cuda-worker`` command from the ``dask-cuda`` project.
283 
284     >>> from dask.distributed import Client, SSHCluster
285     >>> cluster = SSHCluster(
286     ...     ["localhost", "hostwithgpus", "anothergpuhost"],
287     ...     connect_options={"known_hosts": None},
288     ...     scheduler_options={"port": 0, "dashboard_address": ":8797"},
289     ...     worker_module='dask_cuda.dask_cuda_worker')
290     >>> client = Client(cluster)
291 
292     See Also
293     --------
294     dask.distributed.Scheduler
295     dask.distributed.Worker
296     asyncssh.connect
297     """
298     if set(kwargs) & old_cluster_kwargs:
299         from .old_ssh import SSHCluster as OldSSHCluster
300 
301         warnings.warn(
302             "Note that the SSHCluster API has been replaced.  "
303             "We're routing you to the older implementation.  "
304             "This will be removed in the future"
305         )
306         kwargs.setdefault("worker_addrs", hosts)
307         return OldSSHCluster(**kwargs)
308 
309     scheduler = {
310         "cls": Scheduler,
311         "options": {
312             "address": hosts[0],
313             "connect_options": connect_options,
314             "kwargs": scheduler_options,
315             "remote_python": remote_python,
316         },
317     }
318     workers = {
319         i: {
320             "cls": Worker,
321             "options": {
322                 "address": host,
323                 "connect_options": connect_options,
324                 "kwargs": worker_options,
325                 "worker_module": worker_module,
326                 "remote_python": remote_python,
327             },
328         }
329         for i, host in enumerate(hosts[1:])
330     }
331     return SpecCluster(workers, scheduler, name="SSHCluster", **kwargs)
332
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/distributed/deploy/ssh.py b/distributed/deploy/ssh.py
--- a/distributed/deploy/ssh.py
+++ b/distributed/deploy/ssh.py
@@ -110,7 +110,7 @@
         cmd = " ".join(
             [
                 set_env,
-                sys.executable,
+                self.remote_python or sys.executable,
                 "-m",
                 self.worker_module,
                 self.scheduler,
@@ -186,7 +186,12 @@
                 )
 
         cmd = " ".join(
-            [set_env, sys.executable, "-m", "distributed.cli.dask_scheduler",]
+            [
+                set_env,
+                self.remote_python or sys.executable,
+                "-m",
+                "distributed.cli.dask_scheduler",
+            ]
             + cli_keywords(self.kwargs, cls=_Scheduler)
         )
         self.proc = await self.connection.create_process(cmd)

{"golden_diff": "diff --git a/distributed/deploy/ssh.py b/distributed/deploy/ssh.py\n--- a/distributed/deploy/ssh.py\n+++ b/distributed/deploy/ssh.py\n@@ -110,7 +110,7 @@\n         cmd = \" \".join(\n             [\n                 set_env,\n-                sys.executable,\n+                self.remote_python or sys.executable,\n                 \"-m\",\n                 self.worker_module,\n                 self.scheduler,\n@@ -186,7 +186,12 @@\n                 )\n \n         cmd = \" \".join(\n-            [set_env, sys.executable, \"-m\", \"distributed.cli.dask_scheduler\",]\n+            [\n+                set_env,\n+                self.remote_python or sys.executable,\n+                \"-m\",\n+                \"distributed.cli.dask_scheduler\",\n+            ]\n             + cli_keywords(self.kwargs, cls=_Scheduler)\n         )\n         self.proc = await self.connection.create_process(cmd)\n", "issue": "remote_path is ignored in SSHCluster\nCurrent State:\r\n\r\nIn the doc: https://docs.dask.org/en/latest/setup/ssh.html#cmdoption-dask-ssh-remote-python\r\nIt is not specified that one can use `remote_python` in SSHCluster, although both `worker_options` and `scheduler_options` support it:\r\n- https://github.com/dask/distributed/blob/master/distributed/deploy/ssh.py#L87\r\n- https://github.com/dask/distributed/blob/master/distributed/deploy/ssh.py#L163\r\n\r\nFurthermore, these parameters are ignored when passed, dask always using `sys.executable` in both remote scheduler and workers.\r\n\r\nExpected behavior:\r\nI would like to pass `remote_python` to both `worker_options` and `scheduler_options`.\r\n\r\nImplementation details:\r\n\r\n- https://github.com/dask/distributed/blob/master/distributed/deploy/ssh.py#L113\r\n- https://github.com/dask/distributed/blob/master/distributed/deploy/ssh.py#L189\r\n\r\nBoth calls should replaced with `self.remote_python or sys.executable`, as it is in [old_ssh.py](https://github.com/dask/distributed/blob/master/distributed/deploy/old_ssh.py#L295)\r\n\n", "before_files": [{"content": "import logging\nimport sys\nfrom typing import List\nimport warnings\nimport weakref\n\nimport dask\n\nfrom .spec import SpecCluster, ProcessInterface\nfrom ..utils import cli_keywords\nfrom ..scheduler import Scheduler as _Scheduler\nfrom ..worker import Worker as _Worker\nfrom ..utils import serialize_for_cli\n\nlogger = logging.getLogger(__name__)\n\n\nclass Process(ProcessInterface):\n    \"\"\" A superclass for SSH Workers and Nannies\n\n    See Also\n    --------\n    Worker\n    Scheduler\n    \"\"\"\n\n    def __init__(self, **kwargs):\n        self.connection = None\n        self.proc = None\n        super().__init__(**kwargs)\n\n    async def start(self):\n        assert self.connection\n        weakref.finalize(\n            self, self.proc.kill\n        )  # https://github.com/ronf/asyncssh/issues/112\n        await super().start()\n\n    async def close(self):\n        self.proc.kill()  # https://github.com/ronf/asyncssh/issues/112\n        self.connection.close()\n        await super().close()\n\n    def __repr__(self):\n        return \"<SSH %s: status=%s>\" % (type(self).__name__, self.status)\n\n\nclass Worker(Process):\n    \"\"\" A Remote Dask Worker controled by SSH\n\n    Parameters\n    ----------\n    scheduler: str\n        The address of the scheduler\n    address: str\n        The hostname where we should run this worker\n    worker_module: str\n        The python module to run to start the worker.\n    connect_options: dict\n        kwargs to be passed to asyncssh connections\n    remote_python: str\n        Path to Python on remote node to run this worker.\n    kwargs: dict\n        These will be passed through the dask-worker CLI to the\n        dask.distributed.Worker class\n    \"\"\"\n\n    def __init__(\n        self,\n        scheduler: str,\n        address: str,\n        connect_options: dict,\n        kwargs: dict,\n        worker_module=\"distributed.cli.dask_worker\",\n        remote_python=None,\n        loop=None,\n        name=None,\n    ):\n        super().__init__()\n\n        self.address = address\n        self.scheduler = scheduler\n        self.worker_module = worker_module\n        self.connect_options = connect_options\n        self.kwargs = kwargs\n        self.name = name\n        self.remote_python = remote_python\n\n    async def start(self):\n        import asyncssh  # import now to avoid adding to module startup time\n\n        self.connection = await asyncssh.connect(self.address, **self.connect_options)\n\n        result = await self.connection.run(\"uname\")\n        if result.exit_status == 0:\n            set_env = 'env DASK_INTERNAL_INHERIT_CONFIG=\"{}\"'.format(\n                serialize_for_cli(dask.config.global_config)\n            )\n        else:\n            result = await self.connection.run(\"cmd /c ver\")\n            if result.exit_status == 0:\n                set_env = \"set DASK_INTERNAL_INHERIT_CONFIG={} &&\".format(\n                    serialize_for_cli(dask.config.global_config)\n                )\n            else:\n                raise Exception(\n                    \"Worker failed to set DASK_INTERNAL_INHERIT_CONFIG variable \"\n                )\n\n        cmd = \" \".join(\n            [\n                set_env,\n                sys.executable,\n                \"-m\",\n                self.worker_module,\n                self.scheduler,\n                \"--name\",\n                str(self.name),\n            ]\n            + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module)\n        )\n\n        self.proc = await self.connection.create_process(cmd)\n\n        # We watch stderr in order to get the address, then we return\n        while True:\n            line = await self.proc.stderr.readline()\n            if not line.strip():\n                raise Exception(\"Worker failed to start\")\n            logger.info(line.strip())\n            if \"worker at\" in line:\n                self.address = line.split(\"worker at:\")[1].strip()\n                self.status = \"running\"\n                break\n        logger.debug(\"%s\", line)\n        await super().start()\n\n\nclass Scheduler(Process):\n    \"\"\" A Remote Dask Scheduler controlled by SSH\n\n    Parameters\n    ----------\n    address: str\n        The hostname where we should run this worker\n    connect_options: dict\n        kwargs to be passed to asyncssh connections\n    remote_python: str\n        Path to Python on remote node to run this scheduler.\n    kwargs: dict\n        These will be passed through the dask-scheduler CLI to the\n        dask.distributed.Scheduler class\n    \"\"\"\n\n    def __init__(\n        self, address: str, connect_options: dict, kwargs: dict, remote_python=None\n    ):\n        super().__init__()\n\n        self.address = address\n        self.kwargs = kwargs\n        self.connect_options = connect_options\n        self.remote_python = remote_python\n\n    async def start(self):\n        import asyncssh  # import now to avoid adding to module startup time\n\n        logger.debug(\"Created Scheduler Connection\")\n\n        self.connection = await asyncssh.connect(self.address, **self.connect_options)\n\n        result = await self.connection.run(\"uname\")\n        if result.exit_status == 0:\n            set_env = 'env DASK_INTERNAL_INHERIT_CONFIG=\"{}\"'.format(\n                serialize_for_cli(dask.config.global_config)\n            )\n        else:\n            result = await self.connection.run(\"cmd /c ver\")\n            if result.exit_status == 0:\n                set_env = \"set DASK_INTERNAL_INHERIT_CONFIG={} &&\".format(\n                    serialize_for_cli(dask.config.global_config)\n                )\n            else:\n                raise Exception(\n                    \"Scheduler failed to set DASK_INTERNAL_INHERIT_CONFIG variable \"\n                )\n\n        cmd = \" \".join(\n            [set_env, sys.executable, \"-m\", \"distributed.cli.dask_scheduler\",]\n            + cli_keywords(self.kwargs, cls=_Scheduler)\n        )\n        self.proc = await self.connection.create_process(cmd)\n\n        # We watch stderr in order to get the address, then we return\n        while True:\n            line = await self.proc.stderr.readline()\n            if not line.strip():\n                raise Exception(\"Worker failed to start\")\n            logger.info(line.strip())\n            if \"Scheduler at\" in line:\n                self.address = line.split(\"Scheduler at:\")[1].strip()\n                break\n        logger.debug(\"%s\", line)\n        await super().start()\n\n\nold_cluster_kwargs = {\n    \"scheduler_addr\",\n    \"scheduler_port\",\n    \"worker_addrs\",\n    \"nthreads\",\n    \"nprocs\",\n    \"ssh_username\",\n    \"ssh_port\",\n    \"ssh_private_key\",\n    \"nohost\",\n    \"logdir\",\n    \"remote_python\",\n    \"memory_limit\",\n    \"worker_port\",\n    \"nanny_port\",\n    \"remote_dask_worker\",\n}\n\n\ndef SSHCluster(\n    hosts: List[str] = None,\n    connect_options: dict = {},\n    worker_options: dict = {},\n    scheduler_options: dict = {},\n    worker_module: str = \"distributed.cli.dask_worker\",\n    remote_python: str = None,\n    **kwargs\n):\n    \"\"\" Deploy a Dask cluster using SSH\n\n    The SSHCluster function deploys a Dask Scheduler and Workers for you on a\n    set of machine addresses that you provide.  The first address will be used\n    for the scheduler while the rest will be used for the workers (feel free to\n    repeat the first hostname if you want to have the scheduler and worker\n    co-habitate one machine.)\n\n    You may configure the scheduler and workers by passing\n    ``scheduler_options`` and ``worker_options`` dictionary keywords.  See the\n    ``dask.distributed.Scheduler`` and ``dask.distributed.Worker`` classes for\n    details on the available options, but the defaults should work in most\n    situations.\n\n    You may configure your use of SSH itself using the ``connect_options``\n    keyword, which passes values to the ``asyncssh.connect`` function.  For\n    more information on these see the documentation for the ``asyncssh``\n    library https://asyncssh.readthedocs.io .\n\n    Parameters\n    ----------\n    hosts: List[str]\n        List of hostnames or addresses on which to launch our cluster.\n        The first will be used for the scheduler and the rest for workers.\n    connect_options: dict, optional\n        Keywords to pass through to ``asyncssh.connect``.\n    worker_options: dict, optional\n        Keywords to pass on to workers.\n    scheduler_options: dict, optional\n        Keywords to pass on to scheduler.\n    worker_module: str, optional\n        Python module to call to start the worker.\n    remote_python: str, optional\n        Path to Python on remote nodes.\n\n    Examples\n    --------\n    >>> from dask.distributed import Client, SSHCluster\n    >>> cluster = SSHCluster(\n    ...     [\"localhost\", \"localhost\", \"localhost\", \"localhost\"],\n    ...     connect_options={\"known_hosts\": None},\n    ...     worker_options={\"nthreads\": 2},\n    ...     scheduler_options={\"port\": 0, \"dashboard_address\": \":8797\"}\n    ... )\n    >>> client = Client(cluster)\n\n    An example using a different worker module, in particular the\n    ``dask-cuda-worker`` command from the ``dask-cuda`` project.\n\n    >>> from dask.distributed import Client, SSHCluster\n    >>> cluster = SSHCluster(\n    ...     [\"localhost\", \"hostwithgpus\", \"anothergpuhost\"],\n    ...     connect_options={\"known_hosts\": None},\n    ...     scheduler_options={\"port\": 0, \"dashboard_address\": \":8797\"},\n    ...     worker_module='dask_cuda.dask_cuda_worker')\n    >>> client = Client(cluster)\n\n    See Also\n    --------\n    dask.distributed.Scheduler\n    dask.distributed.Worker\n    asyncssh.connect\n    \"\"\"\n    if set(kwargs) & old_cluster_kwargs:\n        from .old_ssh import SSHCluster as OldSSHCluster\n\n        warnings.warn(\n            \"Note that the SSHCluster API has been replaced.  \"\n            \"We're routing you to the older implementation.  \"\n            \"This will be removed in the future\"\n        )\n        kwargs.setdefault(\"worker_addrs\", hosts)\n        return OldSSHCluster(**kwargs)\n\n    scheduler = {\n        \"cls\": Scheduler,\n        \"options\": {\n            \"address\": hosts[0],\n            \"connect_options\": connect_options,\n            \"kwargs\": scheduler_options,\n            \"remote_python\": remote_python,\n        },\n    }\n    workers = {\n        i: {\n            \"cls\": Worker,\n            \"options\": {\n                \"address\": host,\n                \"connect_options\": connect_options,\n                \"kwargs\": worker_options,\n                \"worker_module\": worker_module,\n                \"remote_python\": remote_python,\n            },\n        }\n        for i, host in enumerate(hosts[1:])\n    }\n    return SpecCluster(workers, scheduler, name=\"SSHCluster\", **kwargs)\n", "path": "distributed/deploy/ssh.py"}], "after_files": [{"content": "import logging\nimport sys\nfrom typing import List\nimport warnings\nimport weakref\n\nimport dask\n\nfrom .spec import SpecCluster, ProcessInterface\nfrom ..utils import cli_keywords\nfrom ..scheduler import Scheduler as _Scheduler\nfrom ..worker import Worker as _Worker\nfrom ..utils import serialize_for_cli\n\nlogger = logging.getLogger(__name__)\n\n\nclass Process(ProcessInterface):\n    \"\"\" A superclass for SSH Workers and Nannies\n\n    See Also\n    --------\n    Worker\n    Scheduler\n    \"\"\"\n\n    def __init__(self, **kwargs):\n        self.connection = None\n        self.proc = None\n        super().__init__(**kwargs)\n\n    async def start(self):\n        assert self.connection\n        weakref.finalize(\n            self, self.proc.kill\n        )  # https://github.com/ronf/asyncssh/issues/112\n        await super().start()\n\n    async def close(self):\n        self.proc.kill()  # https://github.com/ronf/asyncssh/issues/112\n        self.connection.close()\n        await super().close()\n\n    def __repr__(self):\n        return \"<SSH %s: status=%s>\" % (type(self).__name__, self.status)\n\n\nclass Worker(Process):\n    \"\"\" A Remote Dask Worker controled by SSH\n\n    Parameters\n    ----------\n    scheduler: str\n        The address of the scheduler\n    address: str\n        The hostname where we should run this worker\n    worker_module: str\n        The python module to run to start the worker.\n    connect_options: dict\n        kwargs to be passed to asyncssh connections\n    remote_python: str\n        Path to Python on remote node to run this worker.\n    kwargs: dict\n        These will be passed through the dask-worker CLI to the\n        dask.distributed.Worker class\n    \"\"\"\n\n    def __init__(\n        self,\n        scheduler: str,\n        address: str,\n        connect_options: dict,\n        kwargs: dict,\n        worker_module=\"distributed.cli.dask_worker\",\n        remote_python=None,\n        loop=None,\n        name=None,\n    ):\n        super().__init__()\n\n        self.address = address\n        self.scheduler = scheduler\n        self.worker_module = worker_module\n        self.connect_options = connect_options\n        self.kwargs = kwargs\n        self.name = name\n        self.remote_python = remote_python\n\n    async def start(self):\n        import asyncssh  # import now to avoid adding to module startup time\n\n        self.connection = await asyncssh.connect(self.address, **self.connect_options)\n\n        result = await self.connection.run(\"uname\")\n        if result.exit_status == 0:\n            set_env = 'env DASK_INTERNAL_INHERIT_CONFIG=\"{}\"'.format(\n                serialize_for_cli(dask.config.global_config)\n            )\n        else:\n            result = await self.connection.run(\"cmd /c ver\")\n            if result.exit_status == 0:\n                set_env = \"set DASK_INTERNAL_INHERIT_CONFIG={} &&\".format(\n                    serialize_for_cli(dask.config.global_config)\n                )\n            else:\n                raise Exception(\n                    \"Worker failed to set DASK_INTERNAL_INHERIT_CONFIG variable \"\n                )\n\n        cmd = \" \".join(\n            [\n                set_env,\n                self.remote_python or sys.executable,\n                \"-m\",\n                self.worker_module,\n                self.scheduler,\n                \"--name\",\n                str(self.name),\n            ]\n            + cli_keywords(self.kwargs, cls=_Worker, cmd=self.worker_module)\n        )\n\n        self.proc = await self.connection.create_process(cmd)\n\n        # We watch stderr in order to get the address, then we return\n        while True:\n            line = await self.proc.stderr.readline()\n            if not line.strip():\n                raise Exception(\"Worker failed to start\")\n            logger.info(line.strip())\n            if \"worker at\" in line:\n                self.address = line.split(\"worker at:\")[1].strip()\n                self.status = \"running\"\n                break\n        logger.debug(\"%s\", line)\n        await super().start()\n\n\nclass Scheduler(Process):\n    \"\"\" A Remote Dask Scheduler controlled by SSH\n\n    Parameters\n    ----------\n    address: str\n        The hostname where we should run this worker\n    connect_options: dict\n        kwargs to be passed to asyncssh connections\n    remote_python: str\n        Path to Python on remote node to run this scheduler.\n    kwargs: dict\n        These will be passed through the dask-scheduler CLI to the\n        dask.distributed.Scheduler class\n    \"\"\"\n\n    def __init__(\n        self, address: str, connect_options: dict, kwargs: dict, remote_python=None\n    ):\n        super().__init__()\n\n        self.address = address\n        self.kwargs = kwargs\n        self.connect_options = connect_options\n        self.remote_python = remote_python\n\n    async def start(self):\n        import asyncssh  # import now to avoid adding to module startup time\n\n        logger.debug(\"Created Scheduler Connection\")\n\n        self.connection = await asyncssh.connect(self.address, **self.connect_options)\n\n        result = await self.connection.run(\"uname\")\n        if result.exit_status == 0:\n            set_env = 'env DASK_INTERNAL_INHERIT_CONFIG=\"{}\"'.format(\n                serialize_for_cli(dask.config.global_config)\n            )\n        else:\n            result = await self.connection.run(\"cmd /c ver\")\n            if result.exit_status == 0:\n                set_env = \"set DASK_INTERNAL_INHERIT_CONFIG={} &&\".format(\n                    serialize_for_cli(dask.config.global_config)\n                )\n            else:\n                raise Exception(\n                    \"Scheduler failed to set DASK_INTERNAL_INHERIT_CONFIG variable \"\n                )\n\n        cmd = \" \".join(\n            [\n                set_env,\n                self.remote_python or sys.executable,\n                \"-m\",\n                \"distributed.cli.dask_scheduler\",\n            ]\n            + cli_keywords(self.kwargs, cls=_Scheduler)\n        )\n        self.proc = await self.connection.create_process(cmd)\n\n        # We watch stderr in order to get the address, then we return\n        while True:\n            line = await self.proc.stderr.readline()\n            if not line.strip():\n                raise Exception(\"Worker failed to start\")\n            logger.info(line.strip())\n            if \"Scheduler at\" in line:\n                self.address = line.split(\"Scheduler at:\")[1].strip()\n                break\n        logger.debug(\"%s\", line)\n        await super().start()\n\n\nold_cluster_kwargs = {\n    \"scheduler_addr\",\n    \"scheduler_port\",\n    \"worker_addrs\",\n    \"nthreads\",\n    \"nprocs\",\n    \"ssh_username\",\n    \"ssh_port\",\n    \"ssh_private_key\",\n    \"nohost\",\n    \"logdir\",\n    \"remote_python\",\n    \"memory_limit\",\n    \"worker_port\",\n    \"nanny_port\",\n    \"remote_dask_worker\",\n}\n\n\ndef SSHCluster(\n    hosts: List[str] = None,\n    connect_options: dict = {},\n    worker_options: dict = {},\n    scheduler_options: dict = {},\n    worker_module: str = \"distributed.cli.dask_worker\",\n    remote_python: str = None,\n    **kwargs\n):\n    \"\"\" Deploy a Dask cluster using SSH\n\n    The SSHCluster function deploys a Dask Scheduler and Workers for you on a\n    set of machine addresses that you provide.  The first address will be used\n    for the scheduler while the rest will be used for the workers (feel free to\n    repeat the first hostname if you want to have the scheduler and worker\n    co-habitate one machine.)\n\n    You may configure the scheduler and workers by passing\n    ``scheduler_options`` and ``worker_options`` dictionary keywords.  See the\n    ``dask.distributed.Scheduler`` and ``dask.distributed.Worker`` classes for\n    details on the available options, but the defaults should work in most\n    situations.\n\n    You may configure your use of SSH itself using the ``connect_options``\n    keyword, which passes values to the ``asyncssh.connect`` function.  For\n    more information on these see the documentation for the ``asyncssh``\n    library https://asyncssh.readthedocs.io .\n\n    Parameters\n    ----------\n    hosts: List[str]\n        List of hostnames or addresses on which to launch our cluster.\n        The first will be used for the scheduler and the rest for workers.\n    connect_options: dict, optional\n        Keywords to pass through to ``asyncssh.connect``.\n    worker_options: dict, optional\n        Keywords to pass on to workers.\n    scheduler_options: dict, optional\n        Keywords to pass on to scheduler.\n    worker_module: str, optional\n        Python module to call to start the worker.\n    remote_python: str, optional\n        Path to Python on remote nodes.\n\n    Examples\n    --------\n    >>> from dask.distributed import Client, SSHCluster\n    >>> cluster = SSHCluster(\n    ...     [\"localhost\", \"localhost\", \"localhost\", \"localhost\"],\n    ...     connect_options={\"known_hosts\": None},\n    ...     worker_options={\"nthreads\": 2},\n    ...     scheduler_options={\"port\": 0, \"dashboard_address\": \":8797\"}\n    ... )\n    >>> client = Client(cluster)\n\n    An example using a different worker module, in particular the\n    ``dask-cuda-worker`` command from the ``dask-cuda`` project.\n\n    >>> from dask.distributed import Client, SSHCluster\n    >>> cluster = SSHCluster(\n    ...     [\"localhost\", \"hostwithgpus\", \"anothergpuhost\"],\n    ...     connect_options={\"known_hosts\": None},\n    ...     scheduler_options={\"port\": 0, \"dashboard_address\": \":8797\"},\n    ...     worker_module='dask_cuda.dask_cuda_worker')\n    >>> client = Client(cluster)\n\n    See Also\n    --------\n    dask.distributed.Scheduler\n    dask.distributed.Worker\n    asyncssh.connect\n    \"\"\"\n    if set(kwargs) & old_cluster_kwargs:\n        from .old_ssh import SSHCluster as OldSSHCluster\n\n        warnings.warn(\n            \"Note that the SSHCluster API has been replaced.  \"\n            \"We're routing you to the older implementation.  \"\n            \"This will be removed in the future\"\n        )\n        kwargs.setdefault(\"worker_addrs\", hosts)\n        return OldSSHCluster(**kwargs)\n\n    scheduler = {\n        \"cls\": Scheduler,\n        \"options\": {\n            \"address\": hosts[0],\n            \"connect_options\": connect_options,\n            \"kwargs\": scheduler_options,\n            \"remote_python\": remote_python,\n        },\n    }\n    workers = {\n        i: {\n            \"cls\": Worker,\n            \"options\": {\n                \"address\": host,\n                \"connect_options\": connect_options,\n                \"kwargs\": worker_options,\n                \"worker_module\": worker_module,\n                \"remote_python\": remote_python,\n            },\n        }\n        for i, host in enumerate(hosts[1:])\n    }\n    return SpecCluster(workers, scheduler, name=\"SSHCluster\", **kwargs)\n", "path": "distributed/deploy/ssh.py"}]}