Python code coverage for Lib/asyncio/locks.py

#	count	content
1	n/a	"""Synchronization primitives."""
2	n/a
3	n/a	__all__ = ['Lock', 'Event', 'Condition', 'Semaphore', 'BoundedSemaphore']
4	n/a
5	n/a	import collections
6	n/a
7	n/a	from . import compat
8	n/a	from . import events
9	n/a	from . import futures
10	n/a	from .coroutines import coroutine
11	n/a
12	n/a
13	n/a	class _ContextManager:
14	n/a	"""Context manager.
15	n/a
16	n/a	This enables the following idiom for acquiring and releasing a
17	n/a	lock around a block:
18	n/a
19	n/a	with (yield from lock):
20	n/a	<block>
21	n/a
22	n/a	while failing loudly when accidentally using:
23	n/a
24	n/a	with lock:
25	n/a	<block>
26	n/a	"""
27	n/a
28	n/a	def __init__(self, lock):
29	n/a	self._lock = lock
30	n/a
31	n/a	def __enter__(self):
32	n/a	# We have no use for the "as ..." clause in the with
33	n/a	# statement for locks.
34	n/a	return None
35	n/a
36	n/a	def __exit__(self, *args):
37	n/a	try:
38	n/a	self._lock.release()
39	n/a	finally:
40	n/a	self._lock = None # Crudely prevent reuse.
41	n/a
42	n/a
43	n/a	class _ContextManagerMixin:
44	n/a	def __enter__(self):
45	n/a	raise RuntimeError(
46	n/a	'"yield from" should be used as context manager expression')
47	n/a
48	n/a	def __exit__(self, *args):
49	n/a	# This must exist because __enter__ exists, even though that
50	n/a	# always raises; that's how the with-statement works.
51	n/a	pass
52	n/a
53	n/a	@coroutine
54	n/a	def __iter__(self):
55	n/a	# This is not a coroutine. It is meant to enable the idiom:
56	n/a	#
57	n/a	# with (yield from lock):
58	n/a	# <block>
59	n/a	#
60	n/a	# as an alternative to:
61	n/a	#
62	n/a	# yield from lock.acquire()
63	n/a	# try:
64	n/a	# <block>
65	n/a	# finally:
66	n/a	# lock.release()
67	n/a	yield from self.acquire()
68	n/a	return _ContextManager(self)
69	n/a
70	n/a	if compat.PY35:
71	n/a
72	n/a	def __await__(self):
73	n/a	# To make "with await lock" work.
74	n/a	yield from self.acquire()
75	n/a	return _ContextManager(self)
76	n/a
77	n/a	@coroutine
78	n/a	def __aenter__(self):
79	n/a	yield from self.acquire()
80	n/a	# We have no use for the "as ..." clause in the with
81	n/a	# statement for locks.
82	n/a	return None
83	n/a
84	n/a	@coroutine
85	n/a	def __aexit__(self, exc_type, exc, tb):
86	n/a	self.release()
87	n/a
88	n/a
89	n/a	class Lock(_ContextManagerMixin):
90	n/a	"""Primitive lock objects.
91	n/a
92	n/a	A primitive lock is a synchronization primitive that is not owned
93	n/a	by a particular coroutine when locked. A primitive lock is in one
94	n/a	of two states, 'locked' or 'unlocked'.
95	n/a
96	n/a	It is created in the unlocked state. It has two basic methods,
97	n/a	acquire() and release(). When the state is unlocked, acquire()
98	n/a	changes the state to locked and returns immediately. When the
99	n/a	state is locked, acquire() blocks until a call to release() in
100	n/a	another coroutine changes it to unlocked, then the acquire() call
101	n/a	resets it to locked and returns. The release() method should only
102	n/a	be called in the locked state; it changes the state to unlocked
103	n/a	and returns immediately. If an attempt is made to release an
104	n/a	unlocked lock, a RuntimeError will be raised.
105	n/a
106	n/a	When more than one coroutine is blocked in acquire() waiting for
107	n/a	the state to turn to unlocked, only one coroutine proceeds when a
108	n/a	release() call resets the state to unlocked; first coroutine which
109	n/a	is blocked in acquire() is being processed.
110	n/a
111	n/a	acquire() is a coroutine and should be called with 'yield from'.
112	n/a
113	n/a	Locks also support the context management protocol. '(yield from lock)'
114	n/a	should be used as the context manager expression.
115	n/a
116	n/a	Usage:
117	n/a
118	n/a	lock = Lock()
119	n/a	...
120	n/a	yield from lock
121	n/a	try:
122	n/a	...
123	n/a	finally:
124	n/a	lock.release()
125	n/a
126	n/a	Context manager usage:
127	n/a
128	n/a	lock = Lock()
129	n/a	...
130	n/a	with (yield from lock):
131	n/a	...
132	n/a
133	n/a	Lock objects can be tested for locking state:
134	n/a
135	n/a	if not lock.locked():
136	n/a	yield from lock
137	n/a	else:
138	n/a	# lock is acquired
139	n/a	...
140	n/a
141	n/a	"""
142	n/a
143	n/a	def __init__(self, *, loop=None):
144	n/a	self._waiters = collections.deque()
145	n/a	self._locked = False
146	n/a	if loop is not None:
147	n/a	self._loop = loop
148	n/a	else:
149	n/a	self._loop = events.get_event_loop()
150	n/a
151	n/a	def __repr__(self):
152	n/a	res = super().__repr__()
153	n/a	extra = 'locked' if self._locked else 'unlocked'
154	n/a	if self._waiters:
155	n/a	extra = '{},waiters:{}'.format(extra, len(self._waiters))
156	n/a	return '<{} [{}]>'.format(res[1:-1], extra)
157	n/a
158	n/a	def locked(self):
159	n/a	"""Return True if lock is acquired."""
160	n/a	return self._locked
161	n/a
162	n/a	@coroutine
163	n/a	def acquire(self):
164	n/a	"""Acquire a lock.
165	n/a
166	n/a	This method blocks until the lock is unlocked, then sets it to
167	n/a	locked and returns True.
168	n/a	"""
169	n/a	if not self._locked and all(w.cancelled() for w in self._waiters):
170	n/a	self._locked = True
171	n/a	return True
172	n/a
173	n/a	fut = self._loop.create_future()
174	n/a	self._waiters.append(fut)
175	n/a	try:
176	n/a	yield from fut
177	n/a	self._locked = True
178	n/a	return True
179	n/a	finally:
180	n/a	self._waiters.remove(fut)
181	n/a
182	n/a	def release(self):
183	n/a	"""Release a lock.
184	n/a
185	n/a	When the lock is locked, reset it to unlocked, and return.
186	n/a	If any other coroutines are blocked waiting for the lock to become
187	n/a	unlocked, allow exactly one of them to proceed.
188	n/a
189	n/a	When invoked on an unlocked lock, a RuntimeError is raised.
190	n/a
191	n/a	There is no return value.
192	n/a	"""
193	n/a	if self._locked:
194	n/a	self._locked = False
195	n/a	# Wake up the first waiter who isn't cancelled.
196	n/a	for fut in self._waiters:
197	n/a	if not fut.done():
198	n/a	fut.set_result(True)
199	n/a	break
200	n/a	else:
201	n/a	raise RuntimeError('Lock is not acquired.')
202	n/a
203	n/a
204	n/a	class Event:
205	n/a	"""Asynchronous equivalent to threading.Event.
206	n/a
207	n/a	Class implementing event objects. An event manages a flag that can be set
208	n/a	to true with the set() method and reset to false with the clear() method.
209	n/a	The wait() method blocks until the flag is true. The flag is initially
210	n/a	false.
211	n/a	"""
212	n/a
213	n/a	def __init__(self, *, loop=None):
214	n/a	self._waiters = collections.deque()
215	n/a	self._value = False
216	n/a	if loop is not None:
217	n/a	self._loop = loop
218	n/a	else:
219	n/a	self._loop = events.get_event_loop()
220	n/a
221	n/a	def __repr__(self):
222	n/a	res = super().__repr__()
223	n/a	extra = 'set' if self._value else 'unset'
224	n/a	if self._waiters:
225	n/a	extra = '{},waiters:{}'.format(extra, len(self._waiters))
226	n/a	return '<{} [{}]>'.format(res[1:-1], extra)
227	n/a
228	n/a	def is_set(self):
229	n/a	"""Return True if and only if the internal flag is true."""
230	n/a	return self._value
231	n/a
232	n/a	def set(self):
233	n/a	"""Set the internal flag to true. All coroutines waiting for it to
234	n/a	become true are awakened. Coroutine that call wait() once the flag is
235	n/a	true will not block at all.
236	n/a	"""
237	n/a	if not self._value:
238	n/a	self._value = True
239	n/a
240	n/a	for fut in self._waiters:
241	n/a	if not fut.done():
242	n/a	fut.set_result(True)
243	n/a
244	n/a	def clear(self):
245	n/a	"""Reset the internal flag to false. Subsequently, coroutines calling
246	n/a	wait() will block until set() is called to set the internal flag
247	n/a	to true again."""
248	n/a	self._value = False
249	n/a
250	n/a	@coroutine
251	n/a	def wait(self):
252	n/a	"""Block until the internal flag is true.
253	n/a
254	n/a	If the internal flag is true on entry, return True
255	n/a	immediately. Otherwise, block until another coroutine calls
256	n/a	set() to set the flag to true, then return True.
257	n/a	"""
258	n/a	if self._value:
259	n/a	return True
260	n/a
261	n/a	fut = self._loop.create_future()
262	n/a	self._waiters.append(fut)
263	n/a	try:
264	n/a	yield from fut
265	n/a	return True
266	n/a	finally:
267	n/a	self._waiters.remove(fut)
268	n/a
269	n/a
270	n/a	class Condition(_ContextManagerMixin):
271	n/a	"""Asynchronous equivalent to threading.Condition.
272	n/a
273	n/a	This class implements condition variable objects. A condition variable
274	n/a	allows one or more coroutines to wait until they are notified by another
275	n/a	coroutine.
276	n/a
277	n/a	A new Lock object is created and used as the underlying lock.
278	n/a	"""
279	n/a
280	n/a	def __init__(self, lock=None, *, loop=None):
281	n/a	if loop is not None:
282	n/a	self._loop = loop
283	n/a	else:
284	n/a	self._loop = events.get_event_loop()
285	n/a
286	n/a	if lock is None:
287	n/a	lock = Lock(loop=self._loop)
288	n/a	elif lock._loop is not self._loop:
289	n/a	raise ValueError("loop argument must agree with lock")
290	n/a
291	n/a	self._lock = lock
292	n/a	# Export the lock's locked(), acquire() and release() methods.
293	n/a	self.locked = lock.locked
294	n/a	self.acquire = lock.acquire
295	n/a	self.release = lock.release
296	n/a
297	n/a	self._waiters = collections.deque()
298	n/a
299	n/a	def __repr__(self):
300	n/a	res = super().__repr__()
301	n/a	extra = 'locked' if self.locked() else 'unlocked'
302	n/a	if self._waiters:
303	n/a	extra = '{},waiters:{}'.format(extra, len(self._waiters))
304	n/a	return '<{} [{}]>'.format(res[1:-1], extra)
305	n/a
306	n/a	@coroutine
307	n/a	def wait(self):
308	n/a	"""Wait until notified.
309	n/a
310	n/a	If the calling coroutine has not acquired the lock when this
311	n/a	method is called, a RuntimeError is raised.
312	n/a
313	n/a	This method releases the underlying lock, and then blocks
314	n/a	until it is awakened by a notify() or notify_all() call for
315	n/a	the same condition variable in another coroutine. Once
316	n/a	awakened, it re-acquires the lock and returns True.
317	n/a	"""
318	n/a	if not self.locked():
319	n/a	raise RuntimeError('cannot wait on un-acquired lock')
320	n/a
321	n/a	self.release()
322	n/a	try:
323	n/a	fut = self._loop.create_future()
324	n/a	self._waiters.append(fut)
325	n/a	try:
326	n/a	yield from fut
327	n/a	return True
328	n/a	finally:
329	n/a	self._waiters.remove(fut)
330	n/a
331	n/a	finally:
332	n/a	# Must reacquire lock even if wait is cancelled
333	n/a	while True:
334	n/a	try:
335	n/a	yield from self.acquire()
336	n/a	break
337	n/a	except futures.CancelledError:
338	n/a	pass
339	n/a
340	n/a	@coroutine
341	n/a	def wait_for(self, predicate):
342	n/a	"""Wait until a predicate becomes true.
343	n/a
344	n/a	The predicate should be a callable which result will be
345	n/a	interpreted as a boolean value. The final predicate value is
346	n/a	the return value.
347	n/a	"""
348	n/a	result = predicate()
349	n/a	while not result:
350	n/a	yield from self.wait()
351	n/a	result = predicate()
352	n/a	return result
353	n/a
354	n/a	def notify(self, n=1):
355	n/a	"""By default, wake up one coroutine waiting on this condition, if any.
356	n/a	If the calling coroutine has not acquired the lock when this method
357	n/a	is called, a RuntimeError is raised.
358	n/a
359	n/a	This method wakes up at most n of the coroutines waiting for the
360	n/a	condition variable; it is a no-op if no coroutines are waiting.
361	n/a
362	n/a	Note: an awakened coroutine does not actually return from its
363	n/a	wait() call until it can reacquire the lock. Since notify() does
364	n/a	not release the lock, its caller should.
365	n/a	"""
366	n/a	if not self.locked():
367	n/a	raise RuntimeError('cannot notify on un-acquired lock')
368	n/a
369	n/a	idx = 0
370	n/a	for fut in self._waiters:
371	n/a	if idx >= n:
372	n/a	break
373	n/a
374	n/a	if not fut.done():
375	n/a	idx += 1
376	n/a	fut.set_result(False)
377	n/a
378	n/a	def notify_all(self):
379	n/a	"""Wake up all threads waiting on this condition. This method acts
380	n/a	like notify(), but wakes up all waiting threads instead of one. If the
381	n/a	calling thread has not acquired the lock when this method is called,
382	n/a	a RuntimeError is raised.
383	n/a	"""
384	n/a	self.notify(len(self._waiters))
385	n/a
386	n/a
387	n/a	class Semaphore(_ContextManagerMixin):
388	n/a	"""A Semaphore implementation.
389	n/a
390	n/a	A semaphore manages an internal counter which is decremented by each
391	n/a	acquire() call and incremented by each release() call. The counter
392	n/a	can never go below zero; when acquire() finds that it is zero, it blocks,
393	n/a	waiting until some other thread calls release().
394	n/a
395	n/a	Semaphores also support the context management protocol.
396	n/a
397	n/a	The optional argument gives the initial value for the internal
398	n/a	counter; it defaults to 1. If the value given is less than 0,
399	n/a	ValueError is raised.
400	n/a	"""
401	n/a
402	n/a	def __init__(self, value=1, *, loop=None):
403	n/a	if value < 0:
404	n/a	raise ValueError("Semaphore initial value must be >= 0")
405	n/a	self._value = value
406	n/a	self._waiters = collections.deque()
407	n/a	if loop is not None:
408	n/a	self._loop = loop
409	n/a	else:
410	n/a	self._loop = events.get_event_loop()
411	n/a
412	n/a	def __repr__(self):
413	n/a	res = super().__repr__()
414	n/a	extra = 'locked' if self.locked() else 'unlocked,value:{}'.format(
415	n/a	self._value)
416	n/a	if self._waiters:
417	n/a	extra = '{},waiters:{}'.format(extra, len(self._waiters))
418	n/a	return '<{} [{}]>'.format(res[1:-1], extra)
419	n/a
420	n/a	def _wake_up_next(self):
421	n/a	while self._waiters:
422	n/a	waiter = self._waiters.popleft()
423	n/a	if not waiter.done():
424	n/a	waiter.set_result(None)
425	n/a	return
426	n/a
427	n/a	def locked(self):
428	n/a	"""Returns True if semaphore can not be acquired immediately."""
429	n/a	return self._value == 0
430	n/a
431	n/a	@coroutine
432	n/a	def acquire(self):
433	n/a	"""Acquire a semaphore.
434	n/a
435	n/a	If the internal counter is larger than zero on entry,
436	n/a	decrement it by one and return True immediately. If it is
437	n/a	zero on entry, block, waiting until some other coroutine has
438	n/a	called release() to make it larger than 0, and then return
439	n/a	True.
440	n/a	"""
441	n/a	while self._value <= 0:
442	n/a	fut = self._loop.create_future()
443	n/a	self._waiters.append(fut)
444	n/a	try:
445	n/a	yield from fut
446	n/a	except:
447	n/a	# See the similar code in Queue.get.
448	n/a	fut.cancel()
449	n/a	if self._value > 0 and not fut.cancelled():
450	n/a	self._wake_up_next()
451	n/a	raise
452	n/a	self._value -= 1
453	n/a	return True
454	n/a
455	n/a	def release(self):
456	n/a	"""Release a semaphore, incrementing the internal counter by one.
457	n/a	When it was zero on entry and another coroutine is waiting for it to
458	n/a	become larger than zero again, wake up that coroutine.
459	n/a	"""
460	n/a	self._value += 1
461	n/a	self._wake_up_next()
462	n/a
463	n/a
464	n/a	class BoundedSemaphore(Semaphore):
465	n/a	"""A bounded semaphore implementation.
466	n/a
467	n/a	This raises ValueError in release() if it would increase the value
468	n/a	above the initial value.
469	n/a	"""
470	n/a
471	n/a	def __init__(self, value=1, *, loop=None):
472	n/a	self._bound_value = value
473	n/a	super().__init__(value, loop=loop)
474	n/a
475	n/a	def release(self):
476	n/a	if self._value >= self._bound_value:
477	n/a	raise ValueError('BoundedSemaphore released too many times')
478	n/a	super().release()