We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Failure on a 3-core VM in distributed\deploy\tests\test_local.py
```
________________________________ test_defaults ________________________________
distributed\deploy\tests\test_local.py:93: in test_defaults
    assert sum(w.ncores for w in c.workers) == max(2, _ncores)
E   assert 2 == 3
E    +  where 2 = sum(<generator object test_defaults.<locals>.<genexpr> at 0x00
000000084A6048>)
E    +  and   3 = max(2, 3)
```


--- 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 logging
4 from threading import Thread
5 from time import sleep
6 
7 from tornado.ioloop import IOLoop
8 from tornado.iostream import StreamClosedError
9 from tornado import gen
10 
11 from ..http.scheduler import HTTPScheduler
12 from ..utils import sync, ignoring, All
13 from ..client import Client
14 from ..nanny import Nanny
15 from ..scheduler import Scheduler
16 from ..worker import Worker, _ncores
17 
18 logger = logging.getLogger(__file__)
19 
20 
21 class LocalCluster(object):
22     """ Create local Scheduler and Workers
23 
24     This creates a "cluster" of a scheduler and workers running on the local
25     machine.
26 
27     Parameters
28     ----------
29     n_workers: int
30         Number of workers to start
31     threads_per_worker: int
32         Number of threads per each worker
33     nanny: boolean
34         If true start the workers in separate processes managed by a nanny.
35         If False keep the workers in the main calling process
36     scheduler_port: int
37         Port of the scheduler.  8786 by default, use 0 to choose a random port
38 
39     Examples
40     --------
41     >>> c = LocalCluster()  # Create a local cluster with as many workers as cores  # doctest: +SKIP
42     >>> c  # doctest: +SKIP
43     LocalCluster("127.0.0.1:8786", workers=8, ncores=8)
44 
45     >>> c = Client(c)  # connect to local cluster  # doctest: +SKIP
46 
47     Add a new worker to the cluster
48     >>> w = c.start_worker(ncores=2)  # doctest: +SKIP
49 
50     Shut down the extra worker
51     >>> c.remove_worker(w)  # doctest: +SKIP
52 
53     Start a diagnostic web server and open a new browser tab
54     >>> c.start_diagnostics_server(show=True)  # doctest: +SKIP
55     """
56     def __init__(self, n_workers=None, threads_per_worker=None, nanny=True,
57             loop=None, start=True, scheduler_port=8786,
58             silence_logs=logging.CRITICAL, diagnostics_port=8787,
59             services={'http': HTTPScheduler}, **kwargs):
60         self.status = None
61         self.nanny = nanny
62         if silence_logs:
63             for l in ['distributed.scheduler',
64                       'distributed.worker',
65                       'distributed.core',
66                       'distributed.nanny']:
67                 logging.getLogger(l).setLevel(silence_logs)
68         if n_workers is None and threads_per_worker is None:
69             if nanny:
70                 n_workers = _ncores
71                 threads_per_worker = 1
72             else:
73                 n_workers = 1
74                 threads_per_worker = _ncores
75         if n_workers is None and threads_per_worker is not None:
76             n_workers = max(1, _ncores // threads_per_worker)
77         if n_workers and threads_per_worker is None:
78             threads_per_worker = max(1, _ncores // n_workers)
79 
80         self.loop = loop or IOLoop()
81         if not self.loop._running:
82             self._thread = Thread(target=self.loop.start)
83             self._thread.daemon = True
84             self._thread.start()
85             while not self.loop._running:
86                 sleep(0.001)
87 
88         self.scheduler = Scheduler(loop=self.loop, ip='127.0.0.1',
89                                    services=services)
90         self.scheduler.start(scheduler_port)
91         self.workers = []
92 
93         if start:
94             _start_worker = self.start_worker
95         else:
96             _start_worker = lambda *args, **kwargs: self.loop.add_callback(self._start_worker, *args, **kwargs)
97         for i in range(n_workers):
98             _start_worker(ncores=threads_per_worker, nanny=nanny)
99         self.status = 'running'
100 
101         self.diagnostics = None
102         if diagnostics_port is not None:
103             self.start_diagnostics_server(diagnostics_port,
104                                           silence=silence_logs)
105 
106     def __str__(self):
107         return 'LocalCluster("%s", workers=%d, ncores=%d)' % (
108                 self.scheduler_address,
109                 len(self.workers),
110                 sum(w.ncores for w in self.workers))
111 
112     __repr__ = __str__
113 
114     @gen.coroutine
115     def _start_worker(self, port=0, nanny=None, **kwargs):
116         if nanny is None:
117             nanny = self.nanny
118         if nanny:
119             W = Nanny
120             kwargs['quiet'] = True
121         else:
122             W = Worker
123         w = W(self.scheduler.ip, self.scheduler.port, loop=self.loop, **kwargs)
124         yield w._start(port)
125 
126         self.workers.append(w)
127 
128         while w.worker_address not in self.scheduler.worker_info:
129             yield gen.sleep(0.01)
130 
131         raise gen.Return(w)
132 
133     def start_worker(self, port=0, ncores=0, **kwargs):
134         """ Add a new worker to the running cluster
135 
136         Parameters
137         ----------
138         port: int (optional)
139             Port on which to serve the worker, defaults to 0 or random
140         ncores: int (optional)
141             Number of threads to use.  Defaults to number of logical cores
142         nanny: boolean
143             If true start worker in separate process managed by a nanny
144 
145         Examples
146         --------
147         >>> c = LocalCluster()  # doctest: +SKIP
148         >>> c.start_worker(ncores=2)  # doctest: +SKIP
149 
150         Returns
151         -------
152         The created Worker or Nanny object.  Can be discarded.
153         """
154         return sync(self.loop, self._start_worker, port, ncores=ncores, **kwargs)
155 
156     @gen.coroutine
157     def _stop_worker(self, w):
158         yield w._close()
159         self.workers.remove(w)
160 
161     def stop_worker(self, w):
162         """ Stop a running worker
163 
164         Examples
165         --------
166         >>> c = LocalCluster()  # doctest: +SKIP
167         >>> w = c.start_worker(ncores=2)  # doctest: +SKIP
168         >>> c.stop_worker(w)  # doctest: +SKIP
169         """
170         sync(self.loop, self._stop_worker, w)
171 
172     def start_diagnostics_server(self, port=8787, show=False,
173             silence=logging.CRITICAL):
174         """ Start Diagnostics Web Server
175 
176         This starts a web application to show diagnostics of what is happening
177         on the cluster.  This application runs in a separate process and is
178         generally available at the following location:
179 
180             http://localhost:8787/status/
181         """
182         try:
183             from distributed.bokeh.application import BokehWebInterface
184         except ImportError:
185             logger.info("To start diagnostics web server please install Bokeh")
186             return
187 
188         assert self.diagnostics is None
189         if 'http' not in self.scheduler.services:
190             self.scheduler.services['http'] = HTTPScheduler(self.scheduler,
191                     io_loop=self.scheduler.loop)
192             self.scheduler.services['http'].listen(0)
193         self.diagnostics = BokehWebInterface(
194                 tcp_port=self.scheduler.port,
195                 http_port=self.scheduler.services['http'].port,
196                 bokeh_port=port, show=show,
197                 log_level=logging.getLevelName(silence).lower())
198 
199     @gen.coroutine
200     def _close(self):
201         with ignoring(gen.TimeoutError, StreamClosedError, OSError):
202             yield All([w._close() for w in self.workers])
203         with ignoring(gen.TimeoutError, StreamClosedError, OSError):
204             yield self.scheduler.close(fast=True)
205         del self.workers[:]
206         if self.diagnostics:
207             self.diagnostics.close()
208 
209     def close(self):
210         """ Close the cluster """
211         if self.status == 'running':
212             self.status = 'closed'
213             if self.loop._running:
214                 sync(self.loop, self._close)
215             if hasattr(self, '_thread'):
216                 sync(self.loop, self.loop.stop)
217                 self._thread.join(timeout=1)
218                 self.loop.close()
219                 del self._thread
220 
221     @gen.coroutine
222     def scale_up(self, n, **kwargs):
223         """ Bring the total count of workers up to ``n``
224 
225         This function/coroutine should bring the total number of workers up to
226         the number ``n``.
227 
228         This can be implemented either as a function or as a Tornado coroutine.
229         """
230         yield [self._start_worker(**kwargs)
231                 for i in range(n - len(self.workers))]
232 
233     @gen.coroutine
234     def scale_down(self, workers):
235         """ Remove ``workers`` from the cluster
236 
237         Given a list of worker addresses this function should remove those
238         workers from the cluster.  This may require tracking which jobs are
239         associated to which worker address.
240 
241         This can be implemented either as a function or as a Tornado coroutine.
242         """
243         workers = set(workers)
244         yield [self._stop_worker(w)
245                 for w in self.workers
246                 if w.worker_address in workers]
247         while workers & set(self.workers):
248             yield gen.sleep(0.01)
249 
250     def __del__(self):
251         self.close()
252 
253     def __enter__(self):
254         return self
255 
256     def __exit__(self, *args):
257         self.close()
258 
259     @property
260     def scheduler_address(self):
261         return self.scheduler.address
262
```
--- 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
@@ -1,6 +1,7 @@
 from __future__ import print_function, division, absolute_import
 
 import logging
+import math
 from threading import Thread
 from time import sleep
 
@@ -75,7 +76,8 @@
         if n_workers is None and threads_per_worker is not None:
             n_workers = max(1, _ncores // threads_per_worker)
         if n_workers and threads_per_worker is None:
-            threads_per_worker = max(1, _ncores // n_workers)
+            # Overcommit threads per worker, rather than undercommit
+            threads_per_worker = max(1, int(math.ceil(_ncores / n_workers)))
 
         self.loop = loop or IOLoop()
         if not self.loop._running:

{"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@@ -1,6 +1,7 @@\n from __future__ import print_function, division, absolute_import\n \n import logging\n+import math\n from threading import Thread\n from time import sleep\n \n@@ -75,7 +76,8 @@\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-            threads_per_worker = max(1, _ncores // n_workers)\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 not self.loop._running:\n", "issue": "Failure on a 3-core VM in distributed\\deploy\\tests\\test_local.py\n```\n________________________________ test_defaults ________________________________\ndistributed\\deploy\\tests\\test_local.py:93: in test_defaults\n    assert sum(w.ncores for w in c.workers) == max(2, _ncores)\nE   assert 2 == 3\nE    +  where 2 = sum(<generator object test_defaults.<locals>.<genexpr> at 0x00\n000000084A6048>)\nE    +  and   3 = max(2, 3)\n```\n\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport logging\nfrom threading import Thread\nfrom time import sleep\n\nfrom tornado.ioloop import IOLoop\nfrom tornado.iostream import StreamClosedError\nfrom tornado import gen\n\nfrom ..http.scheduler import HTTPScheduler\nfrom ..utils import sync, ignoring, All\nfrom ..client import Client\nfrom ..nanny import Nanny\nfrom ..scheduler import Scheduler\nfrom ..worker import Worker, _ncores\n\nlogger = logging.getLogger(__file__)\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    threads_per_worker: int\n        Number of threads per each worker\n    nanny: boolean\n        If true start the workers in separate processes managed by a nanny.\n        If False keep the workers in the main calling process\n    scheduler_port: int\n        Port of the scheduler.  8786 by default, use 0 to choose a random port\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    Start a diagnostic web server and open a new browser tab\n    >>> c.start_diagnostics_server(show=True)  # doctest: +SKIP\n    \"\"\"\n    def __init__(self, n_workers=None, threads_per_worker=None, nanny=True,\n            loop=None, start=True, scheduler_port=8786,\n            silence_logs=logging.CRITICAL, diagnostics_port=8787,\n            services={'http': HTTPScheduler}, **kwargs):\n        self.status = None\n        self.nanny = nanny\n        if silence_logs:\n            for l in ['distributed.scheduler',\n                      'distributed.worker',\n                      'distributed.core',\n                      'distributed.nanny']:\n                logging.getLogger(l).setLevel(silence_logs)\n        if n_workers is None and threads_per_worker is None:\n            if nanny:\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            threads_per_worker = max(1, _ncores // n_workers)\n\n        self.loop = loop or IOLoop()\n        if not self.loop._running:\n            self._thread = Thread(target=self.loop.start)\n            self._thread.daemon = True\n            self._thread.start()\n            while not self.loop._running:\n                sleep(0.001)\n\n        self.scheduler = Scheduler(loop=self.loop, ip='127.0.0.1',\n                                   services=services)\n        self.scheduler.start(scheduler_port)\n        self.workers = []\n\n        if start:\n            _start_worker = self.start_worker\n        else:\n            _start_worker = lambda *args, **kwargs: self.loop.add_callback(self._start_worker, *args, **kwargs)\n        for i in range(n_workers):\n            _start_worker(ncores=threads_per_worker, nanny=nanny)\n        self.status = 'running'\n\n        self.diagnostics = None\n        if diagnostics_port is not None:\n            self.start_diagnostics_server(diagnostics_port,\n                                          silence=silence_logs)\n\n    def __str__(self):\n        return 'LocalCluster(\"%s\", workers=%d, ncores=%d)' % (\n                self.scheduler_address,\n                len(self.workers),\n                sum(w.ncores for w in self.workers))\n\n    __repr__ = __str__\n\n    @gen.coroutine\n    def _start_worker(self, port=0, nanny=None, **kwargs):\n        if nanny is None:\n            nanny = self.nanny\n        if nanny:\n            W = Nanny\n            kwargs['quiet'] = True\n        else:\n            W = Worker\n        w = W(self.scheduler.ip, self.scheduler.port, loop=self.loop, **kwargs)\n        yield w._start(port)\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, port=0, 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        nanny: boolean\n            If true start worker in separate process managed by a nanny\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, port, ncores=ncores, **kwargs)\n\n    @gen.coroutine\n    def _stop_worker(self, w):\n        yield w._close()\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    def start_diagnostics_server(self, port=8787, show=False,\n            silence=logging.CRITICAL):\n        \"\"\" Start Diagnostics Web Server\n\n        This starts a web application to show diagnostics of what is happening\n        on the cluster.  This application runs in a separate process and is\n        generally available at the following location:\n\n            http://localhost:8787/status/\n        \"\"\"\n        try:\n            from distributed.bokeh.application import BokehWebInterface\n        except ImportError:\n            logger.info(\"To start diagnostics web server please install Bokeh\")\n            return\n\n        assert self.diagnostics is None\n        if 'http' not in self.scheduler.services:\n            self.scheduler.services['http'] = HTTPScheduler(self.scheduler,\n                    io_loop=self.scheduler.loop)\n            self.scheduler.services['http'].listen(0)\n        self.diagnostics = BokehWebInterface(\n                tcp_port=self.scheduler.port,\n                http_port=self.scheduler.services['http'].port,\n                bokeh_port=port, show=show,\n                log_level=logging.getLevelName(silence).lower())\n\n    @gen.coroutine\n    def _close(self):\n        with ignoring(gen.TimeoutError, StreamClosedError, OSError):\n            yield All([w._close() for w in self.workers])\n        with ignoring(gen.TimeoutError, StreamClosedError, OSError):\n            yield self.scheduler.close(fast=True)\n        del self.workers[:]\n        if self.diagnostics:\n            self.diagnostics.close()\n\n    def close(self):\n        \"\"\" Close the cluster \"\"\"\n        if self.status == 'running':\n            self.status = 'closed'\n            if self.loop._running:\n                sync(self.loop, self._close)\n            if hasattr(self, '_thread'):\n                sync(self.loop, self.loop.stop)\n                self._thread.join(timeout=1)\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        return self.scheduler.address\n", "path": "distributed/deploy/local.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nimport logging\nimport math\nfrom threading import Thread\nfrom time import sleep\n\nfrom tornado.ioloop import IOLoop\nfrom tornado.iostream import StreamClosedError\nfrom tornado import gen\n\nfrom ..http.scheduler import HTTPScheduler\nfrom ..utils import sync, ignoring, All\nfrom ..client import Client\nfrom ..nanny import Nanny\nfrom ..scheduler import Scheduler\nfrom ..worker import Worker, _ncores\n\nlogger = logging.getLogger(__file__)\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    threads_per_worker: int\n        Number of threads per each worker\n    nanny: boolean\n        If true start the workers in separate processes managed by a nanny.\n        If False keep the workers in the main calling process\n    scheduler_port: int\n        Port of the scheduler.  8786 by default, use 0 to choose a random port\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    Start a diagnostic web server and open a new browser tab\n    >>> c.start_diagnostics_server(show=True)  # doctest: +SKIP\n    \"\"\"\n    def __init__(self, n_workers=None, threads_per_worker=None, nanny=True,\n            loop=None, start=True, scheduler_port=8786,\n            silence_logs=logging.CRITICAL, diagnostics_port=8787,\n            services={'http': HTTPScheduler}, **kwargs):\n        self.status = None\n        self.nanny = nanny\n        if silence_logs:\n            for l in ['distributed.scheduler',\n                      'distributed.worker',\n                      'distributed.core',\n                      'distributed.nanny']:\n                logging.getLogger(l).setLevel(silence_logs)\n        if n_workers is None and threads_per_worker is None:\n            if nanny:\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 not self.loop._running:\n            self._thread = Thread(target=self.loop.start)\n            self._thread.daemon = True\n            self._thread.start()\n            while not self.loop._running:\n                sleep(0.001)\n\n        self.scheduler = Scheduler(loop=self.loop, ip='127.0.0.1',\n                                   services=services)\n        self.scheduler.start(scheduler_port)\n        self.workers = []\n\n        if start:\n            _start_worker = self.start_worker\n        else:\n            _start_worker = lambda *args, **kwargs: self.loop.add_callback(self._start_worker, *args, **kwargs)\n        for i in range(n_workers):\n            _start_worker(ncores=threads_per_worker, nanny=nanny)\n        self.status = 'running'\n\n        self.diagnostics = None\n        if diagnostics_port is not None:\n            self.start_diagnostics_server(diagnostics_port,\n                                          silence=silence_logs)\n\n    def __str__(self):\n        return 'LocalCluster(\"%s\", workers=%d, ncores=%d)' % (\n                self.scheduler_address,\n                len(self.workers),\n                sum(w.ncores for w in self.workers))\n\n    __repr__ = __str__\n\n    @gen.coroutine\n    def _start_worker(self, port=0, nanny=None, **kwargs):\n        if nanny is None:\n            nanny = self.nanny\n        if nanny:\n            W = Nanny\n            kwargs['quiet'] = True\n        else:\n            W = Worker\n        w = W(self.scheduler.ip, self.scheduler.port, loop=self.loop, **kwargs)\n        yield w._start(port)\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, port=0, 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        nanny: boolean\n            If true start worker in separate process managed by a nanny\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, port, ncores=ncores, **kwargs)\n\n    @gen.coroutine\n    def _stop_worker(self, w):\n        yield w._close()\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    def start_diagnostics_server(self, port=8787, show=False,\n            silence=logging.CRITICAL):\n        \"\"\" Start Diagnostics Web Server\n\n        This starts a web application to show diagnostics of what is happening\n        on the cluster.  This application runs in a separate process and is\n        generally available at the following location:\n\n            http://localhost:8787/status/\n        \"\"\"\n        try:\n            from distributed.bokeh.application import BokehWebInterface\n        except ImportError:\n            logger.info(\"To start diagnostics web server please install Bokeh\")\n            return\n\n        assert self.diagnostics is None\n        if 'http' not in self.scheduler.services:\n            self.scheduler.services['http'] = HTTPScheduler(self.scheduler,\n                    io_loop=self.scheduler.loop)\n            self.scheduler.services['http'].listen(0)\n        self.diagnostics = BokehWebInterface(\n                tcp_port=self.scheduler.port,\n                http_port=self.scheduler.services['http'].port,\n                bokeh_port=port, show=show,\n                log_level=logging.getLevelName(silence).lower())\n\n    @gen.coroutine\n    def _close(self):\n        with ignoring(gen.TimeoutError, StreamClosedError, OSError):\n            yield All([w._close() for w in self.workers])\n        with ignoring(gen.TimeoutError, StreamClosedError, OSError):\n            yield self.scheduler.close(fast=True)\n        del self.workers[:]\n        if self.diagnostics:\n            self.diagnostics.close()\n\n    def close(self):\n        \"\"\" Close the cluster \"\"\"\n        if self.status == 'running':\n            self.status = 'closed'\n            if self.loop._running:\n                sync(self.loop, self._close)\n            if hasattr(self, '_thread'):\n                sync(self.loop, self.loop.stop)\n                self._thread.join(timeout=1)\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        return self.scheduler.address\n", "path": "distributed/deploy/local.py"}]}