We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Scheduler timeout fails to exit `dask-scheduler` process
The scheduler timeout introduced in #2652 seems to successfully close the scheduler comms, however the process doesn't exit.

I would expect the process to also exit when the scheduler has closed.

```console
$ pip install git+https://github.com/dask/distributed.git

$ export DASK_DISTRIBUTED__SCHEDULER__IDLE_TIMEOUT='200 ms'

$ python -c 'import distributed; print(distributed.__version__)' 
2.1.0+4.ge5ec8daa 

$ dask-scheduler  
distributed.scheduler - INFO - ----------------------------------------------- 
distributed.scheduler - INFO - Clear task state 
distributed.dashboard.proxy - INFO - To route to workers diagnostics web server please install jupyter-server-proxy: pip install jupyter-server-proxy
distributed.scheduler - INFO -   Scheduler at:     tcp://172.17.0.2:8786 
distributed.scheduler - INFO -   dashboard at:                     :8787
distributed.scheduler - INFO - Local Directory:    /tmp/scheduler-gx0ramly
distributed.scheduler - INFO - -----------------------------------------------
distributed.scheduler - INFO - Scheduler closing... 
distributed.scheduler - INFO - Scheduler closing all comms


... (hit enter a bunch of times and waited a while here before hitting CTRL+c)...


^Cdistributed.scheduler - INFO - End scheduler at 'tcp://172.17.0.2:8786' 
```

The scheduler should exit cleanly to satisfy the requirements in #2346.

--- 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 print_function, division, absolute_import
2 
3 import atexit
4 import dask
5 import logging
6 import gc
7 import os
8 import re
9 import shutil
10 import sys
11 import tempfile
12 import warnings
13 
14 import click
15 
16 from tornado.ioloop import IOLoop
17 
18 from distributed import Scheduler
19 from distributed.security import Security
20 from distributed.cli.utils import check_python_3, install_signal_handlers
21 from distributed.preloading import preload_modules, validate_preload_argv
22 from distributed.proctitle import (
23     enable_proctitle_on_children,
24     enable_proctitle_on_current,
25 )
26 
27 logger = logging.getLogger("distributed.scheduler")
28 
29 
30 pem_file_option_type = click.Path(exists=True, resolve_path=True)
31 
32 
33 @click.command(context_settings=dict(ignore_unknown_options=True))
34 @click.option("--host", type=str, default="", help="URI, IP or hostname of this server")
35 @click.option("--port", type=int, default=None, help="Serving port")
36 @click.option(
37     "--interface",
38     type=str,
39     default=None,
40     help="Preferred network interface like 'eth0' or 'ib0'",
41 )
42 @click.option(
43     "--protocol", type=str, default=None, help="Protocol like tcp, tls, or ucx"
44 )
45 @click.option(
46     "--tls-ca-file",
47     type=pem_file_option_type,
48     default=None,
49     help="CA cert(s) file for TLS (in PEM format)",
50 )
51 @click.option(
52     "--tls-cert",
53     type=pem_file_option_type,
54     default=None,
55     help="certificate file for TLS (in PEM format)",
56 )
57 @click.option(
58     "--tls-key",
59     type=pem_file_option_type,
60     default=None,
61     help="private key file for TLS (in PEM format)",
62 )
63 # XXX default port (or URI) values should be centralized somewhere
64 @click.option(
65     "--bokeh-port", type=int, default=None, help="Deprecated.  See --dashboard-address"
66 )
67 @click.option(
68     "--dashboard-address",
69     type=str,
70     default=":8787",
71     help="Address on which to listen for diagnostics dashboard",
72 )
73 @click.option(
74     "--dashboard/--no-dashboard",
75     "dashboard",
76     default=True,
77     show_default=True,
78     required=False,
79     help="Launch the Dashboard",
80 )
81 @click.option(
82     "--bokeh/--no-bokeh",
83     "bokeh",
84     default=None,
85     required=False,
86     help="Deprecated.  See --dashboard/--no-dashboard.",
87 )
88 @click.option("--show/--no-show", default=False, help="Show web UI")
89 @click.option(
90     "--dashboard-prefix", type=str, default=None, help="Prefix for the dashboard app"
91 )
92 @click.option(
93     "--use-xheaders",
94     type=bool,
95     default=False,
96     show_default=True,
97     help="User xheaders in dashboard app for ssl termination in header",
98 )
99 @click.option("--pid-file", type=str, default="", help="File to write the process PID")
100 @click.option(
101     "--scheduler-file",
102     type=str,
103     default="",
104     help="File to write connection information. "
105     "This may be a good way to share connection information if your "
106     "cluster is on a shared network file system.",
107 )
108 @click.option(
109     "--local-directory", default="", type=str, help="Directory to place scheduler files"
110 )
111 @click.option(
112     "--preload",
113     type=str,
114     multiple=True,
115     is_eager=True,
116     default="",
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.version_option()
124 def main(
125     host,
126     port,
127     bokeh_port,
128     show,
129     dashboard,
130     bokeh,
131     dashboard_prefix,
132     use_xheaders,
133     pid_file,
134     scheduler_file,
135     interface,
136     protocol,
137     local_directory,
138     preload,
139     preload_argv,
140     tls_ca_file,
141     tls_cert,
142     tls_key,
143     dashboard_address,
144 ):
145     g0, g1, g2 = gc.get_threshold()  # https://github.com/dask/distributed/issues/1653
146     gc.set_threshold(g0 * 3, g1 * 3, g2 * 3)
147 
148     enable_proctitle_on_current()
149     enable_proctitle_on_children()
150 
151     if bokeh_port is not None:
152         warnings.warn(
153             "The --bokeh-port flag has been renamed to --dashboard-address. "
154             "Consider adding ``--dashboard-address :%d`` " % bokeh_port
155         )
156         dashboard_address = bokeh_port
157     if bokeh is not None:
158         warnings.warn(
159             "The --bokeh/--no-bokeh flag has been renamed to --dashboard/--no-dashboard. "
160         )
161         dashboard = bokeh
162 
163     if port is None and (not host or not re.search(r":\d", host)):
164         port = 8786
165 
166     sec = Security(
167         tls_ca_file=tls_ca_file, tls_scheduler_cert=tls_cert, tls_scheduler_key=tls_key
168     )
169 
170     if not host and (tls_ca_file or tls_cert or tls_key):
171         host = "tls://"
172 
173     if pid_file:
174         with open(pid_file, "w") as f:
175             f.write(str(os.getpid()))
176 
177         def del_pid_file():
178             if os.path.exists(pid_file):
179                 os.remove(pid_file)
180 
181         atexit.register(del_pid_file)
182 
183     local_directory_created = False
184     if local_directory:
185         if not os.path.exists(local_directory):
186             os.mkdir(local_directory)
187             local_directory_created = True
188     else:
189         local_directory = tempfile.mkdtemp(prefix="scheduler-")
190         local_directory_created = True
191     if local_directory not in sys.path:
192         sys.path.insert(0, local_directory)
193 
194     if sys.platform.startswith("linux"):
195         import resource  # module fails importing on Windows
196 
197         soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
198         limit = max(soft, hard // 2)
199         resource.setrlimit(resource.RLIMIT_NOFILE, (limit, hard))
200 
201     loop = IOLoop.current()
202     logger.info("-" * 47)
203 
204     scheduler = Scheduler(
205         loop=loop,
206         scheduler_file=scheduler_file,
207         security=sec,
208         host=host,
209         port=port,
210         interface=interface,
211         protocol=protocol,
212         dashboard_address=dashboard_address if dashboard else None,
213         service_kwargs={"dashboard": {"prefix": dashboard_prefix}},
214     )
215     scheduler.start()
216     if not preload:
217         preload = dask.config.get("distributed.scheduler.preload")
218     if not preload_argv:
219         preload_argv = dask.config.get("distributed.scheduler.preload-argv")
220     preload_modules(
221         preload, parameter=scheduler, file_dir=local_directory, argv=preload_argv
222     )
223 
224     logger.info("Local Directory: %26s", local_directory)
225     logger.info("-" * 47)
226 
227     install_signal_handlers(loop)
228 
229     try:
230         loop.start()
231         loop.close()
232     finally:
233         scheduler.stop()
234         if local_directory_created:
235             shutil.rmtree(local_directory)
236 
237         logger.info("End scheduler at %r", scheduler.address)
238 
239 
240 def go():
241     check_python_3()
242     main()
243 
244 
245 if __name__ == "__main__":
246     go()
247
```
--- 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
@@ -120,6 +120,12 @@
 @click.argument(
     "preload_argv", nargs=-1, type=click.UNPROCESSED, callback=validate_preload_argv
 )
+@click.option(
+    "--idle-timeout",
+    default=None,
+    type=str,
+    help="Time of inactivity after which to kill the scheduler",
+)
 @click.version_option()
 def main(
     host,
@@ -141,6 +147,7 @@
     tls_cert,
     tls_key,
     dashboard_address,
+    idle_timeout,
 ):
     g0, g1, g2 = gc.get_threshold()  # https://github.com/dask/distributed/issues/1653
     gc.set_threshold(g0 * 3, g1 * 3, g2 * 3)
@@ -211,6 +218,7 @@
         protocol=protocol,
         dashboard_address=dashboard_address if dashboard else None,
         service_kwargs={"dashboard": {"prefix": dashboard_prefix}},
+        idle_timeout=idle_timeout,
     )
     scheduler.start()
     if not preload:
@@ -226,9 +234,12 @@
 
     install_signal_handlers(loop)
 
+    async def run():
+        await scheduler
+        await scheduler.finished()
+
     try:
-        loop.start()
-        loop.close()
+        loop.run_sync(run)
     finally:
         scheduler.stop()
         if local_directory_created:

{"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@@ -120,6 +120,12 @@\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()\n def main(\n     host,\n@@ -141,6 +147,7 @@\n     tls_cert,\n     tls_key,\n     dashboard_address,\n+    idle_timeout,\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@@ -211,6 +218,7 @@\n         protocol=protocol,\n         dashboard_address=dashboard_address if dashboard else None,\n         service_kwargs={\"dashboard\": {\"prefix\": dashboard_prefix}},\n+        idle_timeout=idle_timeout,\n     )\n     scheduler.start()\n     if not preload:\n@@ -226,9 +234,12 @@\n \n     install_signal_handlers(loop)\n \n+    async def run():\n+        await scheduler\n+        await scheduler.finished()\n+\n     try:\n-        loop.start()\n-        loop.close()\n+        loop.run_sync(run)\n     finally:\n         scheduler.stop()\n         if local_directory_created:\n", "issue": "Scheduler timeout fails to exit `dask-scheduler` process\nThe scheduler timeout introduced in #2652 seems to successfully close the scheduler comms, however the process doesn't exit.\r\n\r\nI would expect the process to also exit when the scheduler has closed.\r\n\r\n```console\r\n$ pip install git+https://github.com/dask/distributed.git\r\n\r\n$ export DASK_DISTRIBUTED__SCHEDULER__IDLE_TIMEOUT='200 ms'\r\n\r\n$ python -c 'import distributed; print(distributed.__version__)' \r\n2.1.0+4.ge5ec8daa \r\n\r\n$ dask-scheduler  \r\ndistributed.scheduler - INFO - ----------------------------------------------- \r\ndistributed.scheduler - INFO - Clear task state \r\ndistributed.dashboard.proxy - INFO - To route to workers diagnostics web server please install jupyter-server-proxy: pip install jupyter-server-proxy\r\ndistributed.scheduler - INFO -   Scheduler at:     tcp://172.17.0.2:8786 \r\ndistributed.scheduler - INFO -   dashboard at:                     :8787\r\ndistributed.scheduler - INFO - Local Directory:    /tmp/scheduler-gx0ramly\r\ndistributed.scheduler - INFO - -----------------------------------------------\r\ndistributed.scheduler - INFO - Scheduler closing... \r\ndistributed.scheduler - INFO - Scheduler closing all comms\r\n\r\n\r\n... (hit enter a bunch of times and waited a while here before hitting CTRL+c)...\r\n\r\n\r\n^Cdistributed.scheduler - INFO - End scheduler at 'tcp://172.17.0.2:8786' \r\n```\r\n\r\nThe scheduler should exit cleanly to satisfy the requirements in #2346.\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport atexit\nimport dask\nimport logging\nimport gc\nimport os\nimport re\nimport shutil\nimport sys\nimport tempfile\nimport warnings\n\nimport click\n\nfrom tornado.ioloop import IOLoop\n\nfrom distributed import Scheduler\nfrom distributed.security import Security\nfrom distributed.cli.utils import check_python_3, install_signal_handlers\nfrom distributed.preloading import preload_modules, 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    help=\"Address on which to listen for diagnostics dashboard\",\n)\n@click.option(\n    \"--dashboard/--no-dashboard\",\n    \"dashboard\",\n    default=True,\n    show_default=True,\n    required=False,\n    help=\"Launch the 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(\"--show/--no-show\", default=False, help=\"Show web UI\")\n@click.option(\n    \"--dashboard-prefix\", type=str, default=None, 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    \"--local-directory\", default=\"\", type=str, help=\"Directory to place scheduler files\"\n)\n@click.option(\n    \"--preload\",\n    type=str,\n    multiple=True,\n    is_eager=True,\n    default=\"\",\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.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    scheduler_file,\n    interface,\n    protocol,\n    local_directory,\n    preload,\n    preload_argv,\n    tls_ca_file,\n    tls_cert,\n    tls_key,\n    dashboard_address,\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 = Security(\n        tls_ca_file=tls_ca_file, tls_scheduler_cert=tls_cert, tls_scheduler_key=tls_key\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    local_directory_created = False\n    if local_directory:\n        if not os.path.exists(local_directory):\n            os.mkdir(local_directory)\n            local_directory_created = True\n    else:\n        local_directory = tempfile.mkdtemp(prefix=\"scheduler-\")\n        local_directory_created = True\n    if local_directory not in sys.path:\n        sys.path.insert(0, local_directory)\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    loop = IOLoop.current()\n    logger.info(\"-\" * 47)\n\n    scheduler = Scheduler(\n        loop=loop,\n        scheduler_file=scheduler_file,\n        security=sec,\n        host=host,\n        port=port,\n        interface=interface,\n        protocol=protocol,\n        dashboard_address=dashboard_address if dashboard else None,\n        service_kwargs={\"dashboard\": {\"prefix\": dashboard_prefix}},\n    )\n    scheduler.start()\n    if not preload:\n        preload = dask.config.get(\"distributed.scheduler.preload\")\n    if not preload_argv:\n        preload_argv = dask.config.get(\"distributed.scheduler.preload-argv\")\n    preload_modules(\n        preload, parameter=scheduler, file_dir=local_directory, argv=preload_argv\n    )\n\n    logger.info(\"Local Directory: %26s\", local_directory)\n    logger.info(\"-\" * 47)\n\n    install_signal_handlers(loop)\n\n    try:\n        loop.start()\n        loop.close()\n    finally:\n        scheduler.stop()\n        if local_directory_created:\n            shutil.rmtree(local_directory)\n\n        logger.info(\"End scheduler at %r\", scheduler.address)\n\n\ndef go():\n    check_python_3()\n    main()\n\n\nif __name__ == \"__main__\":\n    go()\n", "path": "distributed/cli/dask_scheduler.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport atexit\nimport dask\nimport logging\nimport gc\nimport os\nimport re\nimport shutil\nimport sys\nimport tempfile\nimport warnings\n\nimport click\n\nfrom tornado.ioloop import IOLoop\n\nfrom distributed import Scheduler\nfrom distributed.security import Security\nfrom distributed.cli.utils import check_python_3, install_signal_handlers\nfrom distributed.preloading import preload_modules, 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    help=\"Address on which to listen for diagnostics dashboard\",\n)\n@click.option(\n    \"--dashboard/--no-dashboard\",\n    \"dashboard\",\n    default=True,\n    show_default=True,\n    required=False,\n    help=\"Launch the 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(\"--show/--no-show\", default=False, help=\"Show web UI\")\n@click.option(\n    \"--dashboard-prefix\", type=str, default=None, 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    \"--local-directory\", default=\"\", type=str, help=\"Directory to place scheduler files\"\n)\n@click.option(\n    \"--preload\",\n    type=str,\n    multiple=True,\n    is_eager=True,\n    default=\"\",\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    scheduler_file,\n    interface,\n    protocol,\n    local_directory,\n    preload,\n    preload_argv,\n    tls_ca_file,\n    tls_cert,\n    tls_key,\n    dashboard_address,\n    idle_timeout,\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 = Security(\n        tls_ca_file=tls_ca_file, tls_scheduler_cert=tls_cert, tls_scheduler_key=tls_key\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    local_directory_created = False\n    if local_directory:\n        if not os.path.exists(local_directory):\n            os.mkdir(local_directory)\n            local_directory_created = True\n    else:\n        local_directory = tempfile.mkdtemp(prefix=\"scheduler-\")\n        local_directory_created = True\n    if local_directory not in sys.path:\n        sys.path.insert(0, local_directory)\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    loop = IOLoop.current()\n    logger.info(\"-\" * 47)\n\n    scheduler = Scheduler(\n        loop=loop,\n        scheduler_file=scheduler_file,\n        security=sec,\n        host=host,\n        port=port,\n        interface=interface,\n        protocol=protocol,\n        dashboard_address=dashboard_address if dashboard else None,\n        service_kwargs={\"dashboard\": {\"prefix\": dashboard_prefix}},\n        idle_timeout=idle_timeout,\n    )\n    scheduler.start()\n    if not preload:\n        preload = dask.config.get(\"distributed.scheduler.preload\")\n    if not preload_argv:\n        preload_argv = dask.config.get(\"distributed.scheduler.preload-argv\")\n    preload_modules(\n        preload, parameter=scheduler, file_dir=local_directory, argv=preload_argv\n    )\n\n    logger.info(\"Local Directory: %26s\", local_directory)\n    logger.info(\"-\" * 47)\n\n    install_signal_handlers(loop)\n\n    async def run():\n        await scheduler\n        await scheduler.finished()\n\n    try:\n        loop.run_sync(run)\n    finally:\n        scheduler.stop()\n        if local_directory_created:\n            shutil.rmtree(local_directory)\n\n        logger.info(\"End scheduler at %r\", scheduler.address)\n\n\ndef go():\n    check_python_3()\n    main()\n\n\nif __name__ == \"__main__\":\n    go()\n", "path": "distributed/cli/dask_scheduler.py"}]}