We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support different usernames when starting an SSH cluster
I wanted to start an SSH cluster where my username is not same on all the nodes.  It would be great if this was supported.  The most obvious way to configure this would be to allow specifying hosts as `user@hostname`, either on the command line, or the hosts file.

--- 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                 self.remote_python or 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             [
190                 set_env,
191                 self.remote_python or sys.executable,
192                 "-m",
193                 "distributed.cli.dask_scheduler",
194             ]
195             + cli_keywords(self.kwargs, cls=_Scheduler)
196         )
197         self.proc = await self.connection.create_process(cmd)
198 
199         # We watch stderr in order to get the address, then we return
200         while True:
201             line = await self.proc.stderr.readline()
202             if not line.strip():
203                 raise Exception("Worker failed to start")
204             logger.info(line.strip())
205             if "Scheduler at" in line:
206                 self.address = line.split("Scheduler at:")[1].strip()
207                 break
208         logger.debug("%s", line)
209         await super().start()
210 
211 
212 old_cluster_kwargs = {
213     "scheduler_addr",
214     "scheduler_port",
215     "worker_addrs",
216     "nthreads",
217     "nprocs",
218     "ssh_username",
219     "ssh_port",
220     "ssh_private_key",
221     "nohost",
222     "logdir",
223     "remote_python",
224     "memory_limit",
225     "worker_port",
226     "nanny_port",
227     "remote_dask_worker",
228 }
229 
230 
231 def SSHCluster(
232     hosts: List[str] = None,
233     connect_options: dict = {},
234     worker_options: dict = {},
235     scheduler_options: dict = {},
236     worker_module: str = "distributed.cli.dask_worker",
237     remote_python: str = None,
238     **kwargs
239 ):
240     """ Deploy a Dask cluster using SSH
241 
242     The SSHCluster function deploys a Dask Scheduler and Workers for you on a
243     set of machine addresses that you provide.  The first address will be used
244     for the scheduler while the rest will be used for the workers (feel free to
245     repeat the first hostname if you want to have the scheduler and worker
246     co-habitate one machine.)
247 
248     You may configure the scheduler and workers by passing
249     ``scheduler_options`` and ``worker_options`` dictionary keywords.  See the
250     ``dask.distributed.Scheduler`` and ``dask.distributed.Worker`` classes for
251     details on the available options, but the defaults should work in most
252     situations.
253 
254     You may configure your use of SSH itself using the ``connect_options``
255     keyword, which passes values to the ``asyncssh.connect`` function.  For
256     more information on these see the documentation for the ``asyncssh``
257     library https://asyncssh.readthedocs.io .
258 
259     Parameters
260     ----------
261     hosts: List[str]
262         List of hostnames or addresses on which to launch our cluster.
263         The first will be used for the scheduler and the rest for workers.
264     connect_options: dict, optional
265         Keywords to pass through to ``asyncssh.connect``.
266     worker_options: dict, optional
267         Keywords to pass on to workers.
268     scheduler_options: dict, optional
269         Keywords to pass on to scheduler.
270     worker_module: str, optional
271         Python module to call to start the worker.
272     remote_python: str, optional
273         Path to Python on remote nodes.
274 
275     Examples
276     --------
277     >>> from dask.distributed import Client, SSHCluster
278     >>> cluster = SSHCluster(
279     ...     ["localhost", "localhost", "localhost", "localhost"],
280     ...     connect_options={"known_hosts": None},
281     ...     worker_options={"nthreads": 2},
282     ...     scheduler_options={"port": 0, "dashboard_address": ":8797"}
283     ... )
284     >>> client = Client(cluster)
285 
286     An example using a different worker module, in particular the
287     ``dask-cuda-worker`` command from the ``dask-cuda`` project.
288 
289     >>> from dask.distributed import Client, SSHCluster
290     >>> cluster = SSHCluster(
291     ...     ["localhost", "hostwithgpus", "anothergpuhost"],
292     ...     connect_options={"known_hosts": None},
293     ...     scheduler_options={"port": 0, "dashboard_address": ":8797"},
294     ...     worker_module='dask_cuda.dask_cuda_worker')
295     >>> client = Client(cluster)
296 
297     See Also
298     --------
299     dask.distributed.Scheduler
300     dask.distributed.Worker
301     asyncssh.connect
302     """
303     if set(kwargs) & old_cluster_kwargs:
304         from .old_ssh import SSHCluster as OldSSHCluster
305 
306         warnings.warn(
307             "Note that the SSHCluster API has been replaced.  "
308             "We're routing you to the older implementation.  "
309             "This will be removed in the future"
310         )
311         kwargs.setdefault("worker_addrs", hosts)
312         return OldSSHCluster(**kwargs)
313 
314     scheduler = {
315         "cls": Scheduler,
316         "options": {
317             "address": hosts[0],
318             "connect_options": connect_options,
319             "kwargs": scheduler_options,
320             "remote_python": remote_python,
321         },
322     }
323     workers = {
324         i: {
325             "cls": Worker,
326             "options": {
327                 "address": host,
328                 "connect_options": connect_options,
329                 "kwargs": worker_options,
330                 "worker_module": worker_module,
331                 "remote_python": remote_python,
332             },
333         }
334         for i, host in enumerate(hosts[1:])
335     }
336     return SpecCluster(workers, scheduler, name="SSHCluster", **kwargs)
337
```
--- 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
@@ -1,6 +1,6 @@
 import logging
 import sys
-from typing import List
+from typing import List, Union
 import warnings
 import weakref
 
@@ -230,7 +230,7 @@
 
 def SSHCluster(
     hosts: List[str] = None,
-    connect_options: dict = {},
+    connect_options: Union[List[dict], dict] = {},
     worker_options: dict = {},
     scheduler_options: dict = {},
     worker_module: str = "distributed.cli.dask_worker",
@@ -261,8 +261,9 @@
     hosts: List[str]
         List of hostnames or addresses on which to launch our cluster.
         The first will be used for the scheduler and the rest for workers.
-    connect_options: dict, optional
+    connect_options: dict or list of dict, optional
         Keywords to pass through to ``asyncssh.connect``.
+        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
@@ -311,11 +312,19 @@
         kwargs.setdefault("worker_addrs", hosts)
         return OldSSHCluster(**kwargs)
 
+    if isinstance(connect_options, list) and len(connect_options) != len(hosts):
+        raise RuntimeError(
+            "When specifying a list of connect_options you must provide a "
+            "dictionary for each address."
+        )
+
     scheduler = {
         "cls": Scheduler,
         "options": {
             "address": hosts[0],
-            "connect_options": connect_options,
+            "connect_options": connect_options
+            if isinstance(connect_options, dict)
+            else connect_options[0],
             "kwargs": scheduler_options,
             "remote_python": remote_python,
         },
@@ -325,7 +334,9 @@
             "cls": Worker,
             "options": {
                 "address": host,
-                "connect_options": connect_options,
+                "connect_options": connect_options
+                if isinstance(connect_options, dict)
+                else connect_options[i + 1],
                 "kwargs": worker_options,
                 "worker_module": worker_module,
                 "remote_python": remote_python,

{"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@@ -1,6 +1,6 @@\n import logging\n import sys\n-from typing import List\n+from typing import List, Union\n import warnings\n import weakref\n \n@@ -230,7 +230,7 @@\n \n def SSHCluster(\n     hosts: List[str] = None,\n-    connect_options: dict = {},\n+    connect_options: Union[List[dict], dict] = {},\n     worker_options: dict = {},\n     scheduler_options: dict = {},\n     worker_module: str = \"distributed.cli.dask_worker\",\n@@ -261,8 +261,9 @@\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+    connect_options: dict or list of dict, optional\n         Keywords to pass through to ``asyncssh.connect``.\n+        If a list it must have the same length as ``hosts``.\n     worker_options: dict, optional\n         Keywords to pass on to workers.\n     scheduler_options: dict, optional\n@@ -311,11 +312,19 @@\n         kwargs.setdefault(\"worker_addrs\", hosts)\n         return OldSSHCluster(**kwargs)\n \n+    if isinstance(connect_options, list) and len(connect_options) != len(hosts):\n+        raise RuntimeError(\n+            \"When specifying a list of connect_options you must provide a \"\n+            \"dictionary for each address.\"\n+        )\n+\n     scheduler = {\n         \"cls\": Scheduler,\n         \"options\": {\n             \"address\": hosts[0],\n-            \"connect_options\": connect_options,\n+            \"connect_options\": connect_options\n+            if isinstance(connect_options, dict)\n+            else connect_options[0],\n             \"kwargs\": scheduler_options,\n             \"remote_python\": remote_python,\n         },\n@@ -325,7 +334,9 @@\n             \"cls\": Worker,\n             \"options\": {\n                 \"address\": host,\n-                \"connect_options\": connect_options,\n+                \"connect_options\": connect_options\n+                if isinstance(connect_options, dict)\n+                else connect_options[i + 1],\n                 \"kwargs\": worker_options,\n                 \"worker_module\": worker_module,\n                 \"remote_python\": remote_python,\n", "issue": "Support different usernames when starting an SSH cluster\nI wanted to start an SSH cluster where my username is not same on all the nodes.  It would be great if this was supported.  The most obvious way to configure this would be to allow specifying hosts as `user@hostname`, either on the command line, or the hosts file.\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                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"}], "after_files": [{"content": "import logging\nimport sys\nfrom typing import List, Union\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: Union[List[dict], 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 or list of dict, optional\n        Keywords to pass through to ``asyncssh.connect``.\n        If a list it must have the same length as ``hosts``.\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    if isinstance(connect_options, list) and len(connect_options) != len(hosts):\n        raise RuntimeError(\n            \"When specifying a list of connect_options you must provide a \"\n            \"dictionary for each address.\"\n        )\n\n    scheduler = {\n        \"cls\": Scheduler,\n        \"options\": {\n            \"address\": hosts[0],\n            \"connect_options\": connect_options\n            if isinstance(connect_options, dict)\n            else connect_options[0],\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                if isinstance(connect_options, dict)\n                else connect_options[i + 1],\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"}]}