We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support for multiple protocols or heterogenous protocols
I have a use case where the `client<=>scheduler` communication needs to be proxied, so we use the [websocket protocol](
https://github.com/dask/distributed/blob/main/distributed/comm/ws.py).

However, doing so means that `scheduler<=>worker` and `worker<=>worker` communication must also use websockets as we assume consistency. 

This results in performance issues on systems with high-performance networking like Infiniband or NVLink. We usually use [UCX](https://github.com/dask/distributed/blob/main/distributed/comm/ucx.py) for `worker<=>worker` communication to get the best out of the hardware.

I wanted to start a discussion around a solution for this. The two things that spring to mind are:
- Add support for multiple protocols simultaneously.
- Allow for one protocol for `client<=>scheduler` and a different one for `worker<=>worker`.

--- 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/cli/dask_scheduler.py`
Content:
```
1 from __future__ import annotations
2 
3 import asyncio
4 import atexit
5 import gc
6 import logging
7 import os
8 import re
9 import sys
10 import warnings
11 
12 import click
13 
14 from distributed import Scheduler
15 from distributed._signals import wait_for_signals
16 from distributed.preloading import validate_preload_argv
17 from distributed.proctitle import (
18     enable_proctitle_on_children,
19     enable_proctitle_on_current,
20 )
21 
22 logger = logging.getLogger("distributed.scheduler")
23 
24 
25 pem_file_option_type = click.Path(exists=True, resolve_path=True)
26 
27 
28 @click.command(context_settings=dict(ignore_unknown_options=True))
29 @click.option("--host", type=str, default="", help="URI, IP or hostname of this server")
30 @click.option("--port", type=int, default=None, help="Serving port")
31 @click.option(
32     "--interface",
33     type=str,
34     default=None,
35     help="Preferred network interface like 'eth0' or 'ib0'",
36 )
37 @click.option(
38     "--protocol", type=str, default=None, help="Protocol like tcp, tls, or ucx"
39 )
40 @click.option(
41     "--tls-ca-file",
42     type=pem_file_option_type,
43     default=None,
44     help="CA cert(s) file for TLS (in PEM format)",
45 )
46 @click.option(
47     "--tls-cert",
48     type=pem_file_option_type,
49     default=None,
50     help="certificate file for TLS (in PEM format)",
51 )
52 @click.option(
53     "--tls-key",
54     type=pem_file_option_type,
55     default=None,
56     help="private key file for TLS (in PEM format)",
57 )
58 # XXX default port (or URI) values should be centralized somewhere
59 @click.option(
60     "--bokeh-port", type=int, default=None, help="Deprecated.  See --dashboard-address"
61 )
62 @click.option(
63     "--dashboard-address",
64     type=str,
65     default=":8787",
66     show_default=True,
67     help="Address on which to listen for diagnostics dashboard",
68 )
69 @click.option(
70     "--dashboard/--no-dashboard",
71     "dashboard",
72     default=True,
73     required=False,
74     help="Launch the Dashboard [default: --dashboard]",
75 )
76 @click.option(
77     "--bokeh/--no-bokeh",
78     "bokeh",
79     default=None,
80     required=False,
81     help="Deprecated.  See --dashboard/--no-dashboard.",
82 )
83 @click.option(
84     "--jupyter/--no-jupyter",
85     "jupyter",
86     default=False,
87     required=False,
88     help="Start a Jupyter Server in the same process.  Warning: This will make"
89     "it possible for anyone with access to your dashboard address to run"
90     "Python code",
91 )
92 @click.option("--show/--no-show", default=False, help="Show web UI [default: --show]")
93 @click.option(
94     "--dashboard-prefix", type=str, default="", help="Prefix for the dashboard app"
95 )
96 @click.option(
97     "--use-xheaders",
98     type=bool,
99     default=False,
100     show_default=True,
101     help="User xheaders in dashboard app for ssl termination in header",
102 )
103 @click.option("--pid-file", type=str, default="", help="File to write the process PID")
104 @click.option(
105     "--scheduler-file",
106     type=str,
107     default="",
108     help="File to write connection information. "
109     "This may be a good way to share connection information if your "
110     "cluster is on a shared network file system.",
111 )
112 @click.option(
113     "--preload",
114     type=str,
115     multiple=True,
116     is_eager=True,
117     help="Module that should be loaded by the scheduler process  "
118     'like "foo.bar" or "/path/to/foo.py".',
119 )
120 @click.argument(
121     "preload_argv", nargs=-1, type=click.UNPROCESSED, callback=validate_preload_argv
122 )
123 @click.option(
124     "--idle-timeout",
125     default=None,
126     type=str,
127     help="Time of inactivity after which to kill the scheduler",
128 )
129 @click.version_option()
130 def main(
131     host,
132     port,
133     bokeh_port,
134     show,
135     dashboard,
136     bokeh,
137     dashboard_prefix,
138     use_xheaders,
139     pid_file,
140     tls_ca_file,
141     tls_cert,
142     tls_key,
143     dashboard_address,
144     jupyter,
145     **kwargs,
146 ):
147     g0, g1, g2 = gc.get_threshold()  # https://github.com/dask/distributed/issues/1653
148     gc.set_threshold(g0 * 3, g1 * 3, g2 * 3)
149 
150     enable_proctitle_on_current()
151     enable_proctitle_on_children()
152 
153     if bokeh_port is not None:
154         warnings.warn(
155             "The --bokeh-port flag has been renamed to --dashboard-address. "
156             "Consider adding ``--dashboard-address :%d`` " % bokeh_port
157         )
158         dashboard_address = bokeh_port
159     if bokeh is not None:
160         warnings.warn(
161             "The --bokeh/--no-bokeh flag has been renamed to --dashboard/--no-dashboard. "
162         )
163         dashboard = bokeh
164 
165     if port is None and (not host or not re.search(r":\d", host)):
166         port = 8786
167 
168     sec = {
169         k: v
170         for k, v in [
171             ("tls_ca_file", tls_ca_file),
172             ("tls_scheduler_cert", tls_cert),
173             ("tls_scheduler_key", tls_key),
174         ]
175         if v is not None
176     }
177 
178     if not host and (tls_ca_file or tls_cert or tls_key):
179         host = "tls://"
180 
181     if pid_file:
182         with open(pid_file, "w") as f:
183             f.write(str(os.getpid()))
184 
185         def del_pid_file():
186             if os.path.exists(pid_file):
187                 os.remove(pid_file)
188 
189         atexit.register(del_pid_file)
190 
191     if sys.platform.startswith("linux"):
192         import resource  # module fails importing on Windows
193 
194         soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
195         limit = max(soft, hard // 2)
196         resource.setrlimit(resource.RLIMIT_NOFILE, (limit, hard))
197 
198     async def run():
199         logger.info("-" * 47)
200 
201         scheduler = Scheduler(
202             security=sec,
203             host=host,
204             port=port,
205             dashboard=dashboard,
206             dashboard_address=dashboard_address,
207             http_prefix=dashboard_prefix,
208             jupyter=jupyter,
209             **kwargs,
210         )
211         logger.info("-" * 47)
212 
213         async def wait_for_scheduler_to_finish():
214             """Wait for the scheduler to initialize and finish"""
215             await scheduler
216             await scheduler.finished()
217 
218         async def wait_for_signals_and_close():
219             """Wait for SIGINT or SIGTERM and close the scheduler upon receiving one of those signals"""
220             await wait_for_signals()
221             await scheduler.close()
222 
223         wait_for_signals_and_close_task = asyncio.create_task(
224             wait_for_signals_and_close()
225         )
226         wait_for_scheduler_to_finish_task = asyncio.create_task(
227             wait_for_scheduler_to_finish()
228         )
229 
230         done, _ = await asyncio.wait(
231             [wait_for_signals_and_close_task, wait_for_scheduler_to_finish_task],
232             return_when=asyncio.FIRST_COMPLETED,
233         )
234         # Re-raise exceptions from done tasks
235         [task.result() for task in done]
236         logger.info("Stopped scheduler at %r", scheduler.address)
237 
238     try:
239         asyncio.run(run())
240     finally:
241         logger.info("End scheduler")
242 
243 
244 if __name__ == "__main__":
245     main()  # pragma: no cover
246
```
--- 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/cli/dask_scheduler.py b/distributed/cli/dask_scheduler.py
--- a/distributed/cli/dask_scheduler.py
+++ b/distributed/cli/dask_scheduler.py
@@ -27,7 +27,7 @@
 
 @click.command(context_settings=dict(ignore_unknown_options=True))
 @click.option("--host", type=str, default="", help="URI, IP or hostname of this server")
-@click.option("--port", type=int, default=None, help="Serving port")
+@click.option("--port", type=str, default=None, help="Serving port")
 @click.option(
     "--interface",
     type=str,
@@ -130,6 +130,8 @@
 def main(
     host,
     port,
+    protocol,
+    interface,
     bokeh_port,
     show,
     dashboard,
@@ -162,8 +164,29 @@
         )
         dashboard = bokeh
 
+    if interface and "," in interface:
+        interface = interface.split(",")
+
+    if protocol and "," in protocol:
+        protocol = protocol.split(",")
+
+    if port:
+        if "," in port:
+            port = [int(p) for p in port.split(",")]
+        else:
+            port = int(port)
+
     if port is None and (not host or not re.search(r":\d", host)):
-        port = 8786
+        if isinstance(protocol, list):
+            port = [8786] + [0] * (len(protocol) - 1)
+        else:
+            port = 8786
+
+    if isinstance(protocol, list) or isinstance(port, list):
+        if (not isinstance(protocol, list) or not isinstance(port, list)) or len(
+            port
+        ) != len(protocol):
+            raise ValueError("--protocol and --port must both be lists of equal length")
 
     sec = {
         k: v
@@ -202,6 +225,8 @@
             security=sec,
             host=host,
             port=port,
+            protocol=protocol,
+            interface=interface,
             dashboard=dashboard,
             dashboard_address=dashboard_address,
             http_prefix=dashboard_prefix,

{"golden_diff": "diff --git a/distributed/cli/dask_scheduler.py b/distributed/cli/dask_scheduler.py\n--- a/distributed/cli/dask_scheduler.py\n+++ b/distributed/cli/dask_scheduler.py\n@@ -27,7 +27,7 @@\n \n @click.command(context_settings=dict(ignore_unknown_options=True))\n @click.option(\"--host\", type=str, default=\"\", help=\"URI, IP or hostname of this server\")\n-@click.option(\"--port\", type=int, default=None, help=\"Serving port\")\n+@click.option(\"--port\", type=str, default=None, help=\"Serving port\")\n @click.option(\n     \"--interface\",\n     type=str,\n@@ -130,6 +130,8 @@\n def main(\n     host,\n     port,\n+    protocol,\n+    interface,\n     bokeh_port,\n     show,\n     dashboard,\n@@ -162,8 +164,29 @@\n         )\n         dashboard = bokeh\n \n+    if interface and \",\" in interface:\n+        interface = interface.split(\",\")\n+\n+    if protocol and \",\" in protocol:\n+        protocol = protocol.split(\",\")\n+\n+    if port:\n+        if \",\" in port:\n+            port = [int(p) for p in port.split(\",\")]\n+        else:\n+            port = int(port)\n+\n     if port is None and (not host or not re.search(r\":\\d\", host)):\n-        port = 8786\n+        if isinstance(protocol, list):\n+            port = [8786] + [0] * (len(protocol) - 1)\n+        else:\n+            port = 8786\n+\n+    if isinstance(protocol, list) or isinstance(port, list):\n+        if (not isinstance(protocol, list) or not isinstance(port, list)) or len(\n+            port\n+        ) != len(protocol):\n+            raise ValueError(\"--protocol and --port must both be lists of equal length\")\n \n     sec = {\n         k: v\n@@ -202,6 +225,8 @@\n             security=sec,\n             host=host,\n             port=port,\n+            protocol=protocol,\n+            interface=interface,\n             dashboard=dashboard,\n             dashboard_address=dashboard_address,\n             http_prefix=dashboard_prefix,\n", "issue": "Support for multiple protocols or heterogenous protocols\nI have a use case where the `client<=>scheduler` communication needs to be proxied, so we use the [websocket protocol](\r\nhttps://github.com/dask/distributed/blob/main/distributed/comm/ws.py).\r\n\r\nHowever, doing so means that `scheduler<=>worker` and `worker<=>worker` communication must also use websockets as we assume consistency. \r\n\r\nThis results in performance issues on systems with high-performance networking like Infiniband or NVLink. We usually use [UCX](https://github.com/dask/distributed/blob/main/distributed/comm/ucx.py) for `worker<=>worker` communication to get the best out of the hardware.\r\n\r\nI wanted to start a discussion around a solution for this. The two things that spring to mind are:\r\n- Add support for multiple protocols simultaneously.\r\n- Allow for one protocol for `client<=>scheduler` and a different one for `worker<=>worker`.\n", "before_files": [{"content": "from __future__ import annotations\n\nimport asyncio\nimport atexit\nimport gc\nimport logging\nimport os\nimport re\nimport sys\nimport warnings\n\nimport click\n\nfrom distributed import Scheduler\nfrom distributed._signals import wait_for_signals\nfrom distributed.preloading import validate_preload_argv\nfrom distributed.proctitle import (\n    enable_proctitle_on_children,\n    enable_proctitle_on_current,\n)\n\nlogger = logging.getLogger(\"distributed.scheduler\")\n\n\npem_file_option_type = click.Path(exists=True, resolve_path=True)\n\n\n@click.command(context_settings=dict(ignore_unknown_options=True))\n@click.option(\"--host\", type=str, default=\"\", help=\"URI, IP or hostname of this server\")\n@click.option(\"--port\", type=int, default=None, help=\"Serving port\")\n@click.option(\n    \"--interface\",\n    type=str,\n    default=None,\n    help=\"Preferred network interface like 'eth0' or 'ib0'\",\n)\n@click.option(\n    \"--protocol\", type=str, default=None, help=\"Protocol like tcp, tls, or ucx\"\n)\n@click.option(\n    \"--tls-ca-file\",\n    type=pem_file_option_type,\n    default=None,\n    help=\"CA cert(s) file for TLS (in PEM format)\",\n)\n@click.option(\n    \"--tls-cert\",\n    type=pem_file_option_type,\n    default=None,\n    help=\"certificate file for TLS (in PEM format)\",\n)\n@click.option(\n    \"--tls-key\",\n    type=pem_file_option_type,\n    default=None,\n    help=\"private key file for TLS (in PEM format)\",\n)\n# XXX default port (or URI) values should be centralized somewhere\n@click.option(\n    \"--bokeh-port\", type=int, default=None, help=\"Deprecated.  See --dashboard-address\"\n)\n@click.option(\n    \"--dashboard-address\",\n    type=str,\n    default=\":8787\",\n    show_default=True,\n    help=\"Address on which to listen for diagnostics dashboard\",\n)\n@click.option(\n    \"--dashboard/--no-dashboard\",\n    \"dashboard\",\n    default=True,\n    required=False,\n    help=\"Launch the Dashboard [default: --dashboard]\",\n)\n@click.option(\n    \"--bokeh/--no-bokeh\",\n    \"bokeh\",\n    default=None,\n    required=False,\n    help=\"Deprecated.  See --dashboard/--no-dashboard.\",\n)\n@click.option(\n    \"--jupyter/--no-jupyter\",\n    \"jupyter\",\n    default=False,\n    required=False,\n    help=\"Start a Jupyter Server in the same process.  Warning: This will make\"\n    \"it possible for anyone with access to your dashboard address to run\"\n    \"Python code\",\n)\n@click.option(\"--show/--no-show\", default=False, help=\"Show web UI [default: --show]\")\n@click.option(\n    \"--dashboard-prefix\", type=str, default=\"\", help=\"Prefix for the dashboard app\"\n)\n@click.option(\n    \"--use-xheaders\",\n    type=bool,\n    default=False,\n    show_default=True,\n    help=\"User xheaders in dashboard app for ssl termination in header\",\n)\n@click.option(\"--pid-file\", type=str, default=\"\", help=\"File to write the process PID\")\n@click.option(\n    \"--scheduler-file\",\n    type=str,\n    default=\"\",\n    help=\"File to write connection information. \"\n    \"This may be a good way to share connection information if your \"\n    \"cluster is on a shared network file system.\",\n)\n@click.option(\n    \"--preload\",\n    type=str,\n    multiple=True,\n    is_eager=True,\n    help=\"Module that should be loaded by the scheduler process  \"\n    'like \"foo.bar\" or \"/path/to/foo.py\".',\n)\n@click.argument(\n    \"preload_argv\", nargs=-1, type=click.UNPROCESSED, callback=validate_preload_argv\n)\n@click.option(\n    \"--idle-timeout\",\n    default=None,\n    type=str,\n    help=\"Time of inactivity after which to kill the scheduler\",\n)\n@click.version_option()\ndef main(\n    host,\n    port,\n    bokeh_port,\n    show,\n    dashboard,\n    bokeh,\n    dashboard_prefix,\n    use_xheaders,\n    pid_file,\n    tls_ca_file,\n    tls_cert,\n    tls_key,\n    dashboard_address,\n    jupyter,\n    **kwargs,\n):\n    g0, g1, g2 = gc.get_threshold()  # https://github.com/dask/distributed/issues/1653\n    gc.set_threshold(g0 * 3, g1 * 3, g2 * 3)\n\n    enable_proctitle_on_current()\n    enable_proctitle_on_children()\n\n    if bokeh_port is not None:\n        warnings.warn(\n            \"The --bokeh-port flag has been renamed to --dashboard-address. \"\n            \"Consider adding ``--dashboard-address :%d`` \" % bokeh_port\n        )\n        dashboard_address = bokeh_port\n    if bokeh is not None:\n        warnings.warn(\n            \"The --bokeh/--no-bokeh flag has been renamed to --dashboard/--no-dashboard. \"\n        )\n        dashboard = bokeh\n\n    if port is None and (not host or not re.search(r\":\\d\", host)):\n        port = 8786\n\n    sec = {\n        k: v\n        for k, v in [\n            (\"tls_ca_file\", tls_ca_file),\n            (\"tls_scheduler_cert\", tls_cert),\n            (\"tls_scheduler_key\", tls_key),\n        ]\n        if v is not None\n    }\n\n    if not host and (tls_ca_file or tls_cert or tls_key):\n        host = \"tls://\"\n\n    if pid_file:\n        with open(pid_file, \"w\") as f:\n            f.write(str(os.getpid()))\n\n        def del_pid_file():\n            if os.path.exists(pid_file):\n                os.remove(pid_file)\n\n        atexit.register(del_pid_file)\n\n    if sys.platform.startswith(\"linux\"):\n        import resource  # module fails importing on Windows\n\n        soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)\n        limit = max(soft, hard // 2)\n        resource.setrlimit(resource.RLIMIT_NOFILE, (limit, hard))\n\n    async def run():\n        logger.info(\"-\" * 47)\n\n        scheduler = Scheduler(\n            security=sec,\n            host=host,\n            port=port,\n            dashboard=dashboard,\n            dashboard_address=dashboard_address,\n            http_prefix=dashboard_prefix,\n            jupyter=jupyter,\n            **kwargs,\n        )\n        logger.info(\"-\" * 47)\n\n        async def wait_for_scheduler_to_finish():\n            \"\"\"Wait for the scheduler to initialize and finish\"\"\"\n            await scheduler\n            await scheduler.finished()\n\n        async def wait_for_signals_and_close():\n            \"\"\"Wait for SIGINT or SIGTERM and close the scheduler upon receiving one of those signals\"\"\"\n            await wait_for_signals()\n            await scheduler.close()\n\n        wait_for_signals_and_close_task = asyncio.create_task(\n            wait_for_signals_and_close()\n        )\n        wait_for_scheduler_to_finish_task = asyncio.create_task(\n            wait_for_scheduler_to_finish()\n        )\n\n        done, _ = await asyncio.wait(\n            [wait_for_signals_and_close_task, wait_for_scheduler_to_finish_task],\n            return_when=asyncio.FIRST_COMPLETED,\n        )\n        # Re-raise exceptions from done tasks\n        [task.result() for task in done]\n        logger.info(\"Stopped scheduler at %r\", scheduler.address)\n\n    try:\n        asyncio.run(run())\n    finally:\n        logger.info(\"End scheduler\")\n\n\nif __name__ == \"__main__\":\n    main()  # pragma: no cover\n", "path": "distributed/cli/dask_scheduler.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport asyncio\nimport atexit\nimport gc\nimport logging\nimport os\nimport re\nimport sys\nimport warnings\n\nimport click\n\nfrom distributed import Scheduler\nfrom distributed._signals import wait_for_signals\nfrom distributed.preloading import validate_preload_argv\nfrom distributed.proctitle import (\n    enable_proctitle_on_children,\n    enable_proctitle_on_current,\n)\n\nlogger = logging.getLogger(\"distributed.scheduler\")\n\n\npem_file_option_type = click.Path(exists=True, resolve_path=True)\n\n\n@click.command(context_settings=dict(ignore_unknown_options=True))\n@click.option(\"--host\", type=str, default=\"\", help=\"URI, IP or hostname of this server\")\n@click.option(\"--port\", type=str, default=None, help=\"Serving port\")\n@click.option(\n    \"--interface\",\n    type=str,\n    default=None,\n    help=\"Preferred network interface like 'eth0' or 'ib0'\",\n)\n@click.option(\n    \"--protocol\", type=str, default=None, help=\"Protocol like tcp, tls, or ucx\"\n)\n@click.option(\n    \"--tls-ca-file\",\n    type=pem_file_option_type,\n    default=None,\n    help=\"CA cert(s) file for TLS (in PEM format)\",\n)\n@click.option(\n    \"--tls-cert\",\n    type=pem_file_option_type,\n    default=None,\n    help=\"certificate file for TLS (in PEM format)\",\n)\n@click.option(\n    \"--tls-key\",\n    type=pem_file_option_type,\n    default=None,\n    help=\"private key file for TLS (in PEM format)\",\n)\n# XXX default port (or URI) values should be centralized somewhere\n@click.option(\n    \"--bokeh-port\", type=int, default=None, help=\"Deprecated.  See --dashboard-address\"\n)\n@click.option(\n    \"--dashboard-address\",\n    type=str,\n    default=\":8787\",\n    show_default=True,\n    help=\"Address on which to listen for diagnostics dashboard\",\n)\n@click.option(\n    \"--dashboard/--no-dashboard\",\n    \"dashboard\",\n    default=True,\n    required=False,\n    help=\"Launch the Dashboard [default: --dashboard]\",\n)\n@click.option(\n    \"--bokeh/--no-bokeh\",\n    \"bokeh\",\n    default=None,\n    required=False,\n    help=\"Deprecated.  See --dashboard/--no-dashboard.\",\n)\n@click.option(\n    \"--jupyter/--no-jupyter\",\n    \"jupyter\",\n    default=False,\n    required=False,\n    help=\"Start a Jupyter Server in the same process.  Warning: This will make\"\n    \"it possible for anyone with access to your dashboard address to run\"\n    \"Python code\",\n)\n@click.option(\"--show/--no-show\", default=False, help=\"Show web UI [default: --show]\")\n@click.option(\n    \"--dashboard-prefix\", type=str, default=\"\", help=\"Prefix for the dashboard app\"\n)\n@click.option(\n    \"--use-xheaders\",\n    type=bool,\n    default=False,\n    show_default=True,\n    help=\"User xheaders in dashboard app for ssl termination in header\",\n)\n@click.option(\"--pid-file\", type=str, default=\"\", help=\"File to write the process PID\")\n@click.option(\n    \"--scheduler-file\",\n    type=str,\n    default=\"\",\n    help=\"File to write connection information. \"\n    \"This may be a good way to share connection information if your \"\n    \"cluster is on a shared network file system.\",\n)\n@click.option(\n    \"--preload\",\n    type=str,\n    multiple=True,\n    is_eager=True,\n    help=\"Module that should be loaded by the scheduler process  \"\n    'like \"foo.bar\" or \"/path/to/foo.py\".',\n)\n@click.argument(\n    \"preload_argv\", nargs=-1, type=click.UNPROCESSED, callback=validate_preload_argv\n)\n@click.option(\n    \"--idle-timeout\",\n    default=None,\n    type=str,\n    help=\"Time of inactivity after which to kill the scheduler\",\n)\n@click.version_option()\ndef main(\n    host,\n    port,\n    protocol,\n    interface,\n    bokeh_port,\n    show,\n    dashboard,\n    bokeh,\n    dashboard_prefix,\n    use_xheaders,\n    pid_file,\n    tls_ca_file,\n    tls_cert,\n    tls_key,\n    dashboard_address,\n    jupyter,\n    **kwargs,\n):\n    g0, g1, g2 = gc.get_threshold()  # https://github.com/dask/distributed/issues/1653\n    gc.set_threshold(g0 * 3, g1 * 3, g2 * 3)\n\n    enable_proctitle_on_current()\n    enable_proctitle_on_children()\n\n    if bokeh_port is not None:\n        warnings.warn(\n            \"The --bokeh-port flag has been renamed to --dashboard-address. \"\n            \"Consider adding ``--dashboard-address :%d`` \" % bokeh_port\n        )\n        dashboard_address = bokeh_port\n    if bokeh is not None:\n        warnings.warn(\n            \"The --bokeh/--no-bokeh flag has been renamed to --dashboard/--no-dashboard. \"\n        )\n        dashboard = bokeh\n\n    if interface and \",\" in interface:\n        interface = interface.split(\",\")\n\n    if protocol and \",\" in protocol:\n        protocol = protocol.split(\",\")\n\n    if port:\n        if \",\" in port:\n            port = [int(p) for p in port.split(\",\")]\n        else:\n            port = int(port)\n\n    if port is None and (not host or not re.search(r\":\\d\", host)):\n        if isinstance(protocol, list):\n            port = [8786] + [0] * (len(protocol) - 1)\n        else:\n            port = 8786\n\n    if isinstance(protocol, list) or isinstance(port, list):\n        if (not isinstance(protocol, list) or not isinstance(port, list)) or len(\n            port\n        ) != len(protocol):\n            raise ValueError(\"--protocol and --port must both be lists of equal length\")\n\n    sec = {\n        k: v\n        for k, v in [\n            (\"tls_ca_file\", tls_ca_file),\n            (\"tls_scheduler_cert\", tls_cert),\n            (\"tls_scheduler_key\", tls_key),\n        ]\n        if v is not None\n    }\n\n    if not host and (tls_ca_file or tls_cert or tls_key):\n        host = \"tls://\"\n\n    if pid_file:\n        with open(pid_file, \"w\") as f:\n            f.write(str(os.getpid()))\n\n        def del_pid_file():\n            if os.path.exists(pid_file):\n                os.remove(pid_file)\n\n        atexit.register(del_pid_file)\n\n    if sys.platform.startswith(\"linux\"):\n        import resource  # module fails importing on Windows\n\n        soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)\n        limit = max(soft, hard // 2)\n        resource.setrlimit(resource.RLIMIT_NOFILE, (limit, hard))\n\n    async def run():\n        logger.info(\"-\" * 47)\n\n        scheduler = Scheduler(\n            security=sec,\n            host=host,\n            port=port,\n            protocol=protocol,\n            interface=interface,\n            dashboard=dashboard,\n            dashboard_address=dashboard_address,\n            http_prefix=dashboard_prefix,\n            jupyter=jupyter,\n            **kwargs,\n        )\n        logger.info(\"-\" * 47)\n\n        async def wait_for_scheduler_to_finish():\n            \"\"\"Wait for the scheduler to initialize and finish\"\"\"\n            await scheduler\n            await scheduler.finished()\n\n        async def wait_for_signals_and_close():\n            \"\"\"Wait for SIGINT or SIGTERM and close the scheduler upon receiving one of those signals\"\"\"\n            await wait_for_signals()\n            await scheduler.close()\n\n        wait_for_signals_and_close_task = asyncio.create_task(\n            wait_for_signals_and_close()\n        )\n        wait_for_scheduler_to_finish_task = asyncio.create_task(\n            wait_for_scheduler_to_finish()\n        )\n\n        done, _ = await asyncio.wait(\n            [wait_for_signals_and_close_task, wait_for_scheduler_to_finish_task],\n            return_when=asyncio.FIRST_COMPLETED,\n        )\n        # Re-raise exceptions from done tasks\n        [task.result() for task in done]\n        logger.info(\"Stopped scheduler at %r\", scheduler.address)\n\n    try:\n        asyncio.run(run())\n    finally:\n        logger.info(\"End scheduler\")\n\n\nif __name__ == \"__main__\":\n    main()  # pragma: no cover\n", "path": "distributed/cli/dask_scheduler.py"}]}