We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
ValueError / hang on LocalCluster startup
I'm loading my conda environment off NAS storage, and when usage is heavy, there can be long delays in loading things, which I believe can exceed the default timeout (five seconds) here:

https://github.com/dask/distributed/blob/master/distributed/client.py#L428

As a result I see this on a call to `LocalCluster()`:

	Traceback (most recent call last):                                                                                                                            
	  File "lib/python3.5/site-packages/tornado/gen.py", line 910, in error_callback                                                
		future.result()                                                                                                                                           
	  File "lib/python3.5/site-packages/tornado/concurrent.py", line 238, in result                                                 
		raise_exc_info(self._exc_info)                                                                                                                            
	  File "<string>", line 4, in raise_exc_info                                                                                                                  
	  File "lib/python3.5/site-packages/tornado/gen.py", line 1063, in run                                                          
		yielded = self.gen.throw(*exc_info)                                        
	  File "lib/python3.5/site-packages/distributed/nanny.py", line 300, in start
		yield self._wait_until_running()                                           
	  File "lib/python3.5/site-packages/tornado/gen.py", line 1055, in run                                                          
		value = future.result()                                                                                                                                   
	  File "lib/python3.5/site-packages/tornado/concurrent.py", line 238, in result
		raise_exc_info(self._exc_info)                                             
	  File "<string>", line 4, in raise_exc_info                                                                                                                  
	  File "lib/python3.5/site-packages/tornado/gen.py", line 1069, in run
		yielded = self.gen.send(value)                                                                                                                            
	  File "lib/python3.5/site-packages/distributed/nanny.py", line 386, in _wait_until_running
		raise ValueError("Worker not started")                                                                                                                                                                                                                                                                                   
	ValueError: Worker not started         

This `ValueError` does not seem to be passed up to my application; my application seems frozen waiting for the cluster start up (not 100% sure where) - ie, it never returns from the `LocalCluster()` call. Although the multiprocessing pool actually does start up, in this case too late to matter.

How can I catch / retry this situation?

--- 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_worker.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 import atexit
4 from datetime import timedelta
5 from functools import partial
6 import json
7 import logging
8 import os
9 import shutil
10 import signal
11 from sys import exit
12 from time import sleep
13 
14 import click
15 from distributed import Nanny, Worker, rpc
16 from distributed.utils import All, get_ip_interface
17 from distributed.worker import _ncores
18 from distributed.http import HTTPWorker
19 from distributed.metrics import time
20 from distributed.security import Security
21 from distributed.cli.utils import check_python_3, uri_from_host_port
22 
23 from toolz import valmap
24 from tornado.ioloop import IOLoop, TimeoutError
25 from tornado import gen
26 
27 logger = logging.getLogger('distributed.dask_worker')
28 
29 
30 pem_file_option_type = click.Path(exists=True, resolve_path=True)
31 
32 @click.command()
33 @click.argument('scheduler', type=str, required=False)
34 @click.option('--tls-ca-file', type=pem_file_option_type, default=None,
35               help="CA cert(s) file for TLS (in PEM format)")
36 @click.option('--tls-cert', type=pem_file_option_type, default=None,
37               help="certificate file for TLS (in PEM format)")
38 @click.option('--tls-key', type=pem_file_option_type, default=None,
39               help="private key file for TLS (in PEM format)")
40 @click.option('--worker-port', type=int, default=0,
41               help="Serving computation port, defaults to random")
42 @click.option('--http-port', type=int, default=0,
43               help="Serving http port, defaults to random")
44 @click.option('--nanny-port', type=int, default=0,
45               help="Serving nanny port, defaults to random")
46 @click.option('--bokeh-port', type=int, default=8789,
47               help="Bokeh port, defaults to 8789")
48 @click.option('--bokeh/--no-bokeh', 'bokeh', default=True, show_default=True,
49               required=False, help="Launch Bokeh Web UI")
50 @click.option('--host', type=str, default=None,
51               help="Serving host. Should be an ip address that is"
52                    " visible to the scheduler and other workers. "
53                    "See --interface.")
54 @click.option('--interface', type=str, default=None,
55               help="Network interface like 'eth0' or 'ib0'")
56 @click.option('--nthreads', type=int, default=0,
57               help="Number of threads per process.")
58 @click.option('--nprocs', type=int, default=1,
59               help="Number of worker processes.  Defaults to one.")
60 @click.option('--name', type=str, default='',
61               help="A unique name for this worker like 'worker-1'")
62 @click.option('--memory-limit', default='auto',
63               help="Number of bytes before spilling data to disk. "
64                    "This can be an integer (nbytes) "
65                    "float (fraction of total memory) "
66                    "or 'auto'")
67 @click.option('--reconnect/--no-reconnect', default=True,
68               help="Reconnect to scheduler if disconnected")
69 @click.option('--nanny/--no-nanny', default=True,
70               help="Start workers in nanny process for management")
71 @click.option('--pid-file', type=str, default='',
72               help="File to write the process PID")
73 @click.option('--local-directory', default='', type=str,
74               help="Directory to place worker files")
75 @click.option('--resources', type=str, default='',
76               help='Resources for task constraints like "GPU=2 MEM=10e9"')
77 @click.option('--scheduler-file', type=str, default='',
78               help='Filename to JSON encoded scheduler information. '
79                    'Use with dask-scheduler --scheduler-file')
80 @click.option('--death-timeout', type=float, default=None,
81               help="Seconds to wait for a scheduler before closing")
82 @click.option('--bokeh-prefix', type=str, default=None,
83               help="Prefix for the bokeh app")
84 @click.option('--preload', type=str, multiple=True,
85               help='Module that should be loaded by each worker process '
86                    'like "foo.bar" or "/path/to/foo.py"')
87 def main(scheduler, host, worker_port, http_port, nanny_port, nthreads, nprocs,
88          nanny, name, memory_limit, pid_file, reconnect,
89          resources, bokeh, bokeh_port, local_directory, scheduler_file,
90          interface, death_timeout, preload, bokeh_prefix,
91          tls_ca_file, tls_cert, tls_key):
92     sec = Security(tls_ca_file=tls_ca_file,
93                    tls_worker_cert=tls_cert,
94                    tls_worker_key=tls_key,
95                    )
96 
97     if nanny:
98         port = nanny_port
99     else:
100         port = worker_port
101 
102     if nprocs > 1 and worker_port != 0:
103         logger.error("Failed to launch worker.  You cannot use the --port argument when nprocs > 1.")
104         exit(1)
105 
106     if nprocs > 1 and name:
107         logger.error("Failed to launch worker.  You cannot use the --name argument when nprocs > 1.")
108         exit(1)
109 
110     if nprocs > 1 and not nanny:
111         logger.error("Failed to launch worker.  You cannot use the --no-nanny argument when nprocs > 1.")
112         exit(1)
113 
114     if not nthreads:
115         nthreads = _ncores // nprocs
116 
117     if pid_file:
118         with open(pid_file, 'w') as f:
119             f.write(str(os.getpid()))
120 
121         def del_pid_file():
122             if os.path.exists(pid_file):
123                 os.remove(pid_file)
124         atexit.register(del_pid_file)
125 
126     services = {('http', http_port): HTTPWorker}
127 
128     if bokeh:
129         try:
130             from distributed.bokeh.worker import BokehWorker
131         except ImportError:
132             pass
133         else:
134             if bokeh_prefix:
135                 result = (BokehWorker, {'prefix': bokeh_prefix})
136             else:
137                 result = BokehWorker
138             services[('bokeh', bokeh_port)] = result
139 
140     if resources:
141         resources = resources.replace(',', ' ').split()
142         resources = dict(pair.split('=') for pair in resources)
143         resources = valmap(float, resources)
144     else:
145         resources = None
146 
147     loop = IOLoop.current()
148 
149     if nanny:
150         kwargs = {'worker_port': worker_port}
151         t = Nanny
152     else:
153         kwargs = {}
154         if nanny_port:
155             kwargs['service_ports'] = {'nanny': nanny_port}
156         t = Worker
157 
158     if scheduler_file:
159         while not os.path.exists(scheduler_file):
160             sleep(0.01)
161         for i in range(10):
162             try:
163                 with open(scheduler_file) as f:
164                     cfg = json.load(f)
165                 scheduler = cfg['address']
166                 break
167             except (ValueError, KeyError):  # race with scheduler on file
168                 sleep(0.01)
169 
170     if not scheduler:
171         raise ValueError("Need to provide scheduler address like\n"
172                          "dask-worker SCHEDULER_ADDRESS:8786")
173 
174     if interface:
175         if host:
176             raise ValueError("Can not specify both interface and host")
177         else:
178             host = get_ip_interface(interface)
179 
180     if host or port:
181         addr = uri_from_host_port(host, port, 0)
182     else:
183         # Choose appropriate address for scheduler
184         addr = None
185 
186     nannies = [t(scheduler, ncores=nthreads,
187                  services=services, name=name, loop=loop, resources=resources,
188                  memory_limit=memory_limit, reconnect=reconnect,
189                  local_dir=local_directory, death_timeout=death_timeout,
190                  preload=preload, security=sec,
191                  **kwargs)
192                for i in range(nprocs)]
193 
194     @gen.coroutine
195     def close_all():
196         try:
197             if nanny:
198                 yield [n._close(timeout=2) for n in nannies]
199         finally:
200             loop.stop()
201 
202     def handle_signal(signum, frame):
203         logger.info("Exiting on signal %d", signum)
204         if loop._running:
205             loop.add_callback_from_signal(loop.stop)
206         else:
207             exit(0)
208 
209     # NOTE: We can't use the generic install_signal_handlers() function from
210     # distributed.cli.utils because we're handling the signal differently.
211     signal.signal(signal.SIGINT, handle_signal)
212     signal.signal(signal.SIGTERM, handle_signal)
213 
214     for n in nannies:
215         n.start(addr)
216 
217     @gen.coroutine
218     def run():
219         while all(n.status != 'closed' for n in nannies):
220             yield gen.sleep(0.2)
221 
222     try:
223         loop.run_sync(run)
224     except (KeyboardInterrupt, TimeoutError):
225         pass
226     finally:
227         logger.info("End worker")
228 
229     # Clean exit: unregister all workers from scheduler
230     loop.run_sync(close_all)
231 
232 
233 def go():
234     check_python_3()
235     main()
236 
237 if __name__ == '__main__':
238     go()
239
```
--- 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_worker.py b/distributed/cli/dask_worker.py
--- a/distributed/cli/dask_worker.py
+++ b/distributed/cli/dask_worker.py
@@ -211,11 +211,9 @@
     signal.signal(signal.SIGINT, handle_signal)
     signal.signal(signal.SIGTERM, handle_signal)
 
-    for n in nannies:
-        n.start(addr)
-
     @gen.coroutine
     def run():
+        yield [n.start(addr) for n in nannies]
         while all(n.status != 'closed' for n in nannies):
             yield gen.sleep(0.2)

{"golden_diff": "diff --git a/distributed/cli/dask_worker.py b/distributed/cli/dask_worker.py\n--- a/distributed/cli/dask_worker.py\n+++ b/distributed/cli/dask_worker.py\n@@ -211,11 +211,9 @@\n     signal.signal(signal.SIGINT, handle_signal)\n     signal.signal(signal.SIGTERM, handle_signal)\n \n-    for n in nannies:\n-        n.start(addr)\n-\n     @gen.coroutine\n     def run():\n+        yield [n.start(addr) for n in nannies]\n         while all(n.status != 'closed' for n in nannies):\n             yield gen.sleep(0.2)\n", "issue": "ValueError / hang on LocalCluster startup\nI'm loading my conda environment off NAS storage, and when usage is heavy, there can be long delays in loading things, which I believe can exceed the default timeout (five seconds) here:\r\n\r\nhttps://github.com/dask/distributed/blob/master/distributed/client.py#L428\r\n\r\nAs a result I see this on a call to `LocalCluster()`:\r\n\r\n\tTraceback (most recent call last):                                                                                                                            \r\n\t  File \"lib/python3.5/site-packages/tornado/gen.py\", line 910, in error_callback                                                \r\n\t\tfuture.result()                                                                                                                                           \r\n\t  File \"lib/python3.5/site-packages/tornado/concurrent.py\", line 238, in result                                                 \r\n\t\traise_exc_info(self._exc_info)                                                                                                                            \r\n\t  File \"<string>\", line 4, in raise_exc_info                                                                                                                  \r\n\t  File \"lib/python3.5/site-packages/tornado/gen.py\", line 1063, in run                                                          \r\n\t\tyielded = self.gen.throw(*exc_info)                                        \r\n\t  File \"lib/python3.5/site-packages/distributed/nanny.py\", line 300, in start\r\n\t\tyield self._wait_until_running()                                           \r\n\t  File \"lib/python3.5/site-packages/tornado/gen.py\", line 1055, in run                                                          \r\n\t\tvalue = future.result()                                                                                                                                   \r\n\t  File \"lib/python3.5/site-packages/tornado/concurrent.py\", line 238, in result\r\n\t\traise_exc_info(self._exc_info)                                             \r\n\t  File \"<string>\", line 4, in raise_exc_info                                                                                                                  \r\n\t  File \"lib/python3.5/site-packages/tornado/gen.py\", line 1069, in run\r\n\t\tyielded = self.gen.send(value)                                                                                                                            \r\n\t  File \"lib/python3.5/site-packages/distributed/nanny.py\", line 386, in _wait_until_running\r\n\t\traise ValueError(\"Worker not started\")                                                                                                                                                                                                                                                                                   \r\n\tValueError: Worker not started         \r\n\r\nThis `ValueError` does not seem to be passed up to my application; my application seems frozen waiting for the cluster start up (not 100% sure where) - ie, it never returns from the `LocalCluster()` call. Although the multiprocessing pool actually does start up, in this case too late to matter.\r\n\r\nHow can I catch / retry this situation?\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport atexit\nfrom datetime import timedelta\nfrom functools import partial\nimport json\nimport logging\nimport os\nimport shutil\nimport signal\nfrom sys import exit\nfrom time import sleep\n\nimport click\nfrom distributed import Nanny, Worker, rpc\nfrom distributed.utils import All, get_ip_interface\nfrom distributed.worker import _ncores\nfrom distributed.http import HTTPWorker\nfrom distributed.metrics import time\nfrom distributed.security import Security\nfrom distributed.cli.utils import check_python_3, uri_from_host_port\n\nfrom toolz import valmap\nfrom tornado.ioloop import IOLoop, TimeoutError\nfrom tornado import gen\n\nlogger = logging.getLogger('distributed.dask_worker')\n\n\npem_file_option_type = click.Path(exists=True, resolve_path=True)\n\n@click.command()\n@click.argument('scheduler', type=str, required=False)\n@click.option('--tls-ca-file', type=pem_file_option_type, default=None,\n              help=\"CA cert(s) file for TLS (in PEM format)\")\n@click.option('--tls-cert', type=pem_file_option_type, default=None,\n              help=\"certificate file for TLS (in PEM format)\")\n@click.option('--tls-key', type=pem_file_option_type, default=None,\n              help=\"private key file for TLS (in PEM format)\")\n@click.option('--worker-port', type=int, default=0,\n              help=\"Serving computation port, defaults to random\")\n@click.option('--http-port', type=int, default=0,\n              help=\"Serving http port, defaults to random\")\n@click.option('--nanny-port', type=int, default=0,\n              help=\"Serving nanny port, defaults to random\")\n@click.option('--bokeh-port', type=int, default=8789,\n              help=\"Bokeh port, defaults to 8789\")\n@click.option('--bokeh/--no-bokeh', 'bokeh', default=True, show_default=True,\n              required=False, help=\"Launch Bokeh Web UI\")\n@click.option('--host', type=str, default=None,\n              help=\"Serving host. Should be an ip address that is\"\n                   \" visible to the scheduler and other workers. \"\n                   \"See --interface.\")\n@click.option('--interface', type=str, default=None,\n              help=\"Network interface like 'eth0' or 'ib0'\")\n@click.option('--nthreads', type=int, default=0,\n              help=\"Number of threads per process.\")\n@click.option('--nprocs', type=int, default=1,\n              help=\"Number of worker processes.  Defaults to one.\")\n@click.option('--name', type=str, default='',\n              help=\"A unique name for this worker like 'worker-1'\")\n@click.option('--memory-limit', default='auto',\n              help=\"Number of bytes before spilling data to disk. \"\n                   \"This can be an integer (nbytes) \"\n                   \"float (fraction of total memory) \"\n                   \"or 'auto'\")\n@click.option('--reconnect/--no-reconnect', default=True,\n              help=\"Reconnect to scheduler if disconnected\")\n@click.option('--nanny/--no-nanny', default=True,\n              help=\"Start workers in nanny process for management\")\n@click.option('--pid-file', type=str, default='',\n              help=\"File to write the process PID\")\n@click.option('--local-directory', default='', type=str,\n              help=\"Directory to place worker files\")\n@click.option('--resources', type=str, default='',\n              help='Resources for task constraints like \"GPU=2 MEM=10e9\"')\n@click.option('--scheduler-file', type=str, default='',\n              help='Filename to JSON encoded scheduler information. '\n                   'Use with dask-scheduler --scheduler-file')\n@click.option('--death-timeout', type=float, default=None,\n              help=\"Seconds to wait for a scheduler before closing\")\n@click.option('--bokeh-prefix', type=str, default=None,\n              help=\"Prefix for the bokeh app\")\n@click.option('--preload', type=str, multiple=True,\n              help='Module that should be loaded by each worker process '\n                   'like \"foo.bar\" or \"/path/to/foo.py\"')\ndef main(scheduler, host, worker_port, http_port, nanny_port, nthreads, nprocs,\n         nanny, name, memory_limit, pid_file, reconnect,\n         resources, bokeh, bokeh_port, local_directory, scheduler_file,\n         interface, death_timeout, preload, bokeh_prefix,\n         tls_ca_file, tls_cert, tls_key):\n    sec = Security(tls_ca_file=tls_ca_file,\n                   tls_worker_cert=tls_cert,\n                   tls_worker_key=tls_key,\n                   )\n\n    if nanny:\n        port = nanny_port\n    else:\n        port = worker_port\n\n    if nprocs > 1 and worker_port != 0:\n        logger.error(\"Failed to launch worker.  You cannot use the --port argument when nprocs > 1.\")\n        exit(1)\n\n    if nprocs > 1 and name:\n        logger.error(\"Failed to launch worker.  You cannot use the --name argument when nprocs > 1.\")\n        exit(1)\n\n    if nprocs > 1 and not nanny:\n        logger.error(\"Failed to launch worker.  You cannot use the --no-nanny argument when nprocs > 1.\")\n        exit(1)\n\n    if not nthreads:\n        nthreads = _ncores // nprocs\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        atexit.register(del_pid_file)\n\n    services = {('http', http_port): HTTPWorker}\n\n    if bokeh:\n        try:\n            from distributed.bokeh.worker import BokehWorker\n        except ImportError:\n            pass\n        else:\n            if bokeh_prefix:\n                result = (BokehWorker, {'prefix': bokeh_prefix})\n            else:\n                result = BokehWorker\n            services[('bokeh', bokeh_port)] = result\n\n    if resources:\n        resources = resources.replace(',', ' ').split()\n        resources = dict(pair.split('=') for pair in resources)\n        resources = valmap(float, resources)\n    else:\n        resources = None\n\n    loop = IOLoop.current()\n\n    if nanny:\n        kwargs = {'worker_port': worker_port}\n        t = Nanny\n    else:\n        kwargs = {}\n        if nanny_port:\n            kwargs['service_ports'] = {'nanny': nanny_port}\n        t = Worker\n\n    if scheduler_file:\n        while not os.path.exists(scheduler_file):\n            sleep(0.01)\n        for i in range(10):\n            try:\n                with open(scheduler_file) as f:\n                    cfg = json.load(f)\n                scheduler = cfg['address']\n                break\n            except (ValueError, KeyError):  # race with scheduler on file\n                sleep(0.01)\n\n    if not scheduler:\n        raise ValueError(\"Need to provide scheduler address like\\n\"\n                         \"dask-worker SCHEDULER_ADDRESS:8786\")\n\n    if interface:\n        if host:\n            raise ValueError(\"Can not specify both interface and host\")\n        else:\n            host = get_ip_interface(interface)\n\n    if host or port:\n        addr = uri_from_host_port(host, port, 0)\n    else:\n        # Choose appropriate address for scheduler\n        addr = None\n\n    nannies = [t(scheduler, ncores=nthreads,\n                 services=services, name=name, loop=loop, resources=resources,\n                 memory_limit=memory_limit, reconnect=reconnect,\n                 local_dir=local_directory, death_timeout=death_timeout,\n                 preload=preload, security=sec,\n                 **kwargs)\n               for i in range(nprocs)]\n\n    @gen.coroutine\n    def close_all():\n        try:\n            if nanny:\n                yield [n._close(timeout=2) for n in nannies]\n        finally:\n            loop.stop()\n\n    def handle_signal(signum, frame):\n        logger.info(\"Exiting on signal %d\", signum)\n        if loop._running:\n            loop.add_callback_from_signal(loop.stop)\n        else:\n            exit(0)\n\n    # NOTE: We can't use the generic install_signal_handlers() function from\n    # distributed.cli.utils because we're handling the signal differently.\n    signal.signal(signal.SIGINT, handle_signal)\n    signal.signal(signal.SIGTERM, handle_signal)\n\n    for n in nannies:\n        n.start(addr)\n\n    @gen.coroutine\n    def run():\n        while all(n.status != 'closed' for n in nannies):\n            yield gen.sleep(0.2)\n\n    try:\n        loop.run_sync(run)\n    except (KeyboardInterrupt, TimeoutError):\n        pass\n    finally:\n        logger.info(\"End worker\")\n\n    # Clean exit: unregister all workers from scheduler\n    loop.run_sync(close_all)\n\n\ndef go():\n    check_python_3()\n    main()\n\nif __name__ == '__main__':\n    go()\n", "path": "distributed/cli/dask_worker.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport atexit\nfrom datetime import timedelta\nfrom functools import partial\nimport json\nimport logging\nimport os\nimport shutil\nimport signal\nfrom sys import exit\nfrom time import sleep\n\nimport click\nfrom distributed import Nanny, Worker, rpc\nfrom distributed.utils import All, get_ip_interface\nfrom distributed.worker import _ncores\nfrom distributed.http import HTTPWorker\nfrom distributed.metrics import time\nfrom distributed.security import Security\nfrom distributed.cli.utils import check_python_3, uri_from_host_port\n\nfrom toolz import valmap\nfrom tornado.ioloop import IOLoop, TimeoutError\nfrom tornado import gen\n\nlogger = logging.getLogger('distributed.dask_worker')\n\n\npem_file_option_type = click.Path(exists=True, resolve_path=True)\n\n@click.command()\n@click.argument('scheduler', type=str, required=False)\n@click.option('--tls-ca-file', type=pem_file_option_type, default=None,\n              help=\"CA cert(s) file for TLS (in PEM format)\")\n@click.option('--tls-cert', type=pem_file_option_type, default=None,\n              help=\"certificate file for TLS (in PEM format)\")\n@click.option('--tls-key', type=pem_file_option_type, default=None,\n              help=\"private key file for TLS (in PEM format)\")\n@click.option('--worker-port', type=int, default=0,\n              help=\"Serving computation port, defaults to random\")\n@click.option('--http-port', type=int, default=0,\n              help=\"Serving http port, defaults to random\")\n@click.option('--nanny-port', type=int, default=0,\n              help=\"Serving nanny port, defaults to random\")\n@click.option('--bokeh-port', type=int, default=8789,\n              help=\"Bokeh port, defaults to 8789\")\n@click.option('--bokeh/--no-bokeh', 'bokeh', default=True, show_default=True,\n              required=False, help=\"Launch Bokeh Web UI\")\n@click.option('--host', type=str, default=None,\n              help=\"Serving host. Should be an ip address that is\"\n                   \" visible to the scheduler and other workers. \"\n                   \"See --interface.\")\n@click.option('--interface', type=str, default=None,\n              help=\"Network interface like 'eth0' or 'ib0'\")\n@click.option('--nthreads', type=int, default=0,\n              help=\"Number of threads per process.\")\n@click.option('--nprocs', type=int, default=1,\n              help=\"Number of worker processes.  Defaults to one.\")\n@click.option('--name', type=str, default='',\n              help=\"A unique name for this worker like 'worker-1'\")\n@click.option('--memory-limit', default='auto',\n              help=\"Number of bytes before spilling data to disk. \"\n                   \"This can be an integer (nbytes) \"\n                   \"float (fraction of total memory) \"\n                   \"or 'auto'\")\n@click.option('--reconnect/--no-reconnect', default=True,\n              help=\"Reconnect to scheduler if disconnected\")\n@click.option('--nanny/--no-nanny', default=True,\n              help=\"Start workers in nanny process for management\")\n@click.option('--pid-file', type=str, default='',\n              help=\"File to write the process PID\")\n@click.option('--local-directory', default='', type=str,\n              help=\"Directory to place worker files\")\n@click.option('--resources', type=str, default='',\n              help='Resources for task constraints like \"GPU=2 MEM=10e9\"')\n@click.option('--scheduler-file', type=str, default='',\n              help='Filename to JSON encoded scheduler information. '\n                   'Use with dask-scheduler --scheduler-file')\n@click.option('--death-timeout', type=float, default=None,\n              help=\"Seconds to wait for a scheduler before closing\")\n@click.option('--bokeh-prefix', type=str, default=None,\n              help=\"Prefix for the bokeh app\")\n@click.option('--preload', type=str, multiple=True,\n              help='Module that should be loaded by each worker process '\n                   'like \"foo.bar\" or \"/path/to/foo.py\"')\ndef main(scheduler, host, worker_port, http_port, nanny_port, nthreads, nprocs,\n         nanny, name, memory_limit, pid_file, reconnect,\n         resources, bokeh, bokeh_port, local_directory, scheduler_file,\n         interface, death_timeout, preload, bokeh_prefix,\n         tls_ca_file, tls_cert, tls_key):\n    sec = Security(tls_ca_file=tls_ca_file,\n                   tls_worker_cert=tls_cert,\n                   tls_worker_key=tls_key,\n                   )\n\n    if nanny:\n        port = nanny_port\n    else:\n        port = worker_port\n\n    if nprocs > 1 and worker_port != 0:\n        logger.error(\"Failed to launch worker.  You cannot use the --port argument when nprocs > 1.\")\n        exit(1)\n\n    if nprocs > 1 and name:\n        logger.error(\"Failed to launch worker.  You cannot use the --name argument when nprocs > 1.\")\n        exit(1)\n\n    if nprocs > 1 and not nanny:\n        logger.error(\"Failed to launch worker.  You cannot use the --no-nanny argument when nprocs > 1.\")\n        exit(1)\n\n    if not nthreads:\n        nthreads = _ncores // nprocs\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        atexit.register(del_pid_file)\n\n    services = {('http', http_port): HTTPWorker}\n\n    if bokeh:\n        try:\n            from distributed.bokeh.worker import BokehWorker\n        except ImportError:\n            pass\n        else:\n            if bokeh_prefix:\n                result = (BokehWorker, {'prefix': bokeh_prefix})\n            else:\n                result = BokehWorker\n            services[('bokeh', bokeh_port)] = result\n\n    if resources:\n        resources = resources.replace(',', ' ').split()\n        resources = dict(pair.split('=') for pair in resources)\n        resources = valmap(float, resources)\n    else:\n        resources = None\n\n    loop = IOLoop.current()\n\n    if nanny:\n        kwargs = {'worker_port': worker_port}\n        t = Nanny\n    else:\n        kwargs = {}\n        if nanny_port:\n            kwargs['service_ports'] = {'nanny': nanny_port}\n        t = Worker\n\n    if scheduler_file:\n        while not os.path.exists(scheduler_file):\n            sleep(0.01)\n        for i in range(10):\n            try:\n                with open(scheduler_file) as f:\n                    cfg = json.load(f)\n                scheduler = cfg['address']\n                break\n            except (ValueError, KeyError):  # race with scheduler on file\n                sleep(0.01)\n\n    if not scheduler:\n        raise ValueError(\"Need to provide scheduler address like\\n\"\n                         \"dask-worker SCHEDULER_ADDRESS:8786\")\n\n    if interface:\n        if host:\n            raise ValueError(\"Can not specify both interface and host\")\n        else:\n            host = get_ip_interface(interface)\n\n    if host or port:\n        addr = uri_from_host_port(host, port, 0)\n    else:\n        # Choose appropriate address for scheduler\n        addr = None\n\n    nannies = [t(scheduler, ncores=nthreads,\n                 services=services, name=name, loop=loop, resources=resources,\n                 memory_limit=memory_limit, reconnect=reconnect,\n                 local_dir=local_directory, death_timeout=death_timeout,\n                 preload=preload, security=sec,\n                 **kwargs)\n               for i in range(nprocs)]\n\n    @gen.coroutine\n    def close_all():\n        try:\n            if nanny:\n                yield [n._close(timeout=2) for n in nannies]\n        finally:\n            loop.stop()\n\n    def handle_signal(signum, frame):\n        logger.info(\"Exiting on signal %d\", signum)\n        if loop._running:\n            loop.add_callback_from_signal(loop.stop)\n        else:\n            exit(0)\n\n    # NOTE: We can't use the generic install_signal_handlers() function from\n    # distributed.cli.utils because we're handling the signal differently.\n    signal.signal(signal.SIGINT, handle_signal)\n    signal.signal(signal.SIGTERM, handle_signal)\n\n    @gen.coroutine\n    def run():\n        yield [n.start(addr) for n in nannies]\n        while all(n.status != 'closed' for n in nannies):\n            yield gen.sleep(0.2)\n\n    try:\n        loop.run_sync(run)\n    except (KeyboardInterrupt, TimeoutError):\n        pass\n    finally:\n        logger.info(\"End worker\")\n\n    # Clean exit: unregister all workers from scheduler\n    loop.run_sync(close_all)\n\n\ndef go():\n    check_python_3()\n    main()\n\nif __name__ == '__main__':\n    go()\n", "path": "distributed/cli/dask_worker.py"}]}