Python code coverage for Lib/queue.py

#	count	content
1	n/a	'''A multi-producer, multi-consumer queue.'''
2	n/a
3	n/a	try:
4	n/a	import threading
5	n/a	except ImportError:
6	n/a	import dummy_threading as threading
7	n/a	from collections import deque
8	n/a	from heapq import heappush, heappop
9	n/a	from time import monotonic as time
10	n/a
11	n/a	__all__ = ['Empty', 'Full', 'Queue', 'PriorityQueue', 'LifoQueue']
12	n/a
13	n/a	class Empty(Exception):
14	n/a	'Exception raised by Queue.get(block=0)/get_nowait().'
15	n/a	pass
16	n/a
17	n/a	class Full(Exception):
18	n/a	'Exception raised by Queue.put(block=0)/put_nowait().'
19	n/a	pass
20	n/a
21	n/a	class Queue:
22	n/a	'''Create a queue object with a given maximum size.
23	n/a
24	n/a	If maxsize is <= 0, the queue size is infinite.
25	n/a	'''
26	n/a
27	n/a	def __init__(self, maxsize=0):
28	n/a	self.maxsize = maxsize
29	n/a	self._init(maxsize)
30	n/a
31	n/a	# mutex must be held whenever the queue is mutating. All methods
32	n/a	# that acquire mutex must release it before returning. mutex
33	n/a	# is shared between the three conditions, so acquiring and
34	n/a	# releasing the conditions also acquires and releases mutex.
35	n/a	self.mutex = threading.Lock()
36	n/a
37	n/a	# Notify not_empty whenever an item is added to the queue; a
38	n/a	# thread waiting to get is notified then.
39	n/a	self.not_empty = threading.Condition(self.mutex)
40	n/a
41	n/a	# Notify not_full whenever an item is removed from the queue;
42	n/a	# a thread waiting to put is notified then.
43	n/a	self.not_full = threading.Condition(self.mutex)
44	n/a
45	n/a	# Notify all_tasks_done whenever the number of unfinished tasks
46	n/a	# drops to zero; thread waiting to join() is notified to resume
47	n/a	self.all_tasks_done = threading.Condition(self.mutex)
48	n/a	self.unfinished_tasks = 0
49	n/a
50	n/a	def task_done(self):
51	n/a	'''Indicate that a formerly enqueued task is complete.
52	n/a
53	n/a	Used by Queue consumer threads. For each get() used to fetch a task,
54	n/a	a subsequent call to task_done() tells the queue that the processing
55	n/a	on the task is complete.
56	n/a
57	n/a	If a join() is currently blocking, it will resume when all items
58	n/a	have been processed (meaning that a task_done() call was received
59	n/a	for every item that had been put() into the queue).
60	n/a
61	n/a	Raises a ValueError if called more times than there were items
62	n/a	placed in the queue.
63	n/a	'''
64	n/a	with self.all_tasks_done:
65	n/a	unfinished = self.unfinished_tasks - 1
66	n/a	if unfinished <= 0:
67	n/a	if unfinished < 0:
68	n/a	raise ValueError('task_done() called too many times')
69	n/a	self.all_tasks_done.notify_all()
70	n/a	self.unfinished_tasks = unfinished
71	n/a
72	n/a	def join(self):
73	n/a	'''Blocks until all items in the Queue have been gotten and processed.
74	n/a
75	n/a	The count of unfinished tasks goes up whenever an item is added to the
76	n/a	queue. The count goes down whenever a consumer thread calls task_done()
77	n/a	to indicate the item was retrieved and all work on it is complete.
78	n/a
79	n/a	When the count of unfinished tasks drops to zero, join() unblocks.
80	n/a	'''
81	n/a	with self.all_tasks_done:
82	n/a	while self.unfinished_tasks:
83	n/a	self.all_tasks_done.wait()
84	n/a
85	n/a	def qsize(self):
86	n/a	'''Return the approximate size of the queue (not reliable!).'''
87	n/a	with self.mutex:
88	n/a	return self._qsize()
89	n/a
90	n/a	def empty(self):
91	n/a	'''Return True if the queue is empty, False otherwise (not reliable!).
92	n/a
93	n/a	This method is likely to be removed at some point. Use qsize() == 0
94	n/a	as a direct substitute, but be aware that either approach risks a race
95	n/a	condition where a queue can grow before the result of empty() or
96	n/a	qsize() can be used.
97	n/a
98	n/a	To create code that needs to wait for all queued tasks to be
99	n/a	completed, the preferred technique is to use the join() method.
100	n/a	'''
101	n/a	with self.mutex:
102	n/a	return not self._qsize()
103	n/a
104	n/a	def full(self):
105	n/a	'''Return True if the queue is full, False otherwise (not reliable!).
106	n/a
107	n/a	This method is likely to be removed at some point. Use qsize() >= n
108	n/a	as a direct substitute, but be aware that either approach risks a race
109	n/a	condition where a queue can shrink before the result of full() or
110	n/a	qsize() can be used.
111	n/a	'''
112	n/a	with self.mutex:
113	n/a	return 0 < self.maxsize <= self._qsize()
114	n/a
115	n/a	def put(self, item, block=True, timeout=None):
116	n/a	'''Put an item into the queue.
117	n/a
118	n/a	If optional args 'block' is true and 'timeout' is None (the default),
119	n/a	block if necessary until a free slot is available. If 'timeout' is
120	n/a	a non-negative number, it blocks at most 'timeout' seconds and raises
121	n/a	the Full exception if no free slot was available within that time.
122	n/a	Otherwise ('block' is false), put an item on the queue if a free slot
123	n/a	is immediately available, else raise the Full exception ('timeout'
124	n/a	is ignored in that case).
125	n/a	'''
126	n/a	with self.not_full:
127	n/a	if self.maxsize > 0:
128	n/a	if not block:
129	n/a	if self._qsize() >= self.maxsize:
130	n/a	raise Full
131	n/a	elif timeout is None:
132	n/a	while self._qsize() >= self.maxsize:
133	n/a	self.not_full.wait()
134	n/a	elif timeout < 0:
135	n/a	raise ValueError("'timeout' must be a non-negative number")
136	n/a	else:
137	n/a	endtime = time() + timeout
138	n/a	while self._qsize() >= self.maxsize:
139	n/a	remaining = endtime - time()
140	n/a	if remaining <= 0.0:
141	n/a	raise Full
142	n/a	self.not_full.wait(remaining)
143	n/a	self._put(item)
144	n/a	self.unfinished_tasks += 1
145	n/a	self.not_empty.notify()
146	n/a
147	n/a	def get(self, block=True, timeout=None):
148	n/a	'''Remove and return an item from the queue.
149	n/a
150	n/a	If optional args 'block' is true and 'timeout' is None (the default),
151	n/a	block if necessary until an item is available. If 'timeout' is
152	n/a	a non-negative number, it blocks at most 'timeout' seconds and raises
153	n/a	the Empty exception if no item was available within that time.
154	n/a	Otherwise ('block' is false), return an item if one is immediately
155	n/a	available, else raise the Empty exception ('timeout' is ignored
156	n/a	in that case).
157	n/a	'''
158	n/a	with self.not_empty:
159	n/a	if not block:
160	n/a	if not self._qsize():
161	n/a	raise Empty
162	n/a	elif timeout is None:
163	n/a	while not self._qsize():
164	n/a	self.not_empty.wait()
165	n/a	elif timeout < 0:
166	n/a	raise ValueError("'timeout' must be a non-negative number")
167	n/a	else:
168	n/a	endtime = time() + timeout
169	n/a	while not self._qsize():
170	n/a	remaining = endtime - time()
171	n/a	if remaining <= 0.0:
172	n/a	raise Empty
173	n/a	self.not_empty.wait(remaining)
174	n/a	item = self._get()
175	n/a	self.not_full.notify()
176	n/a	return item
177	n/a
178	n/a	def put_nowait(self, item):
179	n/a	'''Put an item into the queue without blocking.
180	n/a
181	n/a	Only enqueue the item if a free slot is immediately available.
182	n/a	Otherwise raise the Full exception.
183	n/a	'''
184	n/a	return self.put(item, block=False)
185	n/a
186	n/a	def get_nowait(self):
187	n/a	'''Remove and return an item from the queue without blocking.
188	n/a
189	n/a	Only get an item if one is immediately available. Otherwise
190	n/a	raise the Empty exception.
191	n/a	'''
192	n/a	return self.get(block=False)
193	n/a
194	n/a	# Override these methods to implement other queue organizations
195	n/a	# (e.g. stack or priority queue).
196	n/a	# These will only be called with appropriate locks held
197	n/a
198	n/a	# Initialize the queue representation
199	n/a	def _init(self, maxsize):
200	n/a	self.queue = deque()
201	n/a
202	n/a	def _qsize(self):
203	n/a	return len(self.queue)
204	n/a
205	n/a	# Put a new item in the queue
206	n/a	def _put(self, item):
207	n/a	self.queue.append(item)
208	n/a
209	n/a	# Get an item from the queue
210	n/a	def _get(self):
211	n/a	return self.queue.popleft()
212	n/a
213	n/a
214	n/a	class PriorityQueue(Queue):
215	n/a	'''Variant of Queue that retrieves open entries in priority order (lowest first).
216	n/a
217	n/a	Entries are typically tuples of the form: (priority number, data).
218	n/a	'''
219	n/a
220	n/a	def _init(self, maxsize):
221	n/a	self.queue = []
222	n/a
223	n/a	def _qsize(self):
224	n/a	return len(self.queue)
225	n/a
226	n/a	def _put(self, item):
227	n/a	heappush(self.queue, item)
228	n/a
229	n/a	def _get(self):
230	n/a	return heappop(self.queue)
231	n/a
232	n/a
233	n/a	class LifoQueue(Queue):
234	n/a	'''Variant of Queue that retrieves most recently added entries first.'''
235	n/a
236	n/a	def _init(self, maxsize):
237	n/a	self.queue = []
238	n/a
239	n/a	def _qsize(self):
240	n/a	return len(self.queue)
241	n/a
242	n/a	def _put(self, item):
243	n/a	self.queue.append(item)
244	n/a
245	n/a	def _get(self):
246	n/a	return self.queue.pop()