We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Stop Queue from printing it's name on __init__
There's a print statement here:

https://github.com/dask/distributed/blob/8d07bf162b1f6434fc982f36360c5cedbac369a6/distributed/queues.py#L50

Would be nice to replace it with logging.debug()

--- 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/queues.py`
Content:
```
1 from collections import defaultdict
2 import datetime
3 import logging
4 import uuid
5 
6 import tornado.queues
7 from tornado.locks import Event
8 
9 from .client import Future, _get_global_client, Client
10 from .utils import tokey, sync, thread_state
11 from .worker import get_client
12 
13 logger = logging.getLogger(__name__)
14 
15 
16 class QueueExtension(object):
17     """ An extension for the scheduler to manage queues
18 
19     This adds the following routes to the scheduler
20 
21     *  queue_create
22     *  queue_release
23     *  queue_put
24     *  queue_get
25     *  queue_size
26     """
27 
28     def __init__(self, scheduler):
29         self.scheduler = scheduler
30         self.queues = dict()
31         self.client_refcount = dict()
32         self.future_refcount = defaultdict(lambda: 0)
33 
34         self.scheduler.handlers.update(
35             {
36                 "queue_create": self.create,
37                 "queue_put": self.put,
38                 "queue_get": self.get,
39                 "queue_qsize": self.qsize,
40             }
41         )
42 
43         self.scheduler.stream_handlers.update(
44             {"queue-future-release": self.future_release, "queue_release": self.release}
45         )
46 
47         self.scheduler.extensions["queues"] = self
48 
49     def create(self, stream=None, name=None, client=None, maxsize=0):
50         print("name", name)
51         if name not in self.queues:
52             self.queues[name] = tornado.queues.Queue(maxsize=maxsize)
53             self.client_refcount[name] = 1
54         else:
55             self.client_refcount[name] += 1
56 
57     def release(self, stream=None, name=None, client=None):
58         if name not in self.queues:
59             return
60 
61         self.client_refcount[name] -= 1
62         if self.client_refcount[name] == 0:
63             del self.client_refcount[name]
64             futures = self.queues[name]._queue
65             del self.queues[name]
66             keys = [d["value"] for d in futures if d["type"] == "Future"]
67             if keys:
68                 self.scheduler.client_releases_keys(keys=keys, client="queue-%s" % name)
69 
70     async def put(
71         self, stream=None, name=None, key=None, data=None, client=None, timeout=None
72     ):
73         if key is not None:
74             record = {"type": "Future", "value": key}
75             self.future_refcount[name, key] += 1
76             self.scheduler.client_desires_keys(keys=[key], client="queue-%s" % name)
77         else:
78             record = {"type": "msgpack", "value": data}
79         if timeout is not None:
80             timeout = datetime.timedelta(seconds=timeout)
81         await self.queues[name].put(record, timeout=timeout)
82 
83     def future_release(self, name=None, key=None, client=None):
84         self.future_refcount[name, key] -= 1
85         if self.future_refcount[name, key] == 0:
86             self.scheduler.client_releases_keys(keys=[key], client="queue-%s" % name)
87             del self.future_refcount[name, key]
88 
89     async def get(self, stream=None, name=None, client=None, timeout=None, batch=False):
90         def process(record):
91             """ Add task status if known """
92             if record["type"] == "Future":
93                 record = record.copy()
94                 key = record["value"]
95                 ts = self.scheduler.tasks.get(key)
96                 state = ts.state if ts is not None else "lost"
97 
98                 record["state"] = state
99                 if state == "erred":
100                     record["exception"] = ts.exception_blame.exception
101                     record["traceback"] = ts.exception_blame.traceback
102 
103             return record
104 
105         if batch:
106             q = self.queues[name]
107             out = []
108             if batch is True:
109                 while not q.empty():
110                     record = await q.get()
111                     out.append(record)
112             else:
113                 if timeout is not None:
114                     msg = (
115                         "Dask queues don't support simultaneous use of "
116                         "integer batch sizes and timeouts"
117                     )
118                     raise NotImplementedError(msg)
119                 for i in range(batch):
120                     record = await q.get()
121                     out.append(record)
122             out = [process(o) for o in out]
123             return out
124         else:
125             if timeout is not None:
126                 timeout = datetime.timedelta(seconds=timeout)
127             record = await self.queues[name].get(timeout=timeout)
128             record = process(record)
129             return record
130 
131     def qsize(self, stream=None, name=None, client=None):
132         return self.queues[name].qsize()
133 
134 
135 class Queue(object):
136     """ Distributed Queue
137 
138     This allows multiple clients to share futures or small bits of data between
139     each other with a multi-producer/multi-consumer queue.  All metadata is
140     sequentialized through the scheduler.
141 
142     Elements of the Queue must be either Futures or msgpack-encodable data
143     (ints, strings, lists, dicts).  All data is sent through the scheduler so
144     it is wise not to send large objects.  To share large objects scatter the
145     data and share the future instead.
146 
147     .. warning::
148 
149        This object is experimental and has known issues in Python 2
150 
151     Examples
152     --------
153     >>> from dask.distributed import Client, Queue  # doctest: +SKIP
154     >>> client = Client()  # doctest: +SKIP
155     >>> queue = Queue('x')  # doctest: +SKIP
156     >>> future = client.submit(f, x)  # doctest: +SKIP
157     >>> queue.put(future)  # doctest: +SKIP
158 
159     See Also
160     --------
161     Variable: shared variable between clients
162     """
163 
164     def __init__(self, name=None, client=None, maxsize=0):
165         self.client = client or _get_global_client()
166         self.name = name or "queue-" + uuid.uuid4().hex
167         self._event_started = Event()
168         if self.client.asynchronous or getattr(
169             thread_state, "on_event_loop_thread", False
170         ):
171 
172             async def _create_queue():
173                 await self.client.scheduler.queue_create(
174                     name=self.name, maxsize=maxsize
175                 )
176                 self._event_started.set()
177 
178             self.client.loop.add_callback(_create_queue)
179         else:
180             sync(
181                 self.client.loop,
182                 self.client.scheduler.queue_create,
183                 name=self.name,
184                 maxsize=maxsize,
185             )
186             self._event_started.set()
187 
188     def __await__(self):
189         async def _():
190             await self._event_started.wait()
191             return self
192 
193         return _().__await__()
194 
195     async def _put(self, value, timeout=None):
196         if isinstance(value, Future):
197             await self.client.scheduler.queue_put(
198                 key=tokey(value.key), timeout=timeout, name=self.name
199             )
200         else:
201             await self.client.scheduler.queue_put(
202                 data=value, timeout=timeout, name=self.name
203             )
204 
205     def put(self, value, timeout=None, **kwargs):
206         """ Put data into the queue """
207         return self.client.sync(self._put, value, timeout=timeout, **kwargs)
208 
209     def get(self, timeout=None, batch=False, **kwargs):
210         """ Get data from the queue
211 
212         Parameters
213         ----------
214         timeout: Number (optional)
215             Time in seconds to wait before timing out
216         batch: boolean, int (optional)
217             If True then return all elements currently waiting in the queue.
218             If an integer than return that many elements from the queue
219             If False (default) then return one item at a time
220          """
221         return self.client.sync(self._get, timeout=timeout, batch=batch, **kwargs)
222 
223     def qsize(self, **kwargs):
224         """ Current number of elements in the queue """
225         return self.client.sync(self._qsize, **kwargs)
226 
227     async def _get(self, timeout=None, batch=False):
228         resp = await self.client.scheduler.queue_get(
229             timeout=timeout, name=self.name, batch=batch
230         )
231 
232         def process(d):
233             if d["type"] == "Future":
234                 value = Future(d["value"], self.client, inform=True, state=d["state"])
235                 if d["state"] == "erred":
236                     value._state.set_error(d["exception"], d["traceback"])
237                 self.client._send_to_scheduler(
238                     {"op": "queue-future-release", "name": self.name, "key": d["value"]}
239                 )
240             else:
241                 value = d["value"]
242 
243             return value
244 
245         if batch is False:
246             result = process(resp)
247         else:
248             result = list(map(process, resp))
249 
250         return result
251 
252     async def _qsize(self):
253         result = await self.client.scheduler.queue_qsize(name=self.name)
254         return result
255 
256     def close(self):
257         if self.client.status == "running":  # TODO: can leave zombie futures
258             self.client._send_to_scheduler({"op": "queue_release", "name": self.name})
259 
260     def __getstate__(self):
261         return (self.name, self.client.scheduler.address)
262 
263     def __setstate__(self, state):
264         name, address = state
265         try:
266             client = get_client(address)
267             assert client.scheduler.address == address
268         except (AttributeError, AssertionError):
269             client = Client(address, set_as_default=False)
270         self.__init__(name=name, client=client)
271
```
--- 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/queues.py b/distributed/queues.py
--- a/distributed/queues.py
+++ b/distributed/queues.py
@@ -47,7 +47,7 @@
         self.scheduler.extensions["queues"] = self
 
     def create(self, stream=None, name=None, client=None, maxsize=0):
-        print("name", name)
+        logger.debug("Queue name: {}".format(name))
         if name not in self.queues:
             self.queues[name] = tornado.queues.Queue(maxsize=maxsize)
             self.client_refcount[name] = 1

{"golden_diff": "diff --git a/distributed/queues.py b/distributed/queues.py\n--- a/distributed/queues.py\n+++ b/distributed/queues.py\n@@ -47,7 +47,7 @@\n         self.scheduler.extensions[\"queues\"] = self\n \n     def create(self, stream=None, name=None, client=None, maxsize=0):\n-        print(\"name\", name)\n+        logger.debug(\"Queue name: {}\".format(name))\n         if name not in self.queues:\n             self.queues[name] = tornado.queues.Queue(maxsize=maxsize)\n             self.client_refcount[name] = 1\n", "issue": "Stop Queue from printing it's name on __init__\nThere's a print statement here:\r\n\r\nhttps://github.com/dask/distributed/blob/8d07bf162b1f6434fc982f36360c5cedbac369a6/distributed/queues.py#L50\r\n\r\nWould be nice to replace it with logging.debug()\n", "before_files": [{"content": "from collections import defaultdict\nimport datetime\nimport logging\nimport uuid\n\nimport tornado.queues\nfrom tornado.locks import Event\n\nfrom .client import Future, _get_global_client, Client\nfrom .utils import tokey, sync, thread_state\nfrom .worker import get_client\n\nlogger = logging.getLogger(__name__)\n\n\nclass QueueExtension(object):\n    \"\"\" An extension for the scheduler to manage queues\n\n    This adds the following routes to the scheduler\n\n    *  queue_create\n    *  queue_release\n    *  queue_put\n    *  queue_get\n    *  queue_size\n    \"\"\"\n\n    def __init__(self, scheduler):\n        self.scheduler = scheduler\n        self.queues = dict()\n        self.client_refcount = dict()\n        self.future_refcount = defaultdict(lambda: 0)\n\n        self.scheduler.handlers.update(\n            {\n                \"queue_create\": self.create,\n                \"queue_put\": self.put,\n                \"queue_get\": self.get,\n                \"queue_qsize\": self.qsize,\n            }\n        )\n\n        self.scheduler.stream_handlers.update(\n            {\"queue-future-release\": self.future_release, \"queue_release\": self.release}\n        )\n\n        self.scheduler.extensions[\"queues\"] = self\n\n    def create(self, stream=None, name=None, client=None, maxsize=0):\n        print(\"name\", name)\n        if name not in self.queues:\n            self.queues[name] = tornado.queues.Queue(maxsize=maxsize)\n            self.client_refcount[name] = 1\n        else:\n            self.client_refcount[name] += 1\n\n    def release(self, stream=None, name=None, client=None):\n        if name not in self.queues:\n            return\n\n        self.client_refcount[name] -= 1\n        if self.client_refcount[name] == 0:\n            del self.client_refcount[name]\n            futures = self.queues[name]._queue\n            del self.queues[name]\n            keys = [d[\"value\"] for d in futures if d[\"type\"] == \"Future\"]\n            if keys:\n                self.scheduler.client_releases_keys(keys=keys, client=\"queue-%s\" % name)\n\n    async def put(\n        self, stream=None, name=None, key=None, data=None, client=None, timeout=None\n    ):\n        if key is not None:\n            record = {\"type\": \"Future\", \"value\": key}\n            self.future_refcount[name, key] += 1\n            self.scheduler.client_desires_keys(keys=[key], client=\"queue-%s\" % name)\n        else:\n            record = {\"type\": \"msgpack\", \"value\": data}\n        if timeout is not None:\n            timeout = datetime.timedelta(seconds=timeout)\n        await self.queues[name].put(record, timeout=timeout)\n\n    def future_release(self, name=None, key=None, client=None):\n        self.future_refcount[name, key] -= 1\n        if self.future_refcount[name, key] == 0:\n            self.scheduler.client_releases_keys(keys=[key], client=\"queue-%s\" % name)\n            del self.future_refcount[name, key]\n\n    async def get(self, stream=None, name=None, client=None, timeout=None, batch=False):\n        def process(record):\n            \"\"\" Add task status if known \"\"\"\n            if record[\"type\"] == \"Future\":\n                record = record.copy()\n                key = record[\"value\"]\n                ts = self.scheduler.tasks.get(key)\n                state = ts.state if ts is not None else \"lost\"\n\n                record[\"state\"] = state\n                if state == \"erred\":\n                    record[\"exception\"] = ts.exception_blame.exception\n                    record[\"traceback\"] = ts.exception_blame.traceback\n\n            return record\n\n        if batch:\n            q = self.queues[name]\n            out = []\n            if batch is True:\n                while not q.empty():\n                    record = await q.get()\n                    out.append(record)\n            else:\n                if timeout is not None:\n                    msg = (\n                        \"Dask queues don't support simultaneous use of \"\n                        \"integer batch sizes and timeouts\"\n                    )\n                    raise NotImplementedError(msg)\n                for i in range(batch):\n                    record = await q.get()\n                    out.append(record)\n            out = [process(o) for o in out]\n            return out\n        else:\n            if timeout is not None:\n                timeout = datetime.timedelta(seconds=timeout)\n            record = await self.queues[name].get(timeout=timeout)\n            record = process(record)\n            return record\n\n    def qsize(self, stream=None, name=None, client=None):\n        return self.queues[name].qsize()\n\n\nclass Queue(object):\n    \"\"\" Distributed Queue\n\n    This allows multiple clients to share futures or small bits of data between\n    each other with a multi-producer/multi-consumer queue.  All metadata is\n    sequentialized through the scheduler.\n\n    Elements of the Queue must be either Futures or msgpack-encodable data\n    (ints, strings, lists, dicts).  All data is sent through the scheduler so\n    it is wise not to send large objects.  To share large objects scatter the\n    data and share the future instead.\n\n    .. warning::\n\n       This object is experimental and has known issues in Python 2\n\n    Examples\n    --------\n    >>> from dask.distributed import Client, Queue  # doctest: +SKIP\n    >>> client = Client()  # doctest: +SKIP\n    >>> queue = Queue('x')  # doctest: +SKIP\n    >>> future = client.submit(f, x)  # doctest: +SKIP\n    >>> queue.put(future)  # doctest: +SKIP\n\n    See Also\n    --------\n    Variable: shared variable 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 \"queue-\" + uuid.uuid4().hex\n        self._event_started = Event()\n        if self.client.asynchronous or getattr(\n            thread_state, \"on_event_loop_thread\", False\n        ):\n\n            async def _create_queue():\n                await self.client.scheduler.queue_create(\n                    name=self.name, maxsize=maxsize\n                )\n                self._event_started.set()\n\n            self.client.loop.add_callback(_create_queue)\n        else:\n            sync(\n                self.client.loop,\n                self.client.scheduler.queue_create,\n                name=self.name,\n                maxsize=maxsize,\n            )\n            self._event_started.set()\n\n    def __await__(self):\n        async def _():\n            await self._event_started.wait()\n            return self\n\n        return _().__await__()\n\n    async def _put(self, value, timeout=None):\n        if isinstance(value, Future):\n            await self.client.scheduler.queue_put(\n                key=tokey(value.key), timeout=timeout, name=self.name\n            )\n        else:\n            await self.client.scheduler.queue_put(\n                data=value, timeout=timeout, name=self.name\n            )\n\n    def put(self, value, timeout=None, **kwargs):\n        \"\"\" Put data into the queue \"\"\"\n        return self.client.sync(self._put, value, timeout=timeout, **kwargs)\n\n    def get(self, timeout=None, batch=False, **kwargs):\n        \"\"\" Get data from the queue\n\n        Parameters\n        ----------\n        timeout: Number (optional)\n            Time in seconds to wait before timing out\n        batch: boolean, int (optional)\n            If True then return all elements currently waiting in the queue.\n            If an integer than return that many elements from the queue\n            If False (default) then return one item at a time\n         \"\"\"\n        return self.client.sync(self._get, timeout=timeout, batch=batch, **kwargs)\n\n    def qsize(self, **kwargs):\n        \"\"\" Current number of elements in the queue \"\"\"\n        return self.client.sync(self._qsize, **kwargs)\n\n    async def _get(self, timeout=None, batch=False):\n        resp = await self.client.scheduler.queue_get(\n            timeout=timeout, name=self.name, batch=batch\n        )\n\n        def process(d):\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                    {\"op\": \"queue-future-release\", \"name\": self.name, \"key\": d[\"value\"]}\n                )\n            else:\n                value = d[\"value\"]\n\n            return value\n\n        if batch is False:\n            result = process(resp)\n        else:\n            result = list(map(process, resp))\n\n        return result\n\n    async def _qsize(self):\n        result = await self.client.scheduler.queue_qsize(name=self.name)\n        return result\n\n    def close(self):\n        if self.client.status == \"running\":  # TODO: can leave zombie futures\n            self.client._send_to_scheduler({\"op\": \"queue_release\", \"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/queues.py"}], "after_files": [{"content": "from collections import defaultdict\nimport datetime\nimport logging\nimport uuid\n\nimport tornado.queues\nfrom tornado.locks import Event\n\nfrom .client import Future, _get_global_client, Client\nfrom .utils import tokey, sync, thread_state\nfrom .worker import get_client\n\nlogger = logging.getLogger(__name__)\n\n\nclass QueueExtension(object):\n    \"\"\" An extension for the scheduler to manage queues\n\n    This adds the following routes to the scheduler\n\n    *  queue_create\n    *  queue_release\n    *  queue_put\n    *  queue_get\n    *  queue_size\n    \"\"\"\n\n    def __init__(self, scheduler):\n        self.scheduler = scheduler\n        self.queues = dict()\n        self.client_refcount = dict()\n        self.future_refcount = defaultdict(lambda: 0)\n\n        self.scheduler.handlers.update(\n            {\n                \"queue_create\": self.create,\n                \"queue_put\": self.put,\n                \"queue_get\": self.get,\n                \"queue_qsize\": self.qsize,\n            }\n        )\n\n        self.scheduler.stream_handlers.update(\n            {\"queue-future-release\": self.future_release, \"queue_release\": self.release}\n        )\n\n        self.scheduler.extensions[\"queues\"] = self\n\n    def create(self, stream=None, name=None, client=None, maxsize=0):\n        logger.debug(\"Queue name: {}\".format(name))\n        if name not in self.queues:\n            self.queues[name] = tornado.queues.Queue(maxsize=maxsize)\n            self.client_refcount[name] = 1\n        else:\n            self.client_refcount[name] += 1\n\n    def release(self, stream=None, name=None, client=None):\n        if name not in self.queues:\n            return\n\n        self.client_refcount[name] -= 1\n        if self.client_refcount[name] == 0:\n            del self.client_refcount[name]\n            futures = self.queues[name]._queue\n            del self.queues[name]\n            keys = [d[\"value\"] for d in futures if d[\"type\"] == \"Future\"]\n            if keys:\n                self.scheduler.client_releases_keys(keys=keys, client=\"queue-%s\" % name)\n\n    async def put(\n        self, stream=None, name=None, key=None, data=None, client=None, timeout=None\n    ):\n        if key is not None:\n            record = {\"type\": \"Future\", \"value\": key}\n            self.future_refcount[name, key] += 1\n            self.scheduler.client_desires_keys(keys=[key], client=\"queue-%s\" % name)\n        else:\n            record = {\"type\": \"msgpack\", \"value\": data}\n        if timeout is not None:\n            timeout = datetime.timedelta(seconds=timeout)\n        await self.queues[name].put(record, timeout=timeout)\n\n    def future_release(self, name=None, key=None, client=None):\n        self.future_refcount[name, key] -= 1\n        if self.future_refcount[name, key] == 0:\n            self.scheduler.client_releases_keys(keys=[key], client=\"queue-%s\" % name)\n            del self.future_refcount[name, key]\n\n    async def get(self, stream=None, name=None, client=None, timeout=None, batch=False):\n        def process(record):\n            \"\"\" Add task status if known \"\"\"\n            if record[\"type\"] == \"Future\":\n                record = record.copy()\n                key = record[\"value\"]\n                ts = self.scheduler.tasks.get(key)\n                state = ts.state if ts is not None else \"lost\"\n\n                record[\"state\"] = state\n                if state == \"erred\":\n                    record[\"exception\"] = ts.exception_blame.exception\n                    record[\"traceback\"] = ts.exception_blame.traceback\n\n            return record\n\n        if batch:\n            q = self.queues[name]\n            out = []\n            if batch is True:\n                while not q.empty():\n                    record = await q.get()\n                    out.append(record)\n            else:\n                if timeout is not None:\n                    msg = (\n                        \"Dask queues don't support simultaneous use of \"\n                        \"integer batch sizes and timeouts\"\n                    )\n                    raise NotImplementedError(msg)\n                for i in range(batch):\n                    record = await q.get()\n                    out.append(record)\n            out = [process(o) for o in out]\n            return out\n        else:\n            if timeout is not None:\n                timeout = datetime.timedelta(seconds=timeout)\n            record = await self.queues[name].get(timeout=timeout)\n            record = process(record)\n            return record\n\n    def qsize(self, stream=None, name=None, client=None):\n        return self.queues[name].qsize()\n\n\nclass Queue(object):\n    \"\"\" Distributed Queue\n\n    This allows multiple clients to share futures or small bits of data between\n    each other with a multi-producer/multi-consumer queue.  All metadata is\n    sequentialized through the scheduler.\n\n    Elements of the Queue must be either Futures or msgpack-encodable data\n    (ints, strings, lists, dicts).  All data is sent through the scheduler so\n    it is wise not to send large objects.  To share large objects scatter the\n    data and share the future instead.\n\n    .. warning::\n\n       This object is experimental and has known issues in Python 2\n\n    Examples\n    --------\n    >>> from dask.distributed import Client, Queue  # doctest: +SKIP\n    >>> client = Client()  # doctest: +SKIP\n    >>> queue = Queue('x')  # doctest: +SKIP\n    >>> future = client.submit(f, x)  # doctest: +SKIP\n    >>> queue.put(future)  # doctest: +SKIP\n\n    See Also\n    --------\n    Variable: shared variable 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 \"queue-\" + uuid.uuid4().hex\n        self._event_started = Event()\n        if self.client.asynchronous or getattr(\n            thread_state, \"on_event_loop_thread\", False\n        ):\n\n            async def _create_queue():\n                await self.client.scheduler.queue_create(\n                    name=self.name, maxsize=maxsize\n                )\n                self._event_started.set()\n\n            self.client.loop.add_callback(_create_queue)\n        else:\n            sync(\n                self.client.loop,\n                self.client.scheduler.queue_create,\n                name=self.name,\n                maxsize=maxsize,\n            )\n            self._event_started.set()\n\n    def __await__(self):\n        async def _():\n            await self._event_started.wait()\n            return self\n\n        return _().__await__()\n\n    async def _put(self, value, timeout=None):\n        if isinstance(value, Future):\n            await self.client.scheduler.queue_put(\n                key=tokey(value.key), timeout=timeout, name=self.name\n            )\n        else:\n            await self.client.scheduler.queue_put(\n                data=value, timeout=timeout, name=self.name\n            )\n\n    def put(self, value, timeout=None, **kwargs):\n        \"\"\" Put data into the queue \"\"\"\n        return self.client.sync(self._put, value, timeout=timeout, **kwargs)\n\n    def get(self, timeout=None, batch=False, **kwargs):\n        \"\"\" Get data from the queue\n\n        Parameters\n        ----------\n        timeout: Number (optional)\n            Time in seconds to wait before timing out\n        batch: boolean, int (optional)\n            If True then return all elements currently waiting in the queue.\n            If an integer than return that many elements from the queue\n            If False (default) then return one item at a time\n         \"\"\"\n        return self.client.sync(self._get, timeout=timeout, batch=batch, **kwargs)\n\n    def qsize(self, **kwargs):\n        \"\"\" Current number of elements in the queue \"\"\"\n        return self.client.sync(self._qsize, **kwargs)\n\n    async def _get(self, timeout=None, batch=False):\n        resp = await self.client.scheduler.queue_get(\n            timeout=timeout, name=self.name, batch=batch\n        )\n\n        def process(d):\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                    {\"op\": \"queue-future-release\", \"name\": self.name, \"key\": d[\"value\"]}\n                )\n            else:\n                value = d[\"value\"]\n\n            return value\n\n        if batch is False:\n            result = process(resp)\n        else:\n            result = list(map(process, resp))\n\n        return result\n\n    async def _qsize(self):\n        result = await self.client.scheduler.queue_qsize(name=self.name)\n        return result\n\n    def close(self):\n        if self.client.status == \"running\":  # TODO: can leave zombie futures\n            self.client._send_to_scheduler({\"op\": \"queue_release\", \"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/queues.py"}]}