Python code coverage for Lib/asyncio/queues.py

#	count	content
1	n/a	"""Queues"""
2	n/a
3	n/a	__all__ = ['Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty']
4	n/a
5	n/a	import collections
6	n/a	import heapq
7	n/a
8	n/a	from . import compat
9	n/a	from . import events
10	n/a	from . import locks
11	n/a	from .coroutines import coroutine
12	n/a
13	n/a
14	n/a	class QueueEmpty(Exception):
15	n/a	"""Exception raised when Queue.get_nowait() is called on a Queue object
16	n/a	which is empty.
17	n/a	"""
18	n/a	pass
19	n/a
20	n/a
21	n/a	class QueueFull(Exception):
22	n/a	"""Exception raised when the Queue.put_nowait() method is called on a Queue
23	n/a	object which is full.
24	n/a	"""
25	n/a	pass
26	n/a
27	n/a
28	n/a	class Queue:
29	n/a	"""A queue, useful for coordinating producer and consumer coroutines.
30	n/a
31	n/a	If maxsize is less than or equal to zero, the queue size is infinite. If it
32	n/a	is an integer greater than 0, then "yield from put()" will block when the
33	n/a	queue reaches maxsize, until an item is removed by get().
34	n/a
35	n/a	Unlike the standard library Queue, you can reliably know this Queue's size
36	n/a	with qsize(), since your single-threaded asyncio application won't be
37	n/a	interrupted between calling qsize() and doing an operation on the Queue.
38	n/a	"""
39	n/a
40	n/a	def __init__(self, maxsize=0, *, loop=None):
41	n/a	if loop is None:
42	n/a	self._loop = events.get_event_loop()
43	n/a	else:
44	n/a	self._loop = loop
45	n/a	self._maxsize = maxsize
46	n/a
47	n/a	# Futures.
48	n/a	self._getters = collections.deque()
49	n/a	# Futures.
50	n/a	self._putters = collections.deque()
51	n/a	self._unfinished_tasks = 0
52	n/a	self._finished = locks.Event(loop=self._loop)
53	n/a	self._finished.set()
54	n/a	self._init(maxsize)
55	n/a
56	n/a	# These three are overridable in subclasses.
57	n/a
58	n/a	def _init(self, maxsize):
59	n/a	self._queue = collections.deque()
60	n/a
61	n/a	def _get(self):
62	n/a	return self._queue.popleft()
63	n/a
64	n/a	def _put(self, item):
65	n/a	self._queue.append(item)
66	n/a
67	n/a	# End of the overridable methods.
68	n/a
69	n/a	def _wakeup_next(self, waiters):
70	n/a	# Wake up the next waiter (if any) that isn't cancelled.
71	n/a	while waiters:
72	n/a	waiter = waiters.popleft()
73	n/a	if not waiter.done():
74	n/a	waiter.set_result(None)
75	n/a	break
76	n/a
77	n/a	def __repr__(self):
78	n/a	return '<{} at {:#x} {}>'.format(
79	n/a	type(self).__name__, id(self), self._format())
80	n/a
81	n/a	def __str__(self):
82	n/a	return '<{} {}>'.format(type(self).__name__, self._format())
83	n/a
84	n/a	def _format(self):
85	n/a	result = 'maxsize={!r}'.format(self._maxsize)
86	n/a	if getattr(self, '_queue', None):
87	n/a	result += ' _queue={!r}'.format(list(self._queue))
88	n/a	if self._getters:
89	n/a	result += ' _getters[{}]'.format(len(self._getters))
90	n/a	if self._putters:
91	n/a	result += ' _putters[{}]'.format(len(self._putters))
92	n/a	if self._unfinished_tasks:
93	n/a	result += ' tasks={}'.format(self._unfinished_tasks)
94	n/a	return result
95	n/a
96	n/a	def qsize(self):
97	n/a	"""Number of items in the queue."""
98	n/a	return len(self._queue)
99	n/a
100	n/a	@property
101	n/a	def maxsize(self):
102	n/a	"""Number of items allowed in the queue."""
103	n/a	return self._maxsize
104	n/a
105	n/a	def empty(self):
106	n/a	"""Return True if the queue is empty, False otherwise."""
107	n/a	return not self._queue
108	n/a
109	n/a	def full(self):
110	n/a	"""Return True if there are maxsize items in the queue.
111	n/a
112	n/a	Note: if the Queue was initialized with maxsize=0 (the default),
113	n/a	then full() is never True.
114	n/a	"""
115	n/a	if self._maxsize <= 0:
116	n/a	return False
117	n/a	else:
118	n/a	return self.qsize() >= self._maxsize
119	n/a
120	n/a	@coroutine
121	n/a	def put(self, item):
122	n/a	"""Put an item into the queue.
123	n/a
124	n/a	Put an item into the queue. If the queue is full, wait until a free
125	n/a	slot is available before adding item.
126	n/a
127	n/a	This method is a coroutine.
128	n/a	"""
129	n/a	while self.full():
130	n/a	putter = self._loop.create_future()
131	n/a	self._putters.append(putter)
132	n/a	try:
133	n/a	yield from putter
134	n/a	except:
135	n/a	putter.cancel() # Just in case putter is not done yet.
136	n/a	if not self.full() and not putter.cancelled():
137	n/a	# We were woken up by get_nowait(), but can't take
138	n/a	# the call. Wake up the next in line.
139	n/a	self._wakeup_next(self._putters)
140	n/a	raise
141	n/a	return self.put_nowait(item)
142	n/a
143	n/a	def put_nowait(self, item):
144	n/a	"""Put an item into the queue without blocking.
145	n/a
146	n/a	If no free slot is immediately available, raise QueueFull.
147	n/a	"""
148	n/a	if self.full():
149	n/a	raise QueueFull
150	n/a	self._put(item)
151	n/a	self._unfinished_tasks += 1
152	n/a	self._finished.clear()
153	n/a	self._wakeup_next(self._getters)
154	n/a
155	n/a	@coroutine
156	n/a	def get(self):
157	n/a	"""Remove and return an item from the queue.
158	n/a
159	n/a	If queue is empty, wait until an item is available.
160	n/a
161	n/a	This method is a coroutine.
162	n/a	"""
163	n/a	while self.empty():
164	n/a	getter = self._loop.create_future()
165	n/a	self._getters.append(getter)
166	n/a	try:
167	n/a	yield from getter
168	n/a	except:
169	n/a	getter.cancel() # Just in case getter is not done yet.
170	n/a	if not self.empty() and not getter.cancelled():
171	n/a	# We were woken up by put_nowait(), but can't take
172	n/a	# the call. Wake up the next in line.
173	n/a	self._wakeup_next(self._getters)
174	n/a	raise
175	n/a	return self.get_nowait()
176	n/a
177	n/a	def get_nowait(self):
178	n/a	"""Remove and return an item from the queue.
179	n/a
180	n/a	Return an item if one is immediately available, else raise QueueEmpty.
181	n/a	"""
182	n/a	if self.empty():
183	n/a	raise QueueEmpty
184	n/a	item = self._get()
185	n/a	self._wakeup_next(self._putters)
186	n/a	return item
187	n/a
188	n/a	def task_done(self):
189	n/a	"""Indicate that a formerly enqueued task is complete.
190	n/a
191	n/a	Used by queue consumers. For each get() used to fetch a task,
192	n/a	a subsequent call to task_done() tells the queue that the processing
193	n/a	on the task is complete.
194	n/a
195	n/a	If a join() is currently blocking, it will resume when all items have
196	n/a	been processed (meaning that a task_done() call was received for every
197	n/a	item that had been put() into the queue).
198	n/a
199	n/a	Raises ValueError if called more times than there were items placed in
200	n/a	the queue.
201	n/a	"""
202	n/a	if self._unfinished_tasks <= 0:
203	n/a	raise ValueError('task_done() called too many times')
204	n/a	self._unfinished_tasks -= 1
205	n/a	if self._unfinished_tasks == 0:
206	n/a	self._finished.set()
207	n/a
208	n/a	@coroutine
209	n/a	def join(self):
210	n/a	"""Block until all items in the queue have been gotten and processed.
211	n/a
212	n/a	The count of unfinished tasks goes up whenever an item is added to the
213	n/a	queue. The count goes down whenever a consumer calls task_done() to
214	n/a	indicate that the item was retrieved and all work on it is complete.
215	n/a	When the count of unfinished tasks drops to zero, join() unblocks.
216	n/a	"""
217	n/a	if self._unfinished_tasks > 0:
218	n/a	yield from self._finished.wait()
219	n/a
220	n/a
221	n/a	class PriorityQueue(Queue):
222	n/a	"""A subclass of Queue; retrieves entries in priority order (lowest first).
223	n/a
224	n/a	Entries are typically tuples of the form: (priority number, data).
225	n/a	"""
226	n/a
227	n/a	def _init(self, maxsize):
228	n/a	self._queue = []
229	n/a
230	n/a	def _put(self, item, heappush=heapq.heappush):
231	n/a	heappush(self._queue, item)
232	n/a
233	n/a	def _get(self, heappop=heapq.heappop):
234	n/a	return heappop(self._queue)
235	n/a
236	n/a
237	n/a	class LifoQueue(Queue):
238	n/a	"""A subclass of Queue that retrieves most recently added entries first."""
239	n/a
240	n/a	def _init(self, maxsize):
241	n/a	self._queue = []
242	n/a
243	n/a	def _put(self, item):
244	n/a	self._queue.append(item)
245	n/a
246	n/a	def _get(self):
247	n/a	return self._queue.pop()
248	n/a
249	n/a
250	n/a	if not compat.PY35:
251	n/a	JoinableQueue = Queue
252	n/a	"""Deprecated alias for Queue."""
253	n/a	__all__.append('JoinableQueue')