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/deploy/local.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 import atexit
4 import logging
5 import math
6 from threading import Thread
7 from time import sleep
8 import warnings
9 import weakref
10 
11 from tornado import gen
12 from tornado.ioloop import IOLoop
13 
14 from ..core import CommClosedError
15 from ..utils import sync, ignoring, All, silence_logging
16 from ..nanny import Nanny
17 from ..scheduler import Scheduler
18 from ..worker import Worker, _ncores
19 
20 logger = logging.getLogger(__name__)
21 
22 
23 class LocalCluster(object):
24     """ Create local Scheduler and Workers
25 
26     This creates a "cluster" of a scheduler and workers running on the local
27     machine.
28 
29     Parameters
30     ----------
31     n_workers: int
32         Number of workers to start
33     processes: bool
34         Whether to use processes (True) or threads (False).  Defaults to True
35     threads_per_worker: int
36         Number of threads per each worker
37     scheduler_port: int
38         Port of the scheduler.  8786 by default, use 0 to choose a random port
39     silence_logs: logging level
40         Level of logs to print out to stdout.  ``logging.CRITICAL`` by default.
41         Use a falsey value like False or None for no change.
42     ip: string
43         IP address on which the scheduler will listen, defaults to only localhost
44     kwargs: dict
45         Extra worker arguments, will be passed to the Worker constructor.
46 
47     Examples
48     --------
49     >>> c = LocalCluster()  # Create a local cluster with as many workers as cores  # doctest: +SKIP
50     >>> c  # doctest: +SKIP
51     LocalCluster("127.0.0.1:8786", workers=8, ncores=8)
52 
53     >>> c = Client(c)  # connect to local cluster  # doctest: +SKIP
54 
55     Add a new worker to the cluster
56     >>> w = c.start_worker(ncores=2)  # doctest: +SKIP
57 
58     Shut down the extra worker
59     >>> c.remove_worker(w)  # doctest: +SKIP
60     """
61     def __init__(self, n_workers=None, threads_per_worker=None, processes=True,
62                  loop=None, start=True, ip=None, scheduler_port=0,
63                  silence_logs=logging.CRITICAL, diagnostics_port=8787,
64                  services={}, worker_services={}, nanny=None, **worker_kwargs):
65         if nanny is not None:
66             warnings.warning("nanny has been deprecated, used processes=")
67             processes = nanny
68         self.status = None
69         self.processes = processes
70         self.silence_logs = silence_logs
71         if silence_logs:
72             silence_logging(level=silence_logs)
73         if n_workers is None and threads_per_worker is None:
74             if processes:
75                 n_workers = _ncores
76                 threads_per_worker = 1
77             else:
78                 n_workers = 1
79                 threads_per_worker = _ncores
80         if n_workers is None and threads_per_worker is not None:
81             n_workers = max(1, _ncores // threads_per_worker)
82         if n_workers and threads_per_worker is None:
83             # Overcommit threads per worker, rather than undercommit
84             threads_per_worker = max(1, int(math.ceil(_ncores / n_workers)))
85 
86         self.loop = loop or IOLoop()
87         if start and not self.loop._running:
88             self._thread = Thread(target=self.loop.start,
89                                   name="LocalCluster loop")
90             self._thread.daemon = True
91             self._thread.start()
92             while not self.loop._running:
93                 sleep(0.001)
94 
95         if diagnostics_port is not None:
96             try:
97                 from distributed.bokeh.scheduler import BokehScheduler
98                 from distributed.bokeh.worker import BokehWorker
99             except ImportError:
100                 logger.debug("To start diagnostics web server please install Bokeh")
101             else:
102                 services[('bokeh', diagnostics_port)] = BokehScheduler
103                 worker_services[('bokeh', 0)] = BokehWorker
104 
105         self.scheduler = Scheduler(loop=self.loop,
106                                    services=services)
107         self.scheduler_port = scheduler_port
108 
109         self.workers = []
110         self.n_workers = n_workers
111         self.threads_per_worker = threads_per_worker
112         self.worker_services = worker_services
113         self.worker_kwargs = worker_kwargs
114 
115         if start:
116             sync(self.loop, self._start, ip)
117 
118         clusters_to_close.add(self)
119 
120     def __str__(self):
121         return ('LocalCluster(%r, workers=%d, ncores=%d)' %
122                 (self.scheduler_address, len(self.workers),
123                  sum(w.ncores for w in self.workers))
124                 )
125 
126     __repr__ = __str__
127 
128     @gen.coroutine
129     def _start(self, ip=None):
130         """
131         Start all cluster services.
132         Wait on this if you passed `start=False` to the LocalCluster
133         constructor.
134         """
135         if self.status == 'running':
136             return
137         if ip is None and not self.scheduler_port and not self.processes:
138             # Use inproc transport for optimization
139             scheduler_address = 'inproc://'
140         else:
141             if ip is None:
142                 ip = '127.0.0.1'
143             scheduler_address = (ip, self.scheduler_port)
144         self.scheduler.start(scheduler_address)
145 
146         yield self._start_all_workers(
147             self.n_workers, ncores=self.threads_per_worker,
148             services=self.worker_services, **self.worker_kwargs)
149 
150         self.status = 'running'
151 
152     @gen.coroutine
153     def _start_all_workers(self, n_workers, **kwargs):
154         yield [self._start_worker(**kwargs) for i in range(n_workers)]
155 
156     @gen.coroutine
157     def _start_worker(self, port=0, processes=None, death_timeout=60, **kwargs):
158         if processes is not None:
159             raise ValueError("overriding `processes` for individual workers "
160                              "in a LocalCluster is not supported anymore")
161         if port:
162             raise ValueError("overriding `port` for individual workers "
163                              "in a LocalCluster is not supported anymore")
164         if self.processes:
165             W = Nanny
166             kwargs['quiet'] = True
167         else:
168             W = Worker
169 
170         w = W(self.scheduler.address, loop=self.loop,
171               death_timeout=death_timeout,
172               silence_logs=self.silence_logs, **kwargs)
173         yield w._start()
174 
175         self.workers.append(w)
176 
177         while w.worker_address not in self.scheduler.worker_info:
178             yield gen.sleep(0.01)
179 
180         raise gen.Return(w)
181 
182     def start_worker(self, ncores=0, **kwargs):
183         """ Add a new worker to the running cluster
184 
185         Parameters
186         ----------
187         port: int (optional)
188             Port on which to serve the worker, defaults to 0 or random
189         ncores: int (optional)
190             Number of threads to use.  Defaults to number of logical cores
191 
192         Examples
193         --------
194         >>> c = LocalCluster()  # doctest: +SKIP
195         >>> c.start_worker(ncores=2)  # doctest: +SKIP
196 
197         Returns
198         -------
199         The created Worker or Nanny object.  Can be discarded.
200         """
201         return sync(self.loop, self._start_worker, ncores=ncores, **kwargs)
202 
203     @gen.coroutine
204     def _stop_worker(self, w):
205         yield w._close()
206         if w in self.workers:
207             self.workers.remove(w)
208 
209     def stop_worker(self, w):
210         """ Stop a running worker
211 
212         Examples
213         --------
214         >>> c = LocalCluster()  # doctest: +SKIP
215         >>> w = c.start_worker(ncores=2)  # doctest: +SKIP
216         >>> c.stop_worker(w)  # doctest: +SKIP
217         """
218         sync(self.loop, self._stop_worker, w)
219 
220     @gen.coroutine
221     def _close(self):
222         if self.status == 'closed':
223             return
224 
225         with ignoring(gen.TimeoutError, CommClosedError, OSError):
226             yield All([w._close() for w in self.workers])
227         with ignoring(gen.TimeoutError, CommClosedError, OSError):
228             yield self.scheduler.close(fast=True)
229         del self.workers[:]
230         self.status = 'closed'
231 
232     def close(self):
233         """ Close the cluster """
234         if self.status == 'closed':
235             return
236 
237         for w in self.workers:
238             self.loop.add_callback(self._stop_worker, w)
239         for i in range(10):
240             if not self.workers:
241                 break
242             else:
243                 sleep(0.01)
244         if self.loop._running:
245             sync(self.loop, self._close)
246         if hasattr(self, '_thread'):
247             if self.loop._running:
248                 self.loop.add_callback(self.loop.stop)
249             try:
250                 self._thread.join(timeout=1)
251             finally:
252                 self.loop.close()
253             del self._thread
254 
255     @gen.coroutine
256     def scale_up(self, n, **kwargs):
257         """ Bring the total count of workers up to ``n``
258 
259         This function/coroutine should bring the total number of workers up to
260         the number ``n``.
261 
262         This can be implemented either as a function or as a Tornado coroutine.
263         """
264         yield [self._start_worker(**kwargs)
265                for i in range(n - len(self.workers))]
266 
267     @gen.coroutine
268     def scale_down(self, workers):
269         """ Remove ``workers`` from the cluster
270 
271         Given a list of worker addresses this function should remove those
272         workers from the cluster.  This may require tracking which jobs are
273         associated to which worker address.
274 
275         This can be implemented either as a function or as a Tornado coroutine.
276         """
277         workers = set(workers)
278         yield [self._stop_worker(w)
279                 for w in self.workers
280                 if w.worker_address in workers]
281         while workers & set(self.workers):
282             yield gen.sleep(0.01)
283 
284     def __del__(self):
285         self.close()
286 
287     def __enter__(self):
288         return self
289 
290     def __exit__(self, *args):
291         self.close()
292 
293     @property
294     def scheduler_address(self):
295         try:
296             return self.scheduler.address
297         except ValueError:
298             return '<unstarted>'
299 
300 
301 clusters_to_close = weakref.WeakSet()
302 
303 
304 @atexit.register
305 def close_clusters():
306     for cluster in clusters_to_close:
307         cluster.close()
308
```
--- END FILES ---

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

Here is an example:

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

```

diff --git a/distributed/deploy/local.py b/distributed/deploy/local.py
--- a/distributed/deploy/local.py
+++ b/distributed/deploy/local.py
@@ -174,9 +174,13 @@
 
         self.workers.append(w)
 
-        while w.worker_address not in self.scheduler.worker_info:
+        while w.status != 'closed' and w.worker_address not in self.scheduler.worker_info:
             yield gen.sleep(0.01)
 
+        if w.status == 'closed':
+            self.workers.remove(w)
+            raise gen.TimeoutError("Worker failed to start")
+
         raise gen.Return(w)
 
     def start_worker(self, ncores=0, **kwargs):
@@ -249,7 +253,10 @@
             try:
                 self._thread.join(timeout=1)
             finally:
-                self.loop.close()
+                try:
+                    self.loop.close()
+                except ValueError:
+                    pass
             del self._thread
 
     @gen.coroutine

{"golden_diff": "diff --git a/distributed/deploy/local.py b/distributed/deploy/local.py\n--- a/distributed/deploy/local.py\n+++ b/distributed/deploy/local.py\n@@ -174,9 +174,13 @@\n \n         self.workers.append(w)\n \n-        while w.worker_address not in self.scheduler.worker_info:\n+        while w.status != 'closed' and w.worker_address not in self.scheduler.worker_info:\n             yield gen.sleep(0.01)\n \n+        if w.status == 'closed':\n+            self.workers.remove(w)\n+            raise gen.TimeoutError(\"Worker failed to start\")\n+\n         raise gen.Return(w)\n \n     def start_worker(self, ncores=0, **kwargs):\n@@ -249,7 +253,10 @@\n             try:\n                 self._thread.join(timeout=1)\n             finally:\n-                self.loop.close()\n+                try:\n+                    self.loop.close()\n+                except ValueError:\n+                    pass\n             del self._thread\n \n     @gen.coroutine\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\nimport logging\nimport math\nfrom threading import Thread\nfrom time import sleep\nimport warnings\nimport weakref\n\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop\n\nfrom ..core import CommClosedError\nfrom ..utils import sync, ignoring, All, silence_logging\nfrom ..nanny import Nanny\nfrom ..scheduler import Scheduler\nfrom ..worker import Worker, _ncores\n\nlogger = logging.getLogger(__name__)\n\n\nclass LocalCluster(object):\n    \"\"\" Create local Scheduler and Workers\n\n    This creates a \"cluster\" of a scheduler and workers running on the local\n    machine.\n\n    Parameters\n    ----------\n    n_workers: int\n        Number of workers to start\n    processes: bool\n        Whether to use processes (True) or threads (False).  Defaults to True\n    threads_per_worker: int\n        Number of threads per each worker\n    scheduler_port: int\n        Port of the scheduler.  8786 by default, use 0 to choose a random port\n    silence_logs: logging level\n        Level of logs to print out to stdout.  ``logging.CRITICAL`` by default.\n        Use a falsey value like False or None for no change.\n    ip: string\n        IP address on which the scheduler will listen, defaults to only localhost\n    kwargs: dict\n        Extra worker arguments, will be passed to the Worker constructor.\n\n    Examples\n    --------\n    >>> c = LocalCluster()  # Create a local cluster with as many workers as cores  # doctest: +SKIP\n    >>> c  # doctest: +SKIP\n    LocalCluster(\"127.0.0.1:8786\", workers=8, ncores=8)\n\n    >>> c = Client(c)  # connect to local cluster  # doctest: +SKIP\n\n    Add a new worker to the cluster\n    >>> w = c.start_worker(ncores=2)  # doctest: +SKIP\n\n    Shut down the extra worker\n    >>> c.remove_worker(w)  # doctest: +SKIP\n    \"\"\"\n    def __init__(self, n_workers=None, threads_per_worker=None, processes=True,\n                 loop=None, start=True, ip=None, scheduler_port=0,\n                 silence_logs=logging.CRITICAL, diagnostics_port=8787,\n                 services={}, worker_services={}, nanny=None, **worker_kwargs):\n        if nanny is not None:\n            warnings.warning(\"nanny has been deprecated, used processes=\")\n            processes = nanny\n        self.status = None\n        self.processes = processes\n        self.silence_logs = silence_logs\n        if silence_logs:\n            silence_logging(level=silence_logs)\n        if n_workers is None and threads_per_worker is None:\n            if processes:\n                n_workers = _ncores\n                threads_per_worker = 1\n            else:\n                n_workers = 1\n                threads_per_worker = _ncores\n        if n_workers is None and threads_per_worker is not None:\n            n_workers = max(1, _ncores // threads_per_worker)\n        if n_workers and threads_per_worker is None:\n            # Overcommit threads per worker, rather than undercommit\n            threads_per_worker = max(1, int(math.ceil(_ncores / n_workers)))\n\n        self.loop = loop or IOLoop()\n        if start and not self.loop._running:\n            self._thread = Thread(target=self.loop.start,\n                                  name=\"LocalCluster loop\")\n            self._thread.daemon = True\n            self._thread.start()\n            while not self.loop._running:\n                sleep(0.001)\n\n        if diagnostics_port is not None:\n            try:\n                from distributed.bokeh.scheduler import BokehScheduler\n                from distributed.bokeh.worker import BokehWorker\n            except ImportError:\n                logger.debug(\"To start diagnostics web server please install Bokeh\")\n            else:\n                services[('bokeh', diagnostics_port)] = BokehScheduler\n                worker_services[('bokeh', 0)] = BokehWorker\n\n        self.scheduler = Scheduler(loop=self.loop,\n                                   services=services)\n        self.scheduler_port = scheduler_port\n\n        self.workers = []\n        self.n_workers = n_workers\n        self.threads_per_worker = threads_per_worker\n        self.worker_services = worker_services\n        self.worker_kwargs = worker_kwargs\n\n        if start:\n            sync(self.loop, self._start, ip)\n\n        clusters_to_close.add(self)\n\n    def __str__(self):\n        return ('LocalCluster(%r, workers=%d, ncores=%d)' %\n                (self.scheduler_address, len(self.workers),\n                 sum(w.ncores for w in self.workers))\n                )\n\n    __repr__ = __str__\n\n    @gen.coroutine\n    def _start(self, ip=None):\n        \"\"\"\n        Start all cluster services.\n        Wait on this if you passed `start=False` to the LocalCluster\n        constructor.\n        \"\"\"\n        if self.status == 'running':\n            return\n        if ip is None and not self.scheduler_port and not self.processes:\n            # Use inproc transport for optimization\n            scheduler_address = 'inproc://'\n        else:\n            if ip is None:\n                ip = '127.0.0.1'\n            scheduler_address = (ip, self.scheduler_port)\n        self.scheduler.start(scheduler_address)\n\n        yield self._start_all_workers(\n            self.n_workers, ncores=self.threads_per_worker,\n            services=self.worker_services, **self.worker_kwargs)\n\n        self.status = 'running'\n\n    @gen.coroutine\n    def _start_all_workers(self, n_workers, **kwargs):\n        yield [self._start_worker(**kwargs) for i in range(n_workers)]\n\n    @gen.coroutine\n    def _start_worker(self, port=0, processes=None, death_timeout=60, **kwargs):\n        if processes is not None:\n            raise ValueError(\"overriding `processes` for individual workers \"\n                             \"in a LocalCluster is not supported anymore\")\n        if port:\n            raise ValueError(\"overriding `port` for individual workers \"\n                             \"in a LocalCluster is not supported anymore\")\n        if self.processes:\n            W = Nanny\n            kwargs['quiet'] = True\n        else:\n            W = Worker\n\n        w = W(self.scheduler.address, loop=self.loop,\n              death_timeout=death_timeout,\n              silence_logs=self.silence_logs, **kwargs)\n        yield w._start()\n\n        self.workers.append(w)\n\n        while w.worker_address not in self.scheduler.worker_info:\n            yield gen.sleep(0.01)\n\n        raise gen.Return(w)\n\n    def start_worker(self, ncores=0, **kwargs):\n        \"\"\" Add a new worker to the running cluster\n\n        Parameters\n        ----------\n        port: int (optional)\n            Port on which to serve the worker, defaults to 0 or random\n        ncores: int (optional)\n            Number of threads to use.  Defaults to number of logical cores\n\n        Examples\n        --------\n        >>> c = LocalCluster()  # doctest: +SKIP\n        >>> c.start_worker(ncores=2)  # doctest: +SKIP\n\n        Returns\n        -------\n        The created Worker or Nanny object.  Can be discarded.\n        \"\"\"\n        return sync(self.loop, self._start_worker, ncores=ncores, **kwargs)\n\n    @gen.coroutine\n    def _stop_worker(self, w):\n        yield w._close()\n        if w in self.workers:\n            self.workers.remove(w)\n\n    def stop_worker(self, w):\n        \"\"\" Stop a running worker\n\n        Examples\n        --------\n        >>> c = LocalCluster()  # doctest: +SKIP\n        >>> w = c.start_worker(ncores=2)  # doctest: +SKIP\n        >>> c.stop_worker(w)  # doctest: +SKIP\n        \"\"\"\n        sync(self.loop, self._stop_worker, w)\n\n    @gen.coroutine\n    def _close(self):\n        if self.status == 'closed':\n            return\n\n        with ignoring(gen.TimeoutError, CommClosedError, OSError):\n            yield All([w._close() for w in self.workers])\n        with ignoring(gen.TimeoutError, CommClosedError, OSError):\n            yield self.scheduler.close(fast=True)\n        del self.workers[:]\n        self.status = 'closed'\n\n    def close(self):\n        \"\"\" Close the cluster \"\"\"\n        if self.status == 'closed':\n            return\n\n        for w in self.workers:\n            self.loop.add_callback(self._stop_worker, w)\n        for i in range(10):\n            if not self.workers:\n                break\n            else:\n                sleep(0.01)\n        if self.loop._running:\n            sync(self.loop, self._close)\n        if hasattr(self, '_thread'):\n            if self.loop._running:\n                self.loop.add_callback(self.loop.stop)\n            try:\n                self._thread.join(timeout=1)\n            finally:\n                self.loop.close()\n            del self._thread\n\n    @gen.coroutine\n    def scale_up(self, n, **kwargs):\n        \"\"\" Bring the total count of workers up to ``n``\n\n        This function/coroutine should bring the total number of workers up to\n        the number ``n``.\n\n        This can be implemented either as a function or as a Tornado coroutine.\n        \"\"\"\n        yield [self._start_worker(**kwargs)\n               for i in range(n - len(self.workers))]\n\n    @gen.coroutine\n    def scale_down(self, workers):\n        \"\"\" Remove ``workers`` from the cluster\n\n        Given a list of worker addresses this function should remove those\n        workers from the cluster.  This may require tracking which jobs are\n        associated to which worker address.\n\n        This can be implemented either as a function or as a Tornado coroutine.\n        \"\"\"\n        workers = set(workers)\n        yield [self._stop_worker(w)\n                for w in self.workers\n                if w.worker_address in workers]\n        while workers & set(self.workers):\n            yield gen.sleep(0.01)\n\n    def __del__(self):\n        self.close()\n\n    def __enter__(self):\n        return self\n\n    def __exit__(self, *args):\n        self.close()\n\n    @property\n    def scheduler_address(self):\n        try:\n            return self.scheduler.address\n        except ValueError:\n            return '<unstarted>'\n\n\nclusters_to_close = weakref.WeakSet()\n\n\n@atexit.register\ndef close_clusters():\n    for cluster in clusters_to_close:\n        cluster.close()\n", "path": "distributed/deploy/local.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport atexit\nimport logging\nimport math\nfrom threading import Thread\nfrom time import sleep\nimport warnings\nimport weakref\n\nfrom tornado import gen\nfrom tornado.ioloop import IOLoop\n\nfrom ..core import CommClosedError\nfrom ..utils import sync, ignoring, All, silence_logging\nfrom ..nanny import Nanny\nfrom ..scheduler import Scheduler\nfrom ..worker import Worker, _ncores\n\nlogger = logging.getLogger(__name__)\n\n\nclass LocalCluster(object):\n    \"\"\" Create local Scheduler and Workers\n\n    This creates a \"cluster\" of a scheduler and workers running on the local\n    machine.\n\n    Parameters\n    ----------\n    n_workers: int\n        Number of workers to start\n    processes: bool\n        Whether to use processes (True) or threads (False).  Defaults to True\n    threads_per_worker: int\n        Number of threads per each worker\n    scheduler_port: int\n        Port of the scheduler.  8786 by default, use 0 to choose a random port\n    silence_logs: logging level\n        Level of logs to print out to stdout.  ``logging.CRITICAL`` by default.\n        Use a falsey value like False or None for no change.\n    ip: string\n        IP address on which the scheduler will listen, defaults to only localhost\n    kwargs: dict\n        Extra worker arguments, will be passed to the Worker constructor.\n\n    Examples\n    --------\n    >>> c = LocalCluster()  # Create a local cluster with as many workers as cores  # doctest: +SKIP\n    >>> c  # doctest: +SKIP\n    LocalCluster(\"127.0.0.1:8786\", workers=8, ncores=8)\n\n    >>> c = Client(c)  # connect to local cluster  # doctest: +SKIP\n\n    Add a new worker to the cluster\n    >>> w = c.start_worker(ncores=2)  # doctest: +SKIP\n\n    Shut down the extra worker\n    >>> c.remove_worker(w)  # doctest: +SKIP\n    \"\"\"\n    def __init__(self, n_workers=None, threads_per_worker=None, processes=True,\n                 loop=None, start=True, ip=None, scheduler_port=0,\n                 silence_logs=logging.CRITICAL, diagnostics_port=8787,\n                 services={}, worker_services={}, nanny=None, **worker_kwargs):\n        if nanny is not None:\n            warnings.warning(\"nanny has been deprecated, used processes=\")\n            processes = nanny\n        self.status = None\n        self.processes = processes\n        self.silence_logs = silence_logs\n        if silence_logs:\n            silence_logging(level=silence_logs)\n        if n_workers is None and threads_per_worker is None:\n            if processes:\n                n_workers = _ncores\n                threads_per_worker = 1\n            else:\n                n_workers = 1\n                threads_per_worker = _ncores\n        if n_workers is None and threads_per_worker is not None:\n            n_workers = max(1, _ncores // threads_per_worker)\n        if n_workers and threads_per_worker is None:\n            # Overcommit threads per worker, rather than undercommit\n            threads_per_worker = max(1, int(math.ceil(_ncores / n_workers)))\n\n        self.loop = loop or IOLoop()\n        if start and not self.loop._running:\n            self._thread = Thread(target=self.loop.start,\n                                  name=\"LocalCluster loop\")\n            self._thread.daemon = True\n            self._thread.start()\n            while not self.loop._running:\n                sleep(0.001)\n\n        if diagnostics_port is not None:\n            try:\n                from distributed.bokeh.scheduler import BokehScheduler\n                from distributed.bokeh.worker import BokehWorker\n            except ImportError:\n                logger.debug(\"To start diagnostics web server please install Bokeh\")\n            else:\n                services[('bokeh', diagnostics_port)] = BokehScheduler\n                worker_services[('bokeh', 0)] = BokehWorker\n\n        self.scheduler = Scheduler(loop=self.loop,\n                                   services=services)\n        self.scheduler_port = scheduler_port\n\n        self.workers = []\n        self.n_workers = n_workers\n        self.threads_per_worker = threads_per_worker\n        self.worker_services = worker_services\n        self.worker_kwargs = worker_kwargs\n\n        if start:\n            sync(self.loop, self._start, ip)\n\n        clusters_to_close.add(self)\n\n    def __str__(self):\n        return ('LocalCluster(%r, workers=%d, ncores=%d)' %\n                (self.scheduler_address, len(self.workers),\n                 sum(w.ncores for w in self.workers))\n                )\n\n    __repr__ = __str__\n\n    @gen.coroutine\n    def _start(self, ip=None):\n        \"\"\"\n        Start all cluster services.\n        Wait on this if you passed `start=False` to the LocalCluster\n        constructor.\n        \"\"\"\n        if self.status == 'running':\n            return\n        if ip is None and not self.scheduler_port and not self.processes:\n            # Use inproc transport for optimization\n            scheduler_address = 'inproc://'\n        else:\n            if ip is None:\n                ip = '127.0.0.1'\n            scheduler_address = (ip, self.scheduler_port)\n        self.scheduler.start(scheduler_address)\n\n        yield self._start_all_workers(\n            self.n_workers, ncores=self.threads_per_worker,\n            services=self.worker_services, **self.worker_kwargs)\n\n        self.status = 'running'\n\n    @gen.coroutine\n    def _start_all_workers(self, n_workers, **kwargs):\n        yield [self._start_worker(**kwargs) for i in range(n_workers)]\n\n    @gen.coroutine\n    def _start_worker(self, port=0, processes=None, death_timeout=60, **kwargs):\n        if processes is not None:\n            raise ValueError(\"overriding `processes` for individual workers \"\n                             \"in a LocalCluster is not supported anymore\")\n        if port:\n            raise ValueError(\"overriding `port` for individual workers \"\n                             \"in a LocalCluster is not supported anymore\")\n        if self.processes:\n            W = Nanny\n            kwargs['quiet'] = True\n        else:\n            W = Worker\n\n        w = W(self.scheduler.address, loop=self.loop,\n              death_timeout=death_timeout,\n              silence_logs=self.silence_logs, **kwargs)\n        yield w._start()\n\n        self.workers.append(w)\n\n        while w.status != 'closed' and w.worker_address not in self.scheduler.worker_info:\n            yield gen.sleep(0.01)\n\n        if w.status == 'closed':\n            self.workers.remove(w)\n            raise gen.TimeoutError(\"Worker failed to start\")\n\n        raise gen.Return(w)\n\n    def start_worker(self, ncores=0, **kwargs):\n        \"\"\" Add a new worker to the running cluster\n\n        Parameters\n        ----------\n        port: int (optional)\n            Port on which to serve the worker, defaults to 0 or random\n        ncores: int (optional)\n            Number of threads to use.  Defaults to number of logical cores\n\n        Examples\n        --------\n        >>> c = LocalCluster()  # doctest: +SKIP\n        >>> c.start_worker(ncores=2)  # doctest: +SKIP\n\n        Returns\n        -------\n        The created Worker or Nanny object.  Can be discarded.\n        \"\"\"\n        return sync(self.loop, self._start_worker, ncores=ncores, **kwargs)\n\n    @gen.coroutine\n    def _stop_worker(self, w):\n        yield w._close()\n        if w in self.workers:\n            self.workers.remove(w)\n\n    def stop_worker(self, w):\n        \"\"\" Stop a running worker\n\n        Examples\n        --------\n        >>> c = LocalCluster()  # doctest: +SKIP\n        >>> w = c.start_worker(ncores=2)  # doctest: +SKIP\n        >>> c.stop_worker(w)  # doctest: +SKIP\n        \"\"\"\n        sync(self.loop, self._stop_worker, w)\n\n    @gen.coroutine\n    def _close(self):\n        if self.status == 'closed':\n            return\n\n        with ignoring(gen.TimeoutError, CommClosedError, OSError):\n            yield All([w._close() for w in self.workers])\n        with ignoring(gen.TimeoutError, CommClosedError, OSError):\n            yield self.scheduler.close(fast=True)\n        del self.workers[:]\n        self.status = 'closed'\n\n    def close(self):\n        \"\"\" Close the cluster \"\"\"\n        if self.status == 'closed':\n            return\n\n        for w in self.workers:\n            self.loop.add_callback(self._stop_worker, w)\n        for i in range(10):\n            if not self.workers:\n                break\n            else:\n                sleep(0.01)\n        if self.loop._running:\n            sync(self.loop, self._close)\n        if hasattr(self, '_thread'):\n            if self.loop._running:\n                self.loop.add_callback(self.loop.stop)\n            try:\n                self._thread.join(timeout=1)\n            finally:\n                try:\n                    self.loop.close()\n                except ValueError:\n                    pass\n            del self._thread\n\n    @gen.coroutine\n    def scale_up(self, n, **kwargs):\n        \"\"\" Bring the total count of workers up to ``n``\n\n        This function/coroutine should bring the total number of workers up to\n        the number ``n``.\n\n        This can be implemented either as a function or as a Tornado coroutine.\n        \"\"\"\n        yield [self._start_worker(**kwargs)\n               for i in range(n - len(self.workers))]\n\n    @gen.coroutine\n    def scale_down(self, workers):\n        \"\"\" Remove ``workers`` from the cluster\n\n        Given a list of worker addresses this function should remove those\n        workers from the cluster.  This may require tracking which jobs are\n        associated to which worker address.\n\n        This can be implemented either as a function or as a Tornado coroutine.\n        \"\"\"\n        workers = set(workers)\n        yield [self._stop_worker(w)\n                for w in self.workers\n                if w.worker_address in workers]\n        while workers & set(self.workers):\n            yield gen.sleep(0.01)\n\n    def __del__(self):\n        self.close()\n\n    def __enter__(self):\n        return self\n\n    def __exit__(self, *args):\n        self.close()\n\n    @property\n    def scheduler_address(self):\n        try:\n            return self.scheduler.address\n        except ValueError:\n            return '<unstarted>'\n\n\nclusters_to_close = weakref.WeakSet()\n\n\n@atexit.register\ndef close_clusters():\n    for cluster in clusters_to_close:\n        cluster.close()\n", "path": "distributed/deploy/local.py"}]}