We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fix profile on msgpack 1.0.6/Windows
- Closes #8212
- Iterates on #3455
- Twin of #8214

Make sure that profile doesn't reach the point where sizeof() starts raising (at depth 140).
This is problematic with msgpack 1.0.6 on Windows because its max recursion depth fell from 512 to 256.

The remaining exception in the unit tests is fixed by #8214

--- 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 annotations
2 
3 import logging
4 import math
5 import warnings
6 
7 import toolz
8 
9 from dask.system import CPU_COUNT
10 from dask.widgets import get_template
11 
12 from distributed.deploy.spec import SpecCluster
13 from distributed.deploy.utils import nprocesses_nthreads
14 from distributed.nanny import Nanny
15 from distributed.scheduler import Scheduler
16 from distributed.security import Security
17 from distributed.worker import Worker
18 from distributed.worker_memory import parse_memory_limit
19 
20 logger = logging.getLogger(__name__)
21 
22 
23 class LocalCluster(SpecCluster):
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, unless
35         worker_class=Worker, in which case it defaults to False.
36     threads_per_worker: int
37         Number of threads per each worker
38     scheduler_port: int
39         Port of the scheduler.  8786 by default, use 0 to choose a random port
40     silence_logs: logging level
41         Level of logs to print out to stdout.  ``logging.WARN`` by default.
42         Use a falsey value like False or None for no change.
43     host: string
44         Host address on which the scheduler will listen, defaults to only localhost
45     ip: string
46         Deprecated.  See ``host`` above.
47     dashboard_address: str
48         Address on which to listen for the Bokeh diagnostics server like
49         'localhost:8787' or '0.0.0.0:8787'.  Defaults to ':8787'.
50         Set to ``None`` to disable the dashboard.
51         Use ':0' for a random port.
52     worker_dashboard_address: str
53         Address on which to listen for the Bokeh worker diagnostics server like
54         'localhost:8787' or '0.0.0.0:8787'.  Defaults to None which disables the dashboard.
55         Use ':0' for a random port.
56     diagnostics_port: int
57         Deprecated.  See dashboard_address.
58     asynchronous: bool (False by default)
59         Set to True if using this cluster within async/await functions or within
60         Tornado gen.coroutines.  This should remain False for normal use.
61     blocked_handlers: List[str]
62         A list of strings specifying a blocklist of handlers to disallow on the
63         Scheduler, like ``['feed', 'run_function']``
64     service_kwargs: Dict[str, Dict]
65         Extra keywords to hand to the running services
66     security : Security or bool, optional
67         Configures communication security in this cluster. Can be a security
68         object, or True. If True, temporary self-signed credentials will
69         be created automatically.
70     protocol: str (optional)
71         Protocol to use like ``tcp://``, ``tls://``, ``inproc://``
72         This defaults to sensible choice given other keyword arguments like
73         ``processes`` and ``security``
74     interface: str (optional)
75         Network interface to use.  Defaults to lo/localhost
76     worker_class: Worker
77         Worker class used to instantiate workers from. Defaults to Worker if
78         processes=False and Nanny if processes=True or omitted.
79     **worker_kwargs:
80         Extra worker arguments. Any additional keyword arguments will be passed
81         to the ``Worker`` class constructor.
82 
83     Examples
84     --------
85     >>> cluster = LocalCluster()  # Create a local cluster  # doctest: +SKIP
86     >>> cluster  # doctest: +SKIP
87     LocalCluster("127.0.0.1:8786", workers=8, threads=8)
88 
89     >>> c = Client(cluster)  # connect to local cluster  # doctest: +SKIP
90 
91     Scale the cluster to three workers
92 
93     >>> cluster.scale(3)  # doctest: +SKIP
94 
95     Pass extra keyword arguments to Bokeh
96 
97     >>> LocalCluster(service_kwargs={'dashboard': {'prefix': '/foo'}})  # doctest: +SKIP
98     """
99 
100     def __init__(
101         self,
102         name=None,
103         n_workers=None,
104         threads_per_worker=None,
105         processes=None,
106         loop=None,
107         start=None,
108         host=None,
109         ip=None,
110         scheduler_port=0,
111         silence_logs=logging.WARN,
112         dashboard_address=":8787",
113         worker_dashboard_address=None,
114         diagnostics_port=None,
115         services=None,
116         worker_services=None,
117         service_kwargs=None,
118         asynchronous=False,
119         security=None,
120         protocol=None,
121         blocked_handlers=None,
122         interface=None,
123         worker_class=None,
124         scheduler_kwargs=None,
125         scheduler_sync_interval=1,
126         **worker_kwargs,
127     ):
128         if ip is not None:
129             # In the future we should warn users about this move
130             # warnings.warn("The ip keyword has been moved to host")
131             host = ip
132 
133         if diagnostics_port is not None:
134             warnings.warn(
135                 "diagnostics_port has been deprecated. "
136                 "Please use `dashboard_address=` instead"
137             )
138             dashboard_address = diagnostics_port
139 
140         if threads_per_worker == 0:
141             warnings.warn(
142                 "Setting `threads_per_worker` to 0 has been deprecated. "
143                 "Please set to None or to a specific int."
144             )
145             threads_per_worker = None
146 
147         if "dashboard" in worker_kwargs:
148             warnings.warn(
149                 "Setting `dashboard` is discouraged. "
150                 "Please set `dashboard_address` to affect the scheduler (more common) "
151                 "and `worker_dashboard_address` for the worker (less common)."
152             )
153 
154         if processes is None:
155             processes = worker_class is None or issubclass(worker_class, Nanny)
156         if worker_class is None:
157             worker_class = Nanny if processes else Worker
158 
159         self.status = None
160         self.processes = processes
161 
162         if security is None:
163             # Falsey values load the default configuration
164             security = Security()
165         elif security is True:
166             # True indicates self-signed temporary credentials should be used
167             security = Security.temporary()
168         elif not isinstance(security, Security):
169             raise TypeError("security must be a Security object")
170 
171         if protocol is None:
172             if host and "://" in host:
173                 protocol = host.split("://")[0]
174             elif security and security.require_encryption:
175                 protocol = "tls://"
176             elif not self.processes and not scheduler_port:
177                 protocol = "inproc://"
178             else:
179                 protocol = "tcp://"
180         if not protocol.endswith("://"):
181             protocol = protocol + "://"
182 
183         if host is None and not protocol.startswith("inproc") and not interface:
184             host = "127.0.0.1"
185 
186         services = services or {}
187         worker_services = worker_services or {}
188         if n_workers is None and threads_per_worker is None:
189             if processes:
190                 n_workers, threads_per_worker = nprocesses_nthreads()
191             else:
192                 n_workers = 1
193                 threads_per_worker = CPU_COUNT
194         if n_workers is None and threads_per_worker is not None:
195             n_workers = max(1, CPU_COUNT // threads_per_worker) if processes else 1
196         if n_workers and threads_per_worker is None:
197             # Overcommit threads per worker, rather than undercommit
198             threads_per_worker = max(1, int(math.ceil(CPU_COUNT / n_workers)))
199         if n_workers and "memory_limit" not in worker_kwargs:
200             worker_kwargs["memory_limit"] = parse_memory_limit("auto", 1, n_workers)
201 
202         worker_kwargs.update(
203             {
204                 "host": host,
205                 "nthreads": threads_per_worker,
206                 "services": worker_services,
207                 "dashboard_address": worker_dashboard_address,
208                 "dashboard": worker_dashboard_address is not None,
209                 "interface": interface,
210                 "protocol": protocol,
211                 "security": security,
212                 "silence_logs": silence_logs,
213             }
214         )
215 
216         scheduler = {
217             "cls": Scheduler,
218             "options": toolz.merge(
219                 dict(
220                     host=host,
221                     services=services,
222                     service_kwargs=service_kwargs,
223                     security=security,
224                     port=scheduler_port,
225                     interface=interface,
226                     protocol=protocol,
227                     dashboard=dashboard_address is not None,
228                     dashboard_address=dashboard_address,
229                     blocked_handlers=blocked_handlers,
230                 ),
231                 scheduler_kwargs or {},
232             ),
233         }
234 
235         worker = {"cls": worker_class, "options": worker_kwargs}
236         workers = {i: worker for i in range(n_workers)}
237 
238         super().__init__(
239             name=name,
240             scheduler=scheduler,
241             workers=workers,
242             worker=worker,
243             loop=loop,
244             asynchronous=asynchronous,
245             silence_logs=silence_logs,
246             security=security,
247             scheduler_sync_interval=scheduler_sync_interval,
248         )
249 
250     def start_worker(self, *args, **kwargs):
251         raise NotImplementedError(
252             "The `cluster.start_worker` function has been removed. "
253             "Please see the `cluster.scale` method instead."
254         )
255 
256     def _repr_html_(self, cluster_status=None):
257         cluster_status = get_template("local_cluster.html.j2").render(
258             status=self.status.name,
259             processes=self.processes,
260             cluster_status=cluster_status,
261         )
262         return super()._repr_html_(cluster_status=cluster_status)
263
```
--- 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
@@ -30,6 +30,19 @@
     ----------
     n_workers: int
         Number of workers to start
+    memory_limit: str, float, int, or None, default "auto"
+        Sets the memory limit *per worker*.
+
+        Notes regarding argument data type:
+
+        * If None or 0, no limit is applied.
+        * If "auto", the total system memory is split evenly between the workers.
+        * If a float, that fraction of the system memory is used *per worker*.
+        * If a string giving a number of bytes (like ``"1GiB"``), that amount is used *per worker*.
+        * If an int, that number of bytes is used *per worker*.
+
+        Note that the limit will only be enforced when ``processes=True``, and the limit is only
+        enforced on a best-effort basis — it's still possible for workers to exceed this limit.
     processes: bool
         Whether to use processes (True) or threads (False).  Defaults to True, unless
         worker_class=Worker, in which case it defaults to False.

{"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@@ -30,6 +30,19 @@\n     ----------\n     n_workers: int\n         Number of workers to start\n+    memory_limit: str, float, int, or None, default \"auto\"\n+        Sets the memory limit *per worker*.\n+\n+        Notes regarding argument data type:\n+\n+        * If None or 0, no limit is applied.\n+        * If \"auto\", the total system memory is split evenly between the workers.\n+        * If a float, that fraction of the system memory is used *per worker*.\n+        * If a string giving a number of bytes (like ``\"1GiB\"``), that amount is used *per worker*.\n+        * If an int, that number of bytes is used *per worker*.\n+\n+        Note that the limit will only be enforced when ``processes=True``, and the limit is only\n+        enforced on a best-effort basis \u2014 it's still possible for workers to exceed this limit.\n     processes: bool\n         Whether to use processes (True) or threads (False).  Defaults to True, unless\n         worker_class=Worker, in which case it defaults to False.\n", "issue": "Fix profile on msgpack 1.0.6/Windows\n- Closes #8212\r\n- Iterates on #3455\r\n- Twin of #8214\r\n\r\nMake sure that profile doesn't reach the point where sizeof() starts raising (at depth 140).\r\nThis is problematic with msgpack 1.0.6 on Windows because its max recursion depth fell from 512 to 256.\r\n\r\nThe remaining exception in the unit tests is fixed by #8214\n", "before_files": [{"content": "from __future__ import annotations\n\nimport logging\nimport math\nimport warnings\n\nimport toolz\n\nfrom dask.system import CPU_COUNT\nfrom dask.widgets import get_template\n\nfrom distributed.deploy.spec import SpecCluster\nfrom distributed.deploy.utils import nprocesses_nthreads\nfrom distributed.nanny import Nanny\nfrom distributed.scheduler import Scheduler\nfrom distributed.security import Security\nfrom distributed.worker import Worker\nfrom distributed.worker_memory import parse_memory_limit\n\nlogger = logging.getLogger(__name__)\n\n\nclass LocalCluster(SpecCluster):\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, unless\n        worker_class=Worker, in which case it defaults to False.\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.WARN`` by default.\n        Use a falsey value like False or None for no change.\n    host: string\n        Host address on which the scheduler will listen, defaults to only localhost\n    ip: string\n        Deprecated.  See ``host`` above.\n    dashboard_address: str\n        Address on which to listen for the Bokeh diagnostics server like\n        'localhost:8787' or '0.0.0.0:8787'.  Defaults to ':8787'.\n        Set to ``None`` to disable the dashboard.\n        Use ':0' for a random port.\n    worker_dashboard_address: str\n        Address on which to listen for the Bokeh worker diagnostics server like\n        'localhost:8787' or '0.0.0.0:8787'.  Defaults to None which disables the dashboard.\n        Use ':0' for a random port.\n    diagnostics_port: int\n        Deprecated.  See dashboard_address.\n    asynchronous: bool (False by default)\n        Set to True if using this cluster within async/await functions or within\n        Tornado gen.coroutines.  This should remain False for normal use.\n    blocked_handlers: List[str]\n        A list of strings specifying a blocklist of handlers to disallow on the\n        Scheduler, like ``['feed', 'run_function']``\n    service_kwargs: Dict[str, Dict]\n        Extra keywords to hand to the running services\n    security : Security or bool, optional\n        Configures communication security in this cluster. Can be a security\n        object, or True. If True, temporary self-signed credentials will\n        be created automatically.\n    protocol: str (optional)\n        Protocol to use like ``tcp://``, ``tls://``, ``inproc://``\n        This defaults to sensible choice given other keyword arguments like\n        ``processes`` and ``security``\n    interface: str (optional)\n        Network interface to use.  Defaults to lo/localhost\n    worker_class: Worker\n        Worker class used to instantiate workers from. Defaults to Worker if\n        processes=False and Nanny if processes=True or omitted.\n    **worker_kwargs:\n        Extra worker arguments. Any additional keyword arguments will be passed\n        to the ``Worker`` class constructor.\n\n    Examples\n    --------\n    >>> cluster = LocalCluster()  # Create a local cluster  # doctest: +SKIP\n    >>> cluster  # doctest: +SKIP\n    LocalCluster(\"127.0.0.1:8786\", workers=8, threads=8)\n\n    >>> c = Client(cluster)  # connect to local cluster  # doctest: +SKIP\n\n    Scale the cluster to three workers\n\n    >>> cluster.scale(3)  # doctest: +SKIP\n\n    Pass extra keyword arguments to Bokeh\n\n    >>> LocalCluster(service_kwargs={'dashboard': {'prefix': '/foo'}})  # doctest: +SKIP\n    \"\"\"\n\n    def __init__(\n        self,\n        name=None,\n        n_workers=None,\n        threads_per_worker=None,\n        processes=None,\n        loop=None,\n        start=None,\n        host=None,\n        ip=None,\n        scheduler_port=0,\n        silence_logs=logging.WARN,\n        dashboard_address=\":8787\",\n        worker_dashboard_address=None,\n        diagnostics_port=None,\n        services=None,\n        worker_services=None,\n        service_kwargs=None,\n        asynchronous=False,\n        security=None,\n        protocol=None,\n        blocked_handlers=None,\n        interface=None,\n        worker_class=None,\n        scheduler_kwargs=None,\n        scheduler_sync_interval=1,\n        **worker_kwargs,\n    ):\n        if ip is not None:\n            # In the future we should warn users about this move\n            # warnings.warn(\"The ip keyword has been moved to host\")\n            host = ip\n\n        if diagnostics_port is not None:\n            warnings.warn(\n                \"diagnostics_port has been deprecated. \"\n                \"Please use `dashboard_address=` instead\"\n            )\n            dashboard_address = diagnostics_port\n\n        if threads_per_worker == 0:\n            warnings.warn(\n                \"Setting `threads_per_worker` to 0 has been deprecated. \"\n                \"Please set to None or to a specific int.\"\n            )\n            threads_per_worker = None\n\n        if \"dashboard\" in worker_kwargs:\n            warnings.warn(\n                \"Setting `dashboard` is discouraged. \"\n                \"Please set `dashboard_address` to affect the scheduler (more common) \"\n                \"and `worker_dashboard_address` for the worker (less common).\"\n            )\n\n        if processes is None:\n            processes = worker_class is None or issubclass(worker_class, Nanny)\n        if worker_class is None:\n            worker_class = Nanny if processes else Worker\n\n        self.status = None\n        self.processes = processes\n\n        if security is None:\n            # Falsey values load the default configuration\n            security = Security()\n        elif security is True:\n            # True indicates self-signed temporary credentials should be used\n            security = Security.temporary()\n        elif not isinstance(security, Security):\n            raise TypeError(\"security must be a Security object\")\n\n        if protocol is None:\n            if host and \"://\" in host:\n                protocol = host.split(\"://\")[0]\n            elif security and security.require_encryption:\n                protocol = \"tls://\"\n            elif not self.processes and not scheduler_port:\n                protocol = \"inproc://\"\n            else:\n                protocol = \"tcp://\"\n        if not protocol.endswith(\"://\"):\n            protocol = protocol + \"://\"\n\n        if host is None and not protocol.startswith(\"inproc\") and not interface:\n            host = \"127.0.0.1\"\n\n        services = services or {}\n        worker_services = worker_services or {}\n        if n_workers is None and threads_per_worker is None:\n            if processes:\n                n_workers, threads_per_worker = nprocesses_nthreads()\n            else:\n                n_workers = 1\n                threads_per_worker = CPU_COUNT\n        if n_workers is None and threads_per_worker is not None:\n            n_workers = max(1, CPU_COUNT // threads_per_worker) if processes else 1\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(CPU_COUNT / n_workers)))\n        if n_workers and \"memory_limit\" not in worker_kwargs:\n            worker_kwargs[\"memory_limit\"] = parse_memory_limit(\"auto\", 1, n_workers)\n\n        worker_kwargs.update(\n            {\n                \"host\": host,\n                \"nthreads\": threads_per_worker,\n                \"services\": worker_services,\n                \"dashboard_address\": worker_dashboard_address,\n                \"dashboard\": worker_dashboard_address is not None,\n                \"interface\": interface,\n                \"protocol\": protocol,\n                \"security\": security,\n                \"silence_logs\": silence_logs,\n            }\n        )\n\n        scheduler = {\n            \"cls\": Scheduler,\n            \"options\": toolz.merge(\n                dict(\n                    host=host,\n                    services=services,\n                    service_kwargs=service_kwargs,\n                    security=security,\n                    port=scheduler_port,\n                    interface=interface,\n                    protocol=protocol,\n                    dashboard=dashboard_address is not None,\n                    dashboard_address=dashboard_address,\n                    blocked_handlers=blocked_handlers,\n                ),\n                scheduler_kwargs or {},\n            ),\n        }\n\n        worker = {\"cls\": worker_class, \"options\": worker_kwargs}\n        workers = {i: worker for i in range(n_workers)}\n\n        super().__init__(\n            name=name,\n            scheduler=scheduler,\n            workers=workers,\n            worker=worker,\n            loop=loop,\n            asynchronous=asynchronous,\n            silence_logs=silence_logs,\n            security=security,\n            scheduler_sync_interval=scheduler_sync_interval,\n        )\n\n    def start_worker(self, *args, **kwargs):\n        raise NotImplementedError(\n            \"The `cluster.start_worker` function has been removed. \"\n            \"Please see the `cluster.scale` method instead.\"\n        )\n\n    def _repr_html_(self, cluster_status=None):\n        cluster_status = get_template(\"local_cluster.html.j2\").render(\n            status=self.status.name,\n            processes=self.processes,\n            cluster_status=cluster_status,\n        )\n        return super()._repr_html_(cluster_status=cluster_status)\n", "path": "distributed/deploy/local.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport logging\nimport math\nimport warnings\n\nimport toolz\n\nfrom dask.system import CPU_COUNT\nfrom dask.widgets import get_template\n\nfrom distributed.deploy.spec import SpecCluster\nfrom distributed.deploy.utils import nprocesses_nthreads\nfrom distributed.nanny import Nanny\nfrom distributed.scheduler import Scheduler\nfrom distributed.security import Security\nfrom distributed.worker import Worker\nfrom distributed.worker_memory import parse_memory_limit\n\nlogger = logging.getLogger(__name__)\n\n\nclass LocalCluster(SpecCluster):\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    memory_limit: str, float, int, or None, default \"auto\"\n        Sets the memory limit *per worker*.\n\n        Notes regarding argument data type:\n\n        * If None or 0, no limit is applied.\n        * If \"auto\", the total system memory is split evenly between the workers.\n        * If a float, that fraction of the system memory is used *per worker*.\n        * If a string giving a number of bytes (like ``\"1GiB\"``), that amount is used *per worker*.\n        * If an int, that number of bytes is used *per worker*.\n\n        Note that the limit will only be enforced when ``processes=True``, and the limit is only\n        enforced on a best-effort basis \u2014 it's still possible for workers to exceed this limit.\n    processes: bool\n        Whether to use processes (True) or threads (False).  Defaults to True, unless\n        worker_class=Worker, in which case it defaults to False.\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.WARN`` by default.\n        Use a falsey value like False or None for no change.\n    host: string\n        Host address on which the scheduler will listen, defaults to only localhost\n    ip: string\n        Deprecated.  See ``host`` above.\n    dashboard_address: str\n        Address on which to listen for the Bokeh diagnostics server like\n        'localhost:8787' or '0.0.0.0:8787'.  Defaults to ':8787'.\n        Set to ``None`` to disable the dashboard.\n        Use ':0' for a random port.\n    worker_dashboard_address: str\n        Address on which to listen for the Bokeh worker diagnostics server like\n        'localhost:8787' or '0.0.0.0:8787'.  Defaults to None which disables the dashboard.\n        Use ':0' for a random port.\n    diagnostics_port: int\n        Deprecated.  See dashboard_address.\n    asynchronous: bool (False by default)\n        Set to True if using this cluster within async/await functions or within\n        Tornado gen.coroutines.  This should remain False for normal use.\n    blocked_handlers: List[str]\n        A list of strings specifying a blocklist of handlers to disallow on the\n        Scheduler, like ``['feed', 'run_function']``\n    service_kwargs: Dict[str, Dict]\n        Extra keywords to hand to the running services\n    security : Security or bool, optional\n        Configures communication security in this cluster. Can be a security\n        object, or True. If True, temporary self-signed credentials will\n        be created automatically.\n    protocol: str (optional)\n        Protocol to use like ``tcp://``, ``tls://``, ``inproc://``\n        This defaults to sensible choice given other keyword arguments like\n        ``processes`` and ``security``\n    interface: str (optional)\n        Network interface to use.  Defaults to lo/localhost\n    worker_class: Worker\n        Worker class used to instantiate workers from. Defaults to Worker if\n        processes=False and Nanny if processes=True or omitted.\n    **worker_kwargs:\n        Extra worker arguments. Any additional keyword arguments will be passed\n        to the ``Worker`` class constructor.\n\n    Examples\n    --------\n    >>> cluster = LocalCluster()  # Create a local cluster  # doctest: +SKIP\n    >>> cluster  # doctest: +SKIP\n    LocalCluster(\"127.0.0.1:8786\", workers=8, threads=8)\n\n    >>> c = Client(cluster)  # connect to local cluster  # doctest: +SKIP\n\n    Scale the cluster to three workers\n\n    >>> cluster.scale(3)  # doctest: +SKIP\n\n    Pass extra keyword arguments to Bokeh\n\n    >>> LocalCluster(service_kwargs={'dashboard': {'prefix': '/foo'}})  # doctest: +SKIP\n    \"\"\"\n\n    def __init__(\n        self,\n        name=None,\n        n_workers=None,\n        threads_per_worker=None,\n        processes=None,\n        loop=None,\n        start=None,\n        host=None,\n        ip=None,\n        scheduler_port=0,\n        silence_logs=logging.WARN,\n        dashboard_address=\":8787\",\n        worker_dashboard_address=None,\n        diagnostics_port=None,\n        services=None,\n        worker_services=None,\n        service_kwargs=None,\n        asynchronous=False,\n        security=None,\n        protocol=None,\n        blocked_handlers=None,\n        interface=None,\n        worker_class=None,\n        scheduler_kwargs=None,\n        scheduler_sync_interval=1,\n        **worker_kwargs,\n    ):\n        if ip is not None:\n            # In the future we should warn users about this move\n            # warnings.warn(\"The ip keyword has been moved to host\")\n            host = ip\n\n        if diagnostics_port is not None:\n            warnings.warn(\n                \"diagnostics_port has been deprecated. \"\n                \"Please use `dashboard_address=` instead\"\n            )\n            dashboard_address = diagnostics_port\n\n        if threads_per_worker == 0:\n            warnings.warn(\n                \"Setting `threads_per_worker` to 0 has been deprecated. \"\n                \"Please set to None or to a specific int.\"\n            )\n            threads_per_worker = None\n\n        if \"dashboard\" in worker_kwargs:\n            warnings.warn(\n                \"Setting `dashboard` is discouraged. \"\n                \"Please set `dashboard_address` to affect the scheduler (more common) \"\n                \"and `worker_dashboard_address` for the worker (less common).\"\n            )\n\n        if processes is None:\n            processes = worker_class is None or issubclass(worker_class, Nanny)\n        if worker_class is None:\n            worker_class = Nanny if processes else Worker\n\n        self.status = None\n        self.processes = processes\n\n        if security is None:\n            # Falsey values load the default configuration\n            security = Security()\n        elif security is True:\n            # True indicates self-signed temporary credentials should be used\n            security = Security.temporary()\n        elif not isinstance(security, Security):\n            raise TypeError(\"security must be a Security object\")\n\n        if protocol is None:\n            if host and \"://\" in host:\n                protocol = host.split(\"://\")[0]\n            elif security and security.require_encryption:\n                protocol = \"tls://\"\n            elif not self.processes and not scheduler_port:\n                protocol = \"inproc://\"\n            else:\n                protocol = \"tcp://\"\n        if not protocol.endswith(\"://\"):\n            protocol = protocol + \"://\"\n\n        if host is None and not protocol.startswith(\"inproc\") and not interface:\n            host = \"127.0.0.1\"\n\n        services = services or {}\n        worker_services = worker_services or {}\n        if n_workers is None and threads_per_worker is None:\n            if processes:\n                n_workers, threads_per_worker = nprocesses_nthreads()\n            else:\n                n_workers = 1\n                threads_per_worker = CPU_COUNT\n        if n_workers is None and threads_per_worker is not None:\n            n_workers = max(1, CPU_COUNT // threads_per_worker) if processes else 1\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(CPU_COUNT / n_workers)))\n        if n_workers and \"memory_limit\" not in worker_kwargs:\n            worker_kwargs[\"memory_limit\"] = parse_memory_limit(\"auto\", 1, n_workers)\n\n        worker_kwargs.update(\n            {\n                \"host\": host,\n                \"nthreads\": threads_per_worker,\n                \"services\": worker_services,\n                \"dashboard_address\": worker_dashboard_address,\n                \"dashboard\": worker_dashboard_address is not None,\n                \"interface\": interface,\n                \"protocol\": protocol,\n                \"security\": security,\n                \"silence_logs\": silence_logs,\n            }\n        )\n\n        scheduler = {\n            \"cls\": Scheduler,\n            \"options\": toolz.merge(\n                dict(\n                    host=host,\n                    services=services,\n                    service_kwargs=service_kwargs,\n                    security=security,\n                    port=scheduler_port,\n                    interface=interface,\n                    protocol=protocol,\n                    dashboard=dashboard_address is not None,\n                    dashboard_address=dashboard_address,\n                    blocked_handlers=blocked_handlers,\n                ),\n                scheduler_kwargs or {},\n            ),\n        }\n\n        worker = {\"cls\": worker_class, \"options\": worker_kwargs}\n        workers = {i: worker for i in range(n_workers)}\n\n        super().__init__(\n            name=name,\n            scheduler=scheduler,\n            workers=workers,\n            worker=worker,\n            loop=loop,\n            asynchronous=asynchronous,\n            silence_logs=silence_logs,\n            security=security,\n            scheduler_sync_interval=scheduler_sync_interval,\n        )\n\n    def start_worker(self, *args, **kwargs):\n        raise NotImplementedError(\n            \"The `cluster.start_worker` function has been removed. \"\n            \"Please see the `cluster.scale` method instead.\"\n        )\n\n    def _repr_html_(self, cluster_status=None):\n        cluster_status = get_template(\"local_cluster.html.j2\").render(\n            status=self.status.name,\n            processes=self.processes,\n            cluster_status=cluster_status,\n        )\n        return super()._repr_html_(cluster_status=cluster_status)\n", "path": "distributed/deploy/local.py"}]}