We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Rename local_client to worker_client?
The term "local", might be confusing.  Perhaps `worker_client` is more clear?  Or are there better names?

--- 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/__init__.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 from .config import config
4 from .core import connect, rpc
5 from .deploy import LocalCluster
6 from .diagnostics import progress
7 from .client import (Client, Executor, CompatibleExecutor, wait, as_completed,
8         default_client)
9 from .nanny import Nanny
10 from .scheduler import Scheduler
11 from .utils import sync
12 from .worker import Worker
13 from .worker_client import local_client
14 
15 try:
16     from .collections import futures_to_collection
17 except:
18     pass
19 
20 from ._version import get_versions
21 versions = get_versions()
22 __version__ = versions['version']
23 __git_revision__ = versions['full-revisionid']
24 del get_versions, versions
25
```

Path: `distributed/worker_client.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 from contextlib import contextmanager
4 from time import sleep
5 from tornado import gen
6 from toolz import keymap, valmap, merge, assoc
7 import uuid
8 
9 from dask.base import tokenize
10 from tornado import gen
11 
12 from .client import AllExit, Client, Future, pack_data, unpack_remotedata
13 from dask.compatibility import apply
14 from .sizeof import sizeof
15 from .threadpoolexecutor import secede
16 from .utils import All, log_errors, sync, tokey, ignoring
17 from .worker import thread_state
18 
19 
20 @contextmanager
21 def local_client(timeout=3):
22     """ Get client for this thread
23 
24     Note: This interface is new and experimental.  It may change without
25     notice.
26 
27     This context manager is intended to be called within functions that we run
28     on workers.  When run as a context manager it delivers a client
29     ``Client`` object that can submit other tasks directly from that worker.
30 
31     Examples
32     --------
33 
34     >>> def func(x):
35     ...     with local_client() as e:  # connect from worker back to scheduler
36     ...         a = e.submit(inc, x)     # this task can submit more tasks
37     ...         b = e.submit(dec, x)
38     ...         result = e.gather([a, b])  # and gather results
39     ...     return result
40 
41     >>> future = e.submit(func, 1)  # submit func(1) on cluster
42     """
43     address = thread_state.execution_state['scheduler']
44     worker = thread_state.execution_state['worker']
45     secede()  # have this thread secede from the thread pool
46               # so that it doesn't take up a fixed resource while waiting
47     worker.loop.add_callback(worker.transition, thread_state.key, 'long-running')
48 
49     with WorkerClient(address, loop=worker.loop) as wc:
50         # Make sure connection errors are bubbled to the caller
51         sync(wc.loop, wc._start, timeout=timeout)
52         assert wc.status == 'running'
53         yield wc
54 
55 
56 def get_worker():
57     return thread_state.execution_state['worker']
58 
59 
60 class WorkerClient(Client):
61     """ An Client designed to operate from a Worker process
62 
63     This client has had a few methods altered to make it more efficient for
64     working directly from the worker nodes.  In particular scatter/gather first
65     look to the local data dictionary rather than sending data over the network
66     """
67     def __init__(self, *args, **kwargs):
68         loop = kwargs.get('loop')
69         self.worker = get_worker()
70         sync(loop, apply, Client.__init__, (self,) + args, assoc(kwargs, 'start', False))
71 
72     @gen.coroutine
73     def _scatter(self, data, workers=None, broadcast=False):
74         """ Scatter data to local data dictionary
75 
76         Rather than send data out to the cluster we keep data local.  However
77         we do report to the scheduler that the local worker has the scattered
78         data.  This allows other workers to come by and steal this data if
79         desired.
80 
81         Keywords like ``broadcast=`` do not work, however operations like
82         ``.replicate`` work fine after calling scatter, which can fill in for
83         this functionality.
84         """
85         with log_errors():
86             if not (workers is None and broadcast is False):
87                 raise NotImplementedError("Scatter from worker doesn't support workers or broadcast keywords")
88 
89             if isinstance(data, dict) and not all(isinstance(k, (bytes, str))
90                                                    for k in data):
91                 d = yield self._scatter(keymap(tokey, data), workers, broadcast)
92                 raise gen.Return({k: d[tokey(k)] for k in data})
93 
94             if isinstance(data, (list, tuple, set, frozenset)):
95                 keys = []
96                 for x in data:
97                     try:
98                         keys.append(tokenize(x))
99                     except:
100                         keys.append(str(uuid.uuid1()))
101                 data2 = dict(zip(keys, data))
102             elif isinstance(data, dict):
103                 keys = set(data)
104                 data2 = data
105             else:
106                 raise TypeError("Don't know how to scatter %s" % type(data))
107 
108             self.worker.update_data(data=data2, report=False)
109 
110             yield self.scheduler.update_data(
111                     who_has={key: [self.worker.address] for key in data2},
112                     nbytes=valmap(sizeof, data2),
113                     client=self.id)
114 
115             if isinstance(data, dict):
116                 out = {k: Future(k, self) for k in data}
117             elif isinstance(data, (tuple, list, set, frozenset)):
118                 out = type(data)([Future(k, self) for k in keys])
119             else:
120                 raise TypeError(
121                         "Input to scatter must be a list or dict")
122 
123             for key in keys:
124                 self.futures[key].finish(type=None)
125 
126             raise gen.Return(out)
127 
128     @gen.coroutine
129     def _gather(self, futures, errors='raise'):
130         """
131 
132         Exactly like Client._gather, but get data directly from the local
133         worker data dictionary directly rather than through the scheduler.
134 
135         TODO: avoid scheduler for other communications, and assume that we can
136         communicate directly with the other workers.
137         """
138         futures2, keys = unpack_remotedata(futures, byte_keys=True)
139         keys = [tokey(k) for k in keys]
140 
141         @gen.coroutine
142         def wait(k):
143             """ Want to stop the All(...) early if we find an error """
144             yield self.futures[k].event.wait()
145             if self.futures[k].status != 'finished':
146                 raise AllExit()
147 
148         with ignoring(AllExit):
149             yield All([wait(key) for key in keys if key in self.futures])
150 
151         local = {k: self.worker.data[k] for k in keys
152                  if k in self.worker.data}
153 
154         futures3 = {k: Future(k, self) for k in keys if k not in local}
155 
156         futures4 = pack_data(futures2, merge(local, futures3))
157         if not futures3:
158             raise gen.Return(futures4)
159 
160         result = yield Client._gather(self, futures4, errors=errors)
161         raise gen.Return(result)
162
```
--- 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/__init__.py b/distributed/__init__.py
--- a/distributed/__init__.py
+++ b/distributed/__init__.py
@@ -10,7 +10,7 @@
 from .scheduler import Scheduler
 from .utils import sync
 from .worker import Worker
-from .worker_client import local_client
+from .worker_client import local_client, worker_client
 
 try:
     from .collections import futures_to_collection
diff --git a/distributed/worker_client.py b/distributed/worker_client.py
--- a/distributed/worker_client.py
+++ b/distributed/worker_client.py
@@ -18,7 +18,7 @@
 
 
 @contextmanager
-def local_client(timeout=3):
+def worker_client(timeout=3):
     """ Get client for this thread
 
     Note: This interface is new and experimental.  It may change without
@@ -32,7 +32,7 @@
     --------
 
     >>> def func(x):
-    ...     with local_client() as e:  # connect from worker back to scheduler
+    ...     with worker_client() as e:  # connect from worker back to scheduler
     ...         a = e.submit(inc, x)     # this task can submit more tasks
     ...         b = e.submit(dec, x)
     ...         result = e.gather([a, b])  # and gather results
@@ -53,6 +53,8 @@
         yield wc
 
 
+local_client = worker_client
+
 def get_worker():
     return thread_state.execution_state['worker']

{"golden_diff": "diff --git a/distributed/__init__.py b/distributed/__init__.py\n--- a/distributed/__init__.py\n+++ b/distributed/__init__.py\n@@ -10,7 +10,7 @@\n from .scheduler import Scheduler\n from .utils import sync\n from .worker import Worker\n-from .worker_client import local_client\n+from .worker_client import local_client, worker_client\n \n try:\n     from .collections import futures_to_collection\ndiff --git a/distributed/worker_client.py b/distributed/worker_client.py\n--- a/distributed/worker_client.py\n+++ b/distributed/worker_client.py\n@@ -18,7 +18,7 @@\n \n \n @contextmanager\n-def local_client(timeout=3):\n+def worker_client(timeout=3):\n     \"\"\" Get client for this thread\n \n     Note: This interface is new and experimental.  It may change without\n@@ -32,7 +32,7 @@\n     --------\n \n     >>> def func(x):\n-    ...     with local_client() as e:  # connect from worker back to scheduler\n+    ...     with worker_client() as e:  # connect from worker back to scheduler\n     ...         a = e.submit(inc, x)     # this task can submit more tasks\n     ...         b = e.submit(dec, x)\n     ...         result = e.gather([a, b])  # and gather results\n@@ -53,6 +53,8 @@\n         yield wc\n \n \n+local_client = worker_client\n+\n def get_worker():\n     return thread_state.execution_state['worker']\n", "issue": "Rename local_client to worker_client?\nThe term \"local\", might be confusing.  Perhaps `worker_client` is more clear?  Or are there better names?\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom .config import config\nfrom .core import connect, rpc\nfrom .deploy import LocalCluster\nfrom .diagnostics import progress\nfrom .client import (Client, Executor, CompatibleExecutor, wait, as_completed,\n        default_client)\nfrom .nanny import Nanny\nfrom .scheduler import Scheduler\nfrom .utils import sync\nfrom .worker import Worker\nfrom .worker_client import local_client\n\ntry:\n    from .collections import futures_to_collection\nexcept:\n    pass\n\nfrom ._version import get_versions\nversions = get_versions()\n__version__ = versions['version']\n__git_revision__ = versions['full-revisionid']\ndel get_versions, versions\n", "path": "distributed/__init__.py"}, {"content": "from __future__ import print_function, division, absolute_import\n\nfrom contextlib import contextmanager\nfrom time import sleep\nfrom tornado import gen\nfrom toolz import keymap, valmap, merge, assoc\nimport uuid\n\nfrom dask.base import tokenize\nfrom tornado import gen\n\nfrom .client import AllExit, Client, Future, pack_data, unpack_remotedata\nfrom dask.compatibility import apply\nfrom .sizeof import sizeof\nfrom .threadpoolexecutor import secede\nfrom .utils import All, log_errors, sync, tokey, ignoring\nfrom .worker import thread_state\n\n\n@contextmanager\ndef local_client(timeout=3):\n    \"\"\" Get client for this thread\n\n    Note: This interface is new and experimental.  It may change without\n    notice.\n\n    This context manager is intended to be called within functions that we run\n    on workers.  When run as a context manager it delivers a client\n    ``Client`` object that can submit other tasks directly from that worker.\n\n    Examples\n    --------\n\n    >>> def func(x):\n    ...     with local_client() as e:  # connect from worker back to scheduler\n    ...         a = e.submit(inc, x)     # this task can submit more tasks\n    ...         b = e.submit(dec, x)\n    ...         result = e.gather([a, b])  # and gather results\n    ...     return result\n\n    >>> future = e.submit(func, 1)  # submit func(1) on cluster\n    \"\"\"\n    address = thread_state.execution_state['scheduler']\n    worker = thread_state.execution_state['worker']\n    secede()  # have this thread secede from the thread pool\n              # so that it doesn't take up a fixed resource while waiting\n    worker.loop.add_callback(worker.transition, thread_state.key, 'long-running')\n\n    with WorkerClient(address, loop=worker.loop) as wc:\n        # Make sure connection errors are bubbled to the caller\n        sync(wc.loop, wc._start, timeout=timeout)\n        assert wc.status == 'running'\n        yield wc\n\n\ndef get_worker():\n    return thread_state.execution_state['worker']\n\n\nclass WorkerClient(Client):\n    \"\"\" An Client designed to operate from a Worker process\n\n    This client has had a few methods altered to make it more efficient for\n    working directly from the worker nodes.  In particular scatter/gather first\n    look to the local data dictionary rather than sending data over the network\n    \"\"\"\n    def __init__(self, *args, **kwargs):\n        loop = kwargs.get('loop')\n        self.worker = get_worker()\n        sync(loop, apply, Client.__init__, (self,) + args, assoc(kwargs, 'start', False))\n\n    @gen.coroutine\n    def _scatter(self, data, workers=None, broadcast=False):\n        \"\"\" Scatter data to local data dictionary\n\n        Rather than send data out to the cluster we keep data local.  However\n        we do report to the scheduler that the local worker has the scattered\n        data.  This allows other workers to come by and steal this data if\n        desired.\n\n        Keywords like ``broadcast=`` do not work, however operations like\n        ``.replicate`` work fine after calling scatter, which can fill in for\n        this functionality.\n        \"\"\"\n        with log_errors():\n            if not (workers is None and broadcast is False):\n                raise NotImplementedError(\"Scatter from worker doesn't support workers or broadcast keywords\")\n\n            if isinstance(data, dict) and not all(isinstance(k, (bytes, str))\n                                                   for k in data):\n                d = yield self._scatter(keymap(tokey, data), workers, broadcast)\n                raise gen.Return({k: d[tokey(k)] for k in data})\n\n            if isinstance(data, (list, tuple, set, frozenset)):\n                keys = []\n                for x in data:\n                    try:\n                        keys.append(tokenize(x))\n                    except:\n                        keys.append(str(uuid.uuid1()))\n                data2 = dict(zip(keys, data))\n            elif isinstance(data, dict):\n                keys = set(data)\n                data2 = data\n            else:\n                raise TypeError(\"Don't know how to scatter %s\" % type(data))\n\n            self.worker.update_data(data=data2, report=False)\n\n            yield self.scheduler.update_data(\n                    who_has={key: [self.worker.address] for key in data2},\n                    nbytes=valmap(sizeof, data2),\n                    client=self.id)\n\n            if isinstance(data, dict):\n                out = {k: Future(k, self) for k in data}\n            elif isinstance(data, (tuple, list, set, frozenset)):\n                out = type(data)([Future(k, self) for k in keys])\n            else:\n                raise TypeError(\n                        \"Input to scatter must be a list or dict\")\n\n            for key in keys:\n                self.futures[key].finish(type=None)\n\n            raise gen.Return(out)\n\n    @gen.coroutine\n    def _gather(self, futures, errors='raise'):\n        \"\"\"\n\n        Exactly like Client._gather, but get data directly from the local\n        worker data dictionary directly rather than through the scheduler.\n\n        TODO: avoid scheduler for other communications, and assume that we can\n        communicate directly with the other workers.\n        \"\"\"\n        futures2, keys = unpack_remotedata(futures, byte_keys=True)\n        keys = [tokey(k) for k in keys]\n\n        @gen.coroutine\n        def wait(k):\n            \"\"\" Want to stop the All(...) early if we find an error \"\"\"\n            yield self.futures[k].event.wait()\n            if self.futures[k].status != 'finished':\n                raise AllExit()\n\n        with ignoring(AllExit):\n            yield All([wait(key) for key in keys if key in self.futures])\n\n        local = {k: self.worker.data[k] for k in keys\n                 if k in self.worker.data}\n\n        futures3 = {k: Future(k, self) for k in keys if k not in local}\n\n        futures4 = pack_data(futures2, merge(local, futures3))\n        if not futures3:\n            raise gen.Return(futures4)\n\n        result = yield Client._gather(self, futures4, errors=errors)\n        raise gen.Return(result)\n", "path": "distributed/worker_client.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom .config import config\nfrom .core import connect, rpc\nfrom .deploy import LocalCluster\nfrom .diagnostics import progress\nfrom .client import (Client, Executor, CompatibleExecutor, wait, as_completed,\n        default_client)\nfrom .nanny import Nanny\nfrom .scheduler import Scheduler\nfrom .utils import sync\nfrom .worker import Worker\nfrom .worker_client import local_client, worker_client\n\ntry:\n    from .collections import futures_to_collection\nexcept:\n    pass\n\nfrom ._version import get_versions\nversions = get_versions()\n__version__ = versions['version']\n__git_revision__ = versions['full-revisionid']\ndel get_versions, versions\n", "path": "distributed/__init__.py"}, {"content": "from __future__ import print_function, division, absolute_import\n\nfrom contextlib import contextmanager\nfrom time import sleep\nfrom tornado import gen\nfrom toolz import keymap, valmap, merge, assoc\nimport uuid\n\nfrom dask.base import tokenize\nfrom tornado import gen\n\nfrom .client import AllExit, Client, Future, pack_data, unpack_remotedata\nfrom dask.compatibility import apply\nfrom .sizeof import sizeof\nfrom .threadpoolexecutor import secede\nfrom .utils import All, log_errors, sync, tokey, ignoring\nfrom .worker import thread_state\n\n\n@contextmanager\ndef worker_client(timeout=3):\n    \"\"\" Get client for this thread\n\n    Note: This interface is new and experimental.  It may change without\n    notice.\n\n    This context manager is intended to be called within functions that we run\n    on workers.  When run as a context manager it delivers a client\n    ``Client`` object that can submit other tasks directly from that worker.\n\n    Examples\n    --------\n\n    >>> def func(x):\n    ...     with worker_client() as e:  # connect from worker back to scheduler\n    ...         a = e.submit(inc, x)     # this task can submit more tasks\n    ...         b = e.submit(dec, x)\n    ...         result = e.gather([a, b])  # and gather results\n    ...     return result\n\n    >>> future = e.submit(func, 1)  # submit func(1) on cluster\n    \"\"\"\n    address = thread_state.execution_state['scheduler']\n    worker = thread_state.execution_state['worker']\n    secede()  # have this thread secede from the thread pool\n              # so that it doesn't take up a fixed resource while waiting\n    worker.loop.add_callback(worker.transition, thread_state.key, 'long-running')\n\n    with WorkerClient(address, loop=worker.loop) as wc:\n        # Make sure connection errors are bubbled to the caller\n        sync(wc.loop, wc._start, timeout=timeout)\n        assert wc.status == 'running'\n        yield wc\n\n\nlocal_client = worker_client\n\ndef get_worker():\n    return thread_state.execution_state['worker']\n\n\nclass WorkerClient(Client):\n    \"\"\" An Client designed to operate from a Worker process\n\n    This client has had a few methods altered to make it more efficient for\n    working directly from the worker nodes.  In particular scatter/gather first\n    look to the local data dictionary rather than sending data over the network\n    \"\"\"\n    def __init__(self, *args, **kwargs):\n        loop = kwargs.get('loop')\n        self.worker = get_worker()\n        sync(loop, apply, Client.__init__, (self,) + args, assoc(kwargs, 'start', False))\n\n    @gen.coroutine\n    def _scatter(self, data, workers=None, broadcast=False):\n        \"\"\" Scatter data to local data dictionary\n\n        Rather than send data out to the cluster we keep data local.  However\n        we do report to the scheduler that the local worker has the scattered\n        data.  This allows other workers to come by and steal this data if\n        desired.\n\n        Keywords like ``broadcast=`` do not work, however operations like\n        ``.replicate`` work fine after calling scatter, which can fill in for\n        this functionality.\n        \"\"\"\n        with log_errors():\n            if not (workers is None and broadcast is False):\n                raise NotImplementedError(\"Scatter from worker doesn't support workers or broadcast keywords\")\n\n            if isinstance(data, dict) and not all(isinstance(k, (bytes, str))\n                                                   for k in data):\n                d = yield self._scatter(keymap(tokey, data), workers, broadcast)\n                raise gen.Return({k: d[tokey(k)] for k in data})\n\n            if isinstance(data, (list, tuple, set, frozenset)):\n                keys = []\n                for x in data:\n                    try:\n                        keys.append(tokenize(x))\n                    except:\n                        keys.append(str(uuid.uuid1()))\n                data2 = dict(zip(keys, data))\n            elif isinstance(data, dict):\n                keys = set(data)\n                data2 = data\n            else:\n                raise TypeError(\"Don't know how to scatter %s\" % type(data))\n\n            self.worker.update_data(data=data2, report=False)\n\n            yield self.scheduler.update_data(\n                    who_has={key: [self.worker.address] for key in data2},\n                    nbytes=valmap(sizeof, data2),\n                    client=self.id)\n\n            if isinstance(data, dict):\n                out = {k: Future(k, self) for k in data}\n            elif isinstance(data, (tuple, list, set, frozenset)):\n                out = type(data)([Future(k, self) for k in keys])\n            else:\n                raise TypeError(\n                        \"Input to scatter must be a list or dict\")\n\n            for key in keys:\n                self.futures[key].finish(type=None)\n\n            raise gen.Return(out)\n\n    @gen.coroutine\n    def _gather(self, futures, errors='raise'):\n        \"\"\"\n\n        Exactly like Client._gather, but get data directly from the local\n        worker data dictionary directly rather than through the scheduler.\n\n        TODO: avoid scheduler for other communications, and assume that we can\n        communicate directly with the other workers.\n        \"\"\"\n        futures2, keys = unpack_remotedata(futures, byte_keys=True)\n        keys = [tokey(k) for k in keys]\n\n        @gen.coroutine\n        def wait(k):\n            \"\"\" Want to stop the All(...) early if we find an error \"\"\"\n            yield self.futures[k].event.wait()\n            if self.futures[k].status != 'finished':\n                raise AllExit()\n\n        with ignoring(AllExit):\n            yield All([wait(key) for key in keys if key in self.futures])\n\n        local = {k: self.worker.data[k] for k in keys\n                 if k in self.worker.data}\n\n        futures3 = {k: Future(k, self) for k in keys if k not in local}\n\n        futures4 = pack_data(futures2, merge(local, futures3))\n        if not futures3:\n            raise gen.Return(futures4)\n\n        result = yield Client._gather(self, futures4, errors=errors)\n        raise gen.Return(result)\n", "path": "distributed/worker_client.py"}]}