We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Variable delete raises KeyError
It seems that there is an unhandled KeyError in the scheduler variable delete logic.

If using the asynchronous API only visible in the scheduler logs but it does mean that Variables will be "leaking" and staying around instead of being deleted.

It happens regardless of whether you set a value.

Shortest reproduction I can provide is as follows:
```python
from distributed import Client
from distributed import Variable


def main():
    Client()
    Variable().delete()


if __name__ == '__main__':
    main()
```

```python
tornado.application - ERROR - Exception in callback functools.partial(<bound method IOLoop._discard_future_result of <tornado.platform.asyncio.AsyncIOLoop object at 0x7f8fe9265be0>>, <Task finished coro=<VariableExtension.delete() done, defined at ./venv/lib/python3.6/site-packages/distributed/variable.py:101> exception=KeyError('variable-68a51209410248fa9f69f84b25f48343',)>)
Traceback (most recent call last):
  File "./venv/lib/python3.6/site-packages/tornado/ioloop.py", line 743, in _run_callback
    ret = callback()
  File "./venv/lib/python3.6/site-packages/tornado/ioloop.py", line 767, in _discard_future_result
    future.result()
  File "./venv/lib/python3.6/site-packages/distributed/variable.py", line 110, in delete
    del self.waiting_conditions[name]
KeyError: 'variable-68a51209410248fa9f69f84b25f48343'
```

Python 3.6.10
Reproduced with distributed 2.9.3 and 2.10.0

--- 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/variable.py`
Content:
```
1 import asyncio
2 from collections import defaultdict
3 import logging
4 import uuid
5 
6 from tlz import merge
7 
8 from .client import Future, _get_global_client, Client
9 from .utils import tokey, log_errors, TimeoutError, ignoring
10 from .worker import get_client
11 
12 logger = logging.getLogger(__name__)
13 
14 
15 class VariableExtension:
16     """ An extension for the scheduler to manage queues
17 
18     This adds the following routes to the scheduler
19 
20     *  variable-set
21     *  variable-get
22     *  variable-delete
23     """
24 
25     def __init__(self, scheduler):
26         self.scheduler = scheduler
27         self.variables = dict()
28         self.waiting = defaultdict(set)
29         self.waiting_conditions = defaultdict(asyncio.Condition)
30         self.started = asyncio.Condition()
31 
32         self.scheduler.handlers.update(
33             {"variable_set": self.set, "variable_get": self.get}
34         )
35 
36         self.scheduler.stream_handlers["variable-future-release"] = self.future_release
37         self.scheduler.stream_handlers["variable_delete"] = self.delete
38 
39         self.scheduler.extensions["variables"] = self
40 
41     async def set(self, stream=None, name=None, key=None, data=None, client=None):
42         if key is not None:
43             record = {"type": "Future", "value": key}
44             self.scheduler.client_desires_keys(keys=[key], client="variable-%s" % name)
45         else:
46             record = {"type": "msgpack", "value": data}
47         try:
48             old = self.variables[name]
49         except KeyError:
50             pass
51         else:
52             if old["type"] == "Future" and old["value"] != key:
53                 asyncio.ensure_future(self.release(old["value"], name))
54         if name not in self.variables:
55             async with self.started:
56                 self.started.notify_all()
57         self.variables[name] = record
58 
59     async def release(self, key, name):
60         while self.waiting[key, name]:
61             async with self.waiting_conditions[name]:
62                 await self.waiting_conditions[name].wait()
63 
64         self.scheduler.client_releases_keys(keys=[key], client="variable-%s" % name)
65         del self.waiting[key, name]
66 
67     async def future_release(self, name=None, key=None, token=None, client=None):
68         self.waiting[key, name].remove(token)
69         if not self.waiting[key, name]:
70             async with self.waiting_conditions[name]:
71                 self.waiting_conditions[name].notify_all()
72 
73     async def get(self, stream=None, name=None, client=None, timeout=None):
74         start = self.scheduler.loop.time()
75         while name not in self.variables:
76             if timeout is not None:
77                 left = timeout - (self.scheduler.loop.time() - start)
78             else:
79                 left = None
80             if left and left < 0:
81                 raise TimeoutError()
82             try:
83 
84                 async def _():  # Python 3.6 is odd and requires special help here
85                     await self.started.acquire()
86                     await self.started.wait()
87 
88                 await asyncio.wait_for(_(), timeout=left)
89             finally:
90                 with ignoring(RuntimeError):  # Python 3.6 loses lock on finally clause
91                     self.started.release()
92 
93         record = self.variables[name]
94         if record["type"] == "Future":
95             key = record["value"]
96             token = uuid.uuid4().hex
97             ts = self.scheduler.tasks.get(key)
98             state = ts.state if ts is not None else "lost"
99             msg = {"token": token, "state": state}
100             if state == "erred":
101                 msg["exception"] = ts.exception_blame.exception
102                 msg["traceback"] = ts.exception_blame.traceback
103             record = merge(record, msg)
104             self.waiting[key, name].add(token)
105         return record
106 
107     async def delete(self, stream=None, name=None, client=None):
108         with log_errors():
109             try:
110                 old = self.variables[name]
111             except KeyError:
112                 pass
113             else:
114                 if old["type"] == "Future":
115                     await self.release(old["value"], name)
116             del self.waiting_conditions[name]
117             del self.variables[name]
118 
119 
120 class Variable:
121     """ Distributed Global Variable
122 
123     This allows multiple clients to share futures and data between each other
124     with a single mutable variable.  All metadata is sequentialized through the
125     scheduler.  Race conditions can occur.
126 
127     Values must be either Futures or msgpack-encodable data (ints, lists,
128     strings, etc..)  All data will be kept and sent through the scheduler, so
129     it is wise not to send too much.  If you want to share a large amount of
130     data then ``scatter`` it and share the future instead.
131 
132     .. warning::
133 
134        This object is experimental and has known issues in Python 2
135 
136     Parameters
137     ----------
138     name: string (optional)
139         Name used by other clients and the scheduler to identify the variable.
140         If not given, a random name will be generated.
141     client: Client (optional)
142         Client used for communication with the scheduler. Defaults to the
143         value of ``_get_global_client()``.
144 
145     Examples
146     --------
147     >>> from dask.distributed import Client, Variable # doctest: +SKIP
148     >>> client = Client()  # doctest: +SKIP
149     >>> x = Variable('x')  # doctest: +SKIP
150     >>> x.set(123)  # docttest: +SKIP
151     >>> x.get()  # docttest: +SKIP
152     123
153     >>> future = client.submit(f, x)  # doctest: +SKIP
154     >>> x.set(future)  # doctest: +SKIP
155 
156     See Also
157     --------
158     Queue: shared multi-producer/multi-consumer queue between clients
159     """
160 
161     def __init__(self, name=None, client=None, maxsize=0):
162         self.client = client or _get_global_client()
163         self.name = name or "variable-" + uuid.uuid4().hex
164 
165     async def _set(self, value):
166         if isinstance(value, Future):
167             await self.client.scheduler.variable_set(
168                 key=tokey(value.key), name=self.name
169             )
170         else:
171             await self.client.scheduler.variable_set(data=value, name=self.name)
172 
173     def set(self, value, **kwargs):
174         """ Set the value of this variable
175 
176         Parameters
177         ----------
178         value: Future or object
179             Must be either a Future or a msgpack-encodable value
180         """
181         return self.client.sync(self._set, value, **kwargs)
182 
183     async def _get(self, timeout=None):
184         d = await self.client.scheduler.variable_get(
185             timeout=timeout, name=self.name, client=self.client.id
186         )
187         if d["type"] == "Future":
188             value = Future(d["value"], self.client, inform=True, state=d["state"])
189             if d["state"] == "erred":
190                 value._state.set_error(d["exception"], d["traceback"])
191             self.client._send_to_scheduler(
192                 {
193                     "op": "variable-future-release",
194                     "name": self.name,
195                     "key": d["value"],
196                     "token": d["token"],
197                 }
198             )
199         else:
200             value = d["value"]
201         return value
202 
203     def get(self, timeout=None, **kwargs):
204         """ Get the value of this variable """
205         return self.client.sync(self._get, timeout=timeout, **kwargs)
206 
207     def delete(self):
208         """ Delete this variable
209 
210         Caution, this affects all clients currently pointing to this variable.
211         """
212         if self.client.status == "running":  # TODO: can leave zombie futures
213             self.client._send_to_scheduler({"op": "variable_delete", "name": self.name})
214 
215     def __getstate__(self):
216         return (self.name, self.client.scheduler.address)
217 
218     def __setstate__(self, state):
219         name, address = state
220         try:
221             client = get_client(address)
222             assert client.scheduler.address == address
223         except (AttributeError, AssertionError):
224             client = Client(address, set_as_default=False)
225         self.__init__(name=name, client=client)
226
```
--- 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/variable.py b/distributed/variable.py
--- a/distributed/variable.py
+++ b/distributed/variable.py
@@ -113,8 +113,10 @@
             else:
                 if old["type"] == "Future":
                     await self.release(old["value"], name)
-            del self.waiting_conditions[name]
-            del self.variables[name]
+            with ignoring(KeyError):
+                del self.waiting_conditions[name]
+            with ignoring(KeyError):
+                del self.variables[name]
 
 
 class Variable:

{"golden_diff": "diff --git a/distributed/variable.py b/distributed/variable.py\n--- a/distributed/variable.py\n+++ b/distributed/variable.py\n@@ -113,8 +113,10 @@\n             else:\n                 if old[\"type\"] == \"Future\":\n                     await self.release(old[\"value\"], name)\n-            del self.waiting_conditions[name]\n-            del self.variables[name]\n+            with ignoring(KeyError):\n+                del self.waiting_conditions[name]\n+            with ignoring(KeyError):\n+                del self.variables[name]\n \n \n class Variable:\n", "issue": "Variable delete raises KeyError\nIt seems that there is an unhandled KeyError in the scheduler variable delete logic.\r\n\r\nIf using the asynchronous API only visible in the scheduler logs but it does mean that Variables will be \"leaking\" and staying around instead of being deleted.\r\n\r\nIt happens regardless of whether you set a value.\r\n\r\nShortest reproduction I can provide is as follows:\r\n```python\r\nfrom distributed import Client\r\nfrom distributed import Variable\r\n\r\n\r\ndef main():\r\n    Client()\r\n    Variable().delete()\r\n\r\n\r\nif __name__ == '__main__':\r\n    main()\r\n```\r\n\r\n```python\r\ntornado.application - ERROR - Exception in callback functools.partial(<bound method IOLoop._discard_future_result of <tornado.platform.asyncio.AsyncIOLoop object at 0x7f8fe9265be0>>, <Task finished coro=<VariableExtension.delete() done, defined at ./venv/lib/python3.6/site-packages/distributed/variable.py:101> exception=KeyError('variable-68a51209410248fa9f69f84b25f48343',)>)\r\nTraceback (most recent call last):\r\n  File \"./venv/lib/python3.6/site-packages/tornado/ioloop.py\", line 743, in _run_callback\r\n    ret = callback()\r\n  File \"./venv/lib/python3.6/site-packages/tornado/ioloop.py\", line 767, in _discard_future_result\r\n    future.result()\r\n  File \"./venv/lib/python3.6/site-packages/distributed/variable.py\", line 110, in delete\r\n    del self.waiting_conditions[name]\r\nKeyError: 'variable-68a51209410248fa9f69f84b25f48343'\r\n```\r\n\r\nPython 3.6.10\r\nReproduced with distributed 2.9.3 and 2.10.0\n", "before_files": [{"content": "import asyncio\nfrom collections import defaultdict\nimport logging\nimport uuid\n\nfrom tlz import merge\n\nfrom .client import Future, _get_global_client, Client\nfrom .utils import tokey, log_errors, TimeoutError, ignoring\nfrom .worker import get_client\n\nlogger = logging.getLogger(__name__)\n\n\nclass VariableExtension:\n    \"\"\" An extension for the scheduler to manage queues\n\n    This adds the following routes to the scheduler\n\n    *  variable-set\n    *  variable-get\n    *  variable-delete\n    \"\"\"\n\n    def __init__(self, scheduler):\n        self.scheduler = scheduler\n        self.variables = dict()\n        self.waiting = defaultdict(set)\n        self.waiting_conditions = defaultdict(asyncio.Condition)\n        self.started = asyncio.Condition()\n\n        self.scheduler.handlers.update(\n            {\"variable_set\": self.set, \"variable_get\": self.get}\n        )\n\n        self.scheduler.stream_handlers[\"variable-future-release\"] = self.future_release\n        self.scheduler.stream_handlers[\"variable_delete\"] = self.delete\n\n        self.scheduler.extensions[\"variables\"] = self\n\n    async def set(self, stream=None, name=None, key=None, data=None, client=None):\n        if key is not None:\n            record = {\"type\": \"Future\", \"value\": key}\n            self.scheduler.client_desires_keys(keys=[key], client=\"variable-%s\" % name)\n        else:\n            record = {\"type\": \"msgpack\", \"value\": data}\n        try:\n            old = self.variables[name]\n        except KeyError:\n            pass\n        else:\n            if old[\"type\"] == \"Future\" and old[\"value\"] != key:\n                asyncio.ensure_future(self.release(old[\"value\"], name))\n        if name not in self.variables:\n            async with self.started:\n                self.started.notify_all()\n        self.variables[name] = record\n\n    async def release(self, key, name):\n        while self.waiting[key, name]:\n            async with self.waiting_conditions[name]:\n                await self.waiting_conditions[name].wait()\n\n        self.scheduler.client_releases_keys(keys=[key], client=\"variable-%s\" % name)\n        del self.waiting[key, name]\n\n    async def future_release(self, name=None, key=None, token=None, client=None):\n        self.waiting[key, name].remove(token)\n        if not self.waiting[key, name]:\n            async with self.waiting_conditions[name]:\n                self.waiting_conditions[name].notify_all()\n\n    async def get(self, stream=None, name=None, client=None, timeout=None):\n        start = self.scheduler.loop.time()\n        while name not in self.variables:\n            if timeout is not None:\n                left = timeout - (self.scheduler.loop.time() - start)\n            else:\n                left = None\n            if left and left < 0:\n                raise TimeoutError()\n            try:\n\n                async def _():  # Python 3.6 is odd and requires special help here\n                    await self.started.acquire()\n                    await self.started.wait()\n\n                await asyncio.wait_for(_(), timeout=left)\n            finally:\n                with ignoring(RuntimeError):  # Python 3.6 loses lock on finally clause\n                    self.started.release()\n\n        record = self.variables[name]\n        if record[\"type\"] == \"Future\":\n            key = record[\"value\"]\n            token = uuid.uuid4().hex\n            ts = self.scheduler.tasks.get(key)\n            state = ts.state if ts is not None else \"lost\"\n            msg = {\"token\": token, \"state\": state}\n            if state == \"erred\":\n                msg[\"exception\"] = ts.exception_blame.exception\n                msg[\"traceback\"] = ts.exception_blame.traceback\n            record = merge(record, msg)\n            self.waiting[key, name].add(token)\n        return record\n\n    async def delete(self, stream=None, name=None, client=None):\n        with log_errors():\n            try:\n                old = self.variables[name]\n            except KeyError:\n                pass\n            else:\n                if old[\"type\"] == \"Future\":\n                    await self.release(old[\"value\"], name)\n            del self.waiting_conditions[name]\n            del self.variables[name]\n\n\nclass Variable:\n    \"\"\" Distributed Global Variable\n\n    This allows multiple clients to share futures and data between each other\n    with a single mutable variable.  All metadata is sequentialized through the\n    scheduler.  Race conditions can occur.\n\n    Values must be either Futures or msgpack-encodable data (ints, lists,\n    strings, etc..)  All data will be kept and sent through the scheduler, so\n    it is wise not to send too much.  If you want to share a large amount of\n    data then ``scatter`` it and share the future instead.\n\n    .. warning::\n\n       This object is experimental and has known issues in Python 2\n\n    Parameters\n    ----------\n    name: string (optional)\n        Name used by other clients and the scheduler to identify the variable.\n        If not given, a random name will be generated.\n    client: Client (optional)\n        Client used for communication with the scheduler. Defaults to the\n        value of ``_get_global_client()``.\n\n    Examples\n    --------\n    >>> from dask.distributed import Client, Variable # doctest: +SKIP\n    >>> client = Client()  # doctest: +SKIP\n    >>> x = Variable('x')  # doctest: +SKIP\n    >>> x.set(123)  # docttest: +SKIP\n    >>> x.get()  # docttest: +SKIP\n    123\n    >>> future = client.submit(f, x)  # doctest: +SKIP\n    >>> x.set(future)  # doctest: +SKIP\n\n    See Also\n    --------\n    Queue: shared multi-producer/multi-consumer queue between clients\n    \"\"\"\n\n    def __init__(self, name=None, client=None, maxsize=0):\n        self.client = client or _get_global_client()\n        self.name = name or \"variable-\" + uuid.uuid4().hex\n\n    async def _set(self, value):\n        if isinstance(value, Future):\n            await self.client.scheduler.variable_set(\n                key=tokey(value.key), name=self.name\n            )\n        else:\n            await self.client.scheduler.variable_set(data=value, name=self.name)\n\n    def set(self, value, **kwargs):\n        \"\"\" Set the value of this variable\n\n        Parameters\n        ----------\n        value: Future or object\n            Must be either a Future or a msgpack-encodable value\n        \"\"\"\n        return self.client.sync(self._set, value, **kwargs)\n\n    async def _get(self, timeout=None):\n        d = await self.client.scheduler.variable_get(\n            timeout=timeout, name=self.name, client=self.client.id\n        )\n        if d[\"type\"] == \"Future\":\n            value = Future(d[\"value\"], self.client, inform=True, state=d[\"state\"])\n            if d[\"state\"] == \"erred\":\n                value._state.set_error(d[\"exception\"], d[\"traceback\"])\n            self.client._send_to_scheduler(\n                {\n                    \"op\": \"variable-future-release\",\n                    \"name\": self.name,\n                    \"key\": d[\"value\"],\n                    \"token\": d[\"token\"],\n                }\n            )\n        else:\n            value = d[\"value\"]\n        return value\n\n    def get(self, timeout=None, **kwargs):\n        \"\"\" Get the value of this variable \"\"\"\n        return self.client.sync(self._get, timeout=timeout, **kwargs)\n\n    def delete(self):\n        \"\"\" Delete this variable\n\n        Caution, this affects all clients currently pointing to this variable.\n        \"\"\"\n        if self.client.status == \"running\":  # TODO: can leave zombie futures\n            self.client._send_to_scheduler({\"op\": \"variable_delete\", \"name\": self.name})\n\n    def __getstate__(self):\n        return (self.name, self.client.scheduler.address)\n\n    def __setstate__(self, state):\n        name, address = state\n        try:\n            client = get_client(address)\n            assert client.scheduler.address == address\n        except (AttributeError, AssertionError):\n            client = Client(address, set_as_default=False)\n        self.__init__(name=name, client=client)\n", "path": "distributed/variable.py"}], "after_files": [{"content": "import asyncio\nfrom collections import defaultdict\nimport logging\nimport uuid\n\nfrom tlz import merge\n\nfrom .client import Future, _get_global_client, Client\nfrom .utils import tokey, log_errors, TimeoutError, ignoring\nfrom .worker import get_client\n\nlogger = logging.getLogger(__name__)\n\n\nclass VariableExtension:\n    \"\"\" An extension for the scheduler to manage queues\n\n    This adds the following routes to the scheduler\n\n    *  variable-set\n    *  variable-get\n    *  variable-delete\n    \"\"\"\n\n    def __init__(self, scheduler):\n        self.scheduler = scheduler\n        self.variables = dict()\n        self.waiting = defaultdict(set)\n        self.waiting_conditions = defaultdict(asyncio.Condition)\n        self.started = asyncio.Condition()\n\n        self.scheduler.handlers.update(\n            {\"variable_set\": self.set, \"variable_get\": self.get}\n        )\n\n        self.scheduler.stream_handlers[\"variable-future-release\"] = self.future_release\n        self.scheduler.stream_handlers[\"variable_delete\"] = self.delete\n\n        self.scheduler.extensions[\"variables\"] = self\n\n    async def set(self, stream=None, name=None, key=None, data=None, client=None):\n        if key is not None:\n            record = {\"type\": \"Future\", \"value\": key}\n            self.scheduler.client_desires_keys(keys=[key], client=\"variable-%s\" % name)\n        else:\n            record = {\"type\": \"msgpack\", \"value\": data}\n        try:\n            old = self.variables[name]\n        except KeyError:\n            pass\n        else:\n            if old[\"type\"] == \"Future\" and old[\"value\"] != key:\n                asyncio.ensure_future(self.release(old[\"value\"], name))\n        if name not in self.variables:\n            async with self.started:\n                self.started.notify_all()\n        self.variables[name] = record\n\n    async def release(self, key, name):\n        while self.waiting[key, name]:\n            async with self.waiting_conditions[name]:\n                await self.waiting_conditions[name].wait()\n\n        self.scheduler.client_releases_keys(keys=[key], client=\"variable-%s\" % name)\n        del self.waiting[key, name]\n\n    async def future_release(self, name=None, key=None, token=None, client=None):\n        self.waiting[key, name].remove(token)\n        if not self.waiting[key, name]:\n            async with self.waiting_conditions[name]:\n                self.waiting_conditions[name].notify_all()\n\n    async def get(self, stream=None, name=None, client=None, timeout=None):\n        start = self.scheduler.loop.time()\n        while name not in self.variables:\n            if timeout is not None:\n                left = timeout - (self.scheduler.loop.time() - start)\n            else:\n                left = None\n            if left and left < 0:\n                raise TimeoutError()\n            try:\n\n                async def _():  # Python 3.6 is odd and requires special help here\n                    await self.started.acquire()\n                    await self.started.wait()\n\n                await asyncio.wait_for(_(), timeout=left)\n            finally:\n                with ignoring(RuntimeError):  # Python 3.6 loses lock on finally clause\n                    self.started.release()\n\n        record = self.variables[name]\n        if record[\"type\"] == \"Future\":\n            key = record[\"value\"]\n            token = uuid.uuid4().hex\n            ts = self.scheduler.tasks.get(key)\n            state = ts.state if ts is not None else \"lost\"\n            msg = {\"token\": token, \"state\": state}\n            if state == \"erred\":\n                msg[\"exception\"] = ts.exception_blame.exception\n                msg[\"traceback\"] = ts.exception_blame.traceback\n            record = merge(record, msg)\n            self.waiting[key, name].add(token)\n        return record\n\n    async def delete(self, stream=None, name=None, client=None):\n        with log_errors():\n            try:\n                old = self.variables[name]\n            except KeyError:\n                pass\n            else:\n                if old[\"type\"] == \"Future\":\n                    await self.release(old[\"value\"], name)\n            with ignoring(KeyError):\n                del self.waiting_conditions[name]\n            with ignoring(KeyError):\n                del self.variables[name]\n\n\nclass Variable:\n    \"\"\" Distributed Global Variable\n\n    This allows multiple clients to share futures and data between each other\n    with a single mutable variable.  All metadata is sequentialized through the\n    scheduler.  Race conditions can occur.\n\n    Values must be either Futures or msgpack-encodable data (ints, lists,\n    strings, etc..)  All data will be kept and sent through the scheduler, so\n    it is wise not to send too much.  If you want to share a large amount of\n    data then ``scatter`` it and share the future instead.\n\n    .. warning::\n\n       This object is experimental and has known issues in Python 2\n\n    Parameters\n    ----------\n    name: string (optional)\n        Name used by other clients and the scheduler to identify the variable.\n        If not given, a random name will be generated.\n    client: Client (optional)\n        Client used for communication with the scheduler. Defaults to the\n        value of ``_get_global_client()``.\n\n    Examples\n    --------\n    >>> from dask.distributed import Client, Variable # doctest: +SKIP\n    >>> client = Client()  # doctest: +SKIP\n    >>> x = Variable('x')  # doctest: +SKIP\n    >>> x.set(123)  # docttest: +SKIP\n    >>> x.get()  # docttest: +SKIP\n    123\n    >>> future = client.submit(f, x)  # doctest: +SKIP\n    >>> x.set(future)  # doctest: +SKIP\n\n    See Also\n    --------\n    Queue: shared multi-producer/multi-consumer queue between clients\n    \"\"\"\n\n    def __init__(self, name=None, client=None, maxsize=0):\n        self.client = client or _get_global_client()\n        self.name = name or \"variable-\" + uuid.uuid4().hex\n\n    async def _set(self, value):\n        if isinstance(value, Future):\n            await self.client.scheduler.variable_set(\n                key=tokey(value.key), name=self.name\n            )\n        else:\n            await self.client.scheduler.variable_set(data=value, name=self.name)\n\n    def set(self, value, **kwargs):\n        \"\"\" Set the value of this variable\n\n        Parameters\n        ----------\n        value: Future or object\n            Must be either a Future or a msgpack-encodable value\n        \"\"\"\n        return self.client.sync(self._set, value, **kwargs)\n\n    async def _get(self, timeout=None):\n        d = await self.client.scheduler.variable_get(\n            timeout=timeout, name=self.name, client=self.client.id\n        )\n        if d[\"type\"] == \"Future\":\n            value = Future(d[\"value\"], self.client, inform=True, state=d[\"state\"])\n            if d[\"state\"] == \"erred\":\n                value._state.set_error(d[\"exception\"], d[\"traceback\"])\n            self.client._send_to_scheduler(\n                {\n                    \"op\": \"variable-future-release\",\n                    \"name\": self.name,\n                    \"key\": d[\"value\"],\n                    \"token\": d[\"token\"],\n                }\n            )\n        else:\n            value = d[\"value\"]\n        return value\n\n    def get(self, timeout=None, **kwargs):\n        \"\"\" Get the value of this variable \"\"\"\n        return self.client.sync(self._get, timeout=timeout, **kwargs)\n\n    def delete(self):\n        \"\"\" Delete this variable\n\n        Caution, this affects all clients currently pointing to this variable.\n        \"\"\"\n        if self.client.status == \"running\":  # TODO: can leave zombie futures\n            self.client._send_to_scheduler({\"op\": \"variable_delete\", \"name\": self.name})\n\n    def __getstate__(self):\n        return (self.name, self.client.scheduler.address)\n\n    def __setstate__(self, state):\n        name, address = state\n        try:\n            client = get_client(address)\n            assert client.scheduler.address == address\n        except (AttributeError, AssertionError):\n            client = Client(address, set_as_default=False)\n        self.__init__(name=name, client=client)\n", "path": "distributed/variable.py"}]}