Python code coverage for Lib/multiprocessing/synchronize.py

#	count	content
1	n/a	#
2	n/a	# Module implementing synchronization primitives
3	n/a	#
4	n/a	# multiprocessing/synchronize.py
5	n/a	#
6	n/a	# Copyright (c) 2006-2008, R Oudkerk
7	n/a	# Licensed to PSF under a Contributor Agreement.
8	n/a	#
9	n/a
10	n/a	__all__ = [
11	n/a	'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Condition', 'Event'
12	n/a	]
13	n/a
14	n/a	import threading
15	n/a	import sys
16	n/a	import tempfile
17	n/a	import _multiprocessing
18	n/a
19	n/a	from time import time as _time
20	n/a
21	n/a	from . import context
22	n/a	from . import process
23	n/a	from . import util
24	n/a
25	n/a	# Try to import the mp.synchronize module cleanly, if it fails
26	n/a	# raise ImportError for platforms lacking a working sem_open implementation.
27	n/a	# See issue 3770
28	n/a	try:
29	n/a	from _multiprocessing import SemLock, sem_unlink
30	n/a	except (ImportError):
31	n/a	raise ImportError("This platform lacks a functioning sem_open" +
32	n/a	" implementation, therefore, the required" +
33	n/a	" synchronization primitives needed will not" +
34	n/a	" function, see issue 3770.")
35	n/a
36	n/a	#
37	n/a	# Constants
38	n/a	#
39	n/a
40	n/a	RECURSIVE_MUTEX, SEMAPHORE = list(range(2))
41	n/a	SEM_VALUE_MAX = _multiprocessing.SemLock.SEM_VALUE_MAX
42	n/a
43	n/a	#
44	n/a	# Base class for semaphores and mutexes; wraps `_multiprocessing.SemLock`
45	n/a	#
46	n/a
47	n/a	class SemLock(object):
48	n/a
49	n/a	_rand = tempfile._RandomNameSequence()
50	n/a
51	n/a	def __init__(self, kind, value, maxvalue, *, ctx):
52	n/a	if ctx is None:
53	n/a	ctx = context._default_context.get_context()
54	n/a	name = ctx.get_start_method()
55	n/a	unlink_now = sys.platform == 'win32' or name == 'fork'
56	n/a	for i in range(100):
57	n/a	try:
58	n/a	sl = self._semlock = _multiprocessing.SemLock(
59	n/a	kind, value, maxvalue, self._make_name(),
60	n/a	unlink_now)
61	n/a	except FileExistsError:
62	n/a	pass
63	n/a	else:
64	n/a	break
65	n/a	else:
66	n/a	raise FileExistsError('cannot find name for semaphore')
67	n/a
68	n/a	util.debug('created semlock with handle %s' % sl.handle)
69	n/a	self._make_methods()
70	n/a
71	n/a	if sys.platform != 'win32':
72	n/a	def _after_fork(obj):
73	n/a	obj._semlock._after_fork()
74	n/a	util.register_after_fork(self, _after_fork)
75	n/a
76	n/a	if self._semlock.name is not None:
77	n/a	# We only get here if we are on Unix with forking
78	n/a	# disabled. When the object is garbage collected or the
79	n/a	# process shuts down we unlink the semaphore name
80	n/a	from .semaphore_tracker import register
81	n/a	register(self._semlock.name)
82	n/a	util.Finalize(self, SemLock._cleanup, (self._semlock.name,),
83	n/a	exitpriority=0)
84	n/a
85	n/a	@staticmethod
86	n/a	def _cleanup(name):
87	n/a	from .semaphore_tracker import unregister
88	n/a	sem_unlink(name)
89	n/a	unregister(name)
90	n/a
91	n/a	def _make_methods(self):
92	n/a	self.acquire = self._semlock.acquire
93	n/a	self.release = self._semlock.release
94	n/a
95	n/a	def __enter__(self):
96	n/a	return self._semlock.__enter__()
97	n/a
98	n/a	def __exit__(self, *args):
99	n/a	return self._semlock.__exit__(*args)
100	n/a
101	n/a	def __getstate__(self):
102	n/a	context.assert_spawning(self)
103	n/a	sl = self._semlock
104	n/a	if sys.platform == 'win32':
105	n/a	h = context.get_spawning_popen().duplicate_for_child(sl.handle)
106	n/a	else:
107	n/a	h = sl.handle
108	n/a	return (h, sl.kind, sl.maxvalue, sl.name)
109	n/a
110	n/a	def __setstate__(self, state):
111	n/a	self._semlock = _multiprocessing.SemLock._rebuild(*state)
112	n/a	util.debug('recreated blocker with handle %r' % state[0])
113	n/a	self._make_methods()
114	n/a
115	n/a	@staticmethod
116	n/a	def _make_name():
117	n/a	return '%s-%s' % (process.current_process()._config['semprefix'],
118	n/a	next(SemLock._rand))
119	n/a
120	n/a	#
121	n/a	# Semaphore
122	n/a	#
123	n/a
124	n/a	class Semaphore(SemLock):
125	n/a
126	n/a	def __init__(self, value=1, *, ctx):
127	n/a	SemLock.__init__(self, SEMAPHORE, value, SEM_VALUE_MAX, ctx=ctx)
128	n/a
129	n/a	def get_value(self):
130	n/a	return self._semlock._get_value()
131	n/a
132	n/a	def __repr__(self):
133	n/a	try:
134	n/a	value = self._semlock._get_value()
135	n/a	except Exception:
136	n/a	value = 'unknown'
137	n/a	return '<%s(value=%s)>' % (self.__class__.__name__, value)
138	n/a
139	n/a	#
140	n/a	# Bounded semaphore
141	n/a	#
142	n/a
143	n/a	class BoundedSemaphore(Semaphore):
144	n/a
145	n/a	def __init__(self, value=1, *, ctx):
146	n/a	SemLock.__init__(self, SEMAPHORE, value, value, ctx=ctx)
147	n/a
148	n/a	def __repr__(self):
149	n/a	try:
150	n/a	value = self._semlock._get_value()
151	n/a	except Exception:
152	n/a	value = 'unknown'
153	n/a	return '<%s(value=%s, maxvalue=%s)>' % \
154	n/a	(self.__class__.__name__, value, self._semlock.maxvalue)
155	n/a
156	n/a	#
157	n/a	# Non-recursive lock
158	n/a	#
159	n/a
160	n/a	class Lock(SemLock):
161	n/a
162	n/a	def __init__(self, *, ctx):
163	n/a	SemLock.__init__(self, SEMAPHORE, 1, 1, ctx=ctx)
164	n/a
165	n/a	def __repr__(self):
166	n/a	try:
167	n/a	if self._semlock._is_mine():
168	n/a	name = process.current_process().name
169	n/a	if threading.current_thread().name != 'MainThread':
170	n/a	name += '\|' + threading.current_thread().name
171	n/a	elif self._semlock._get_value() == 1:
172	n/a	name = 'None'
173	n/a	elif self._semlock._count() > 0:
174	n/a	name = 'SomeOtherThread'
175	n/a	else:
176	n/a	name = 'SomeOtherProcess'
177	n/a	except Exception:
178	n/a	name = 'unknown'
179	n/a	return '<%s(owner=%s)>' % (self.__class__.__name__, name)
180	n/a
181	n/a	#
182	n/a	# Recursive lock
183	n/a	#
184	n/a
185	n/a	class RLock(SemLock):
186	n/a
187	n/a	def __init__(self, *, ctx):
188	n/a	SemLock.__init__(self, RECURSIVE_MUTEX, 1, 1, ctx=ctx)
189	n/a
190	n/a	def __repr__(self):
191	n/a	try:
192	n/a	if self._semlock._is_mine():
193	n/a	name = process.current_process().name
194	n/a	if threading.current_thread().name != 'MainThread':
195	n/a	name += '\|' + threading.current_thread().name
196	n/a	count = self._semlock._count()
197	n/a	elif self._semlock._get_value() == 1:
198	n/a	name, count = 'None', 0
199	n/a	elif self._semlock._count() > 0:
200	n/a	name, count = 'SomeOtherThread', 'nonzero'
201	n/a	else:
202	n/a	name, count = 'SomeOtherProcess', 'nonzero'
203	n/a	except Exception:
204	n/a	name, count = 'unknown', 'unknown'
205	n/a	return '<%s(%s, %s)>' % (self.__class__.__name__, name, count)
206	n/a
207	n/a	#
208	n/a	# Condition variable
209	n/a	#
210	n/a
211	n/a	class Condition(object):
212	n/a
213	n/a	def __init__(self, lock=None, *, ctx):
214	n/a	self._lock = lock or ctx.RLock()
215	n/a	self._sleeping_count = ctx.Semaphore(0)
216	n/a	self._woken_count = ctx.Semaphore(0)
217	n/a	self._wait_semaphore = ctx.Semaphore(0)
218	n/a	self._make_methods()
219	n/a
220	n/a	def __getstate__(self):
221	n/a	context.assert_spawning(self)
222	n/a	return (self._lock, self._sleeping_count,
223	n/a	self._woken_count, self._wait_semaphore)
224	n/a
225	n/a	def __setstate__(self, state):
226	n/a	(self._lock, self._sleeping_count,
227	n/a	self._woken_count, self._wait_semaphore) = state
228	n/a	self._make_methods()
229	n/a
230	n/a	def __enter__(self):
231	n/a	return self._lock.__enter__()
232	n/a
233	n/a	def __exit__(self, *args):
234	n/a	return self._lock.__exit__(*args)
235	n/a
236	n/a	def _make_methods(self):
237	n/a	self.acquire = self._lock.acquire
238	n/a	self.release = self._lock.release
239	n/a
240	n/a	def __repr__(self):
241	n/a	try:
242	n/a	num_waiters = (self._sleeping_count._semlock._get_value() -
243	n/a	self._woken_count._semlock._get_value())
244	n/a	except Exception:
245	n/a	num_waiters = 'unknown'
246	n/a	return '<%s(%s, %s)>' % (self.__class__.__name__, self._lock, num_waiters)
247	n/a
248	n/a	def wait(self, timeout=None):
249	n/a	assert self._lock._semlock._is_mine(), \
250	n/a	'must acquire() condition before using wait()'
251	n/a
252	n/a	# indicate that this thread is going to sleep
253	n/a	self._sleeping_count.release()
254	n/a
255	n/a	# release lock
256	n/a	count = self._lock._semlock._count()
257	n/a	for i in range(count):
258	n/a	self._lock.release()
259	n/a
260	n/a	try:
261	n/a	# wait for notification or timeout
262	n/a	return self._wait_semaphore.acquire(True, timeout)
263	n/a	finally:
264	n/a	# indicate that this thread has woken
265	n/a	self._woken_count.release()
266	n/a
267	n/a	# reacquire lock
268	n/a	for i in range(count):
269	n/a	self._lock.acquire()
270	n/a
271	n/a	def notify(self):
272	n/a	assert self._lock._semlock._is_mine(), 'lock is not owned'
273	n/a	assert not self._wait_semaphore.acquire(False)
274	n/a
275	n/a	# to take account of timeouts since last notify() we subtract
276	n/a	# woken_count from sleeping_count and rezero woken_count
277	n/a	while self._woken_count.acquire(False):
278	n/a	res = self._sleeping_count.acquire(False)
279	n/a	assert res
280	n/a
281	n/a	if self._sleeping_count.acquire(False): # try grabbing a sleeper
282	n/a	self._wait_semaphore.release() # wake up one sleeper
283	n/a	self._woken_count.acquire() # wait for the sleeper to wake
284	n/a
285	n/a	# rezero _wait_semaphore in case a timeout just happened
286	n/a	self._wait_semaphore.acquire(False)
287	n/a
288	n/a	def notify_all(self):
289	n/a	assert self._lock._semlock._is_mine(), 'lock is not owned'
290	n/a	assert not self._wait_semaphore.acquire(False)
291	n/a
292	n/a	# to take account of timeouts since last notify*() we subtract
293	n/a	# woken_count from sleeping_count and rezero woken_count
294	n/a	while self._woken_count.acquire(False):
295	n/a	res = self._sleeping_count.acquire(False)
296	n/a	assert res
297	n/a
298	n/a	sleepers = 0
299	n/a	while self._sleeping_count.acquire(False):
300	n/a	self._wait_semaphore.release() # wake up one sleeper
301	n/a	sleepers += 1
302	n/a
303	n/a	if sleepers:
304	n/a	for i in range(sleepers):
305	n/a	self._woken_count.acquire() # wait for a sleeper to wake
306	n/a
307	n/a	# rezero wait_semaphore in case some timeouts just happened
308	n/a	while self._wait_semaphore.acquire(False):
309	n/a	pass
310	n/a
311	n/a	def wait_for(self, predicate, timeout=None):
312	n/a	result = predicate()
313	n/a	if result:
314	n/a	return result
315	n/a	if timeout is not None:
316	n/a	endtime = _time() + timeout
317	n/a	else:
318	n/a	endtime = None
319	n/a	waittime = None
320	n/a	while not result:
321	n/a	if endtime is not None:
322	n/a	waittime = endtime - _time()
323	n/a	if waittime <= 0:
324	n/a	break
325	n/a	self.wait(waittime)
326	n/a	result = predicate()
327	n/a	return result
328	n/a
329	n/a	#
330	n/a	# Event
331	n/a	#
332	n/a
333	n/a	class Event(object):
334	n/a
335	n/a	def __init__(self, *, ctx):
336	n/a	self._cond = ctx.Condition(ctx.Lock())
337	n/a	self._flag = ctx.Semaphore(0)
338	n/a
339	n/a	def is_set(self):
340	n/a	with self._cond:
341	n/a	if self._flag.acquire(False):
342	n/a	self._flag.release()
343	n/a	return True
344	n/a	return False
345	n/a
346	n/a	def set(self):
347	n/a	with self._cond:
348	n/a	self._flag.acquire(False)
349	n/a	self._flag.release()
350	n/a	self._cond.notify_all()
351	n/a
352	n/a	def clear(self):
353	n/a	with self._cond:
354	n/a	self._flag.acquire(False)
355	n/a
356	n/a	def wait(self, timeout=None):
357	n/a	with self._cond:
358	n/a	if self._flag.acquire(False):
359	n/a	self._flag.release()
360	n/a	else:
361	n/a	self._cond.wait(timeout)
362	n/a
363	n/a	if self._flag.acquire(False):
364	n/a	self._flag.release()
365	n/a	return True
366	n/a	return False
367	n/a
368	n/a	#
369	n/a	# Barrier
370	n/a	#
371	n/a
372	n/a	class Barrier(threading.Barrier):
373	n/a
374	n/a	def __init__(self, parties, action=None, timeout=None, *, ctx):
375	n/a	import struct
376	n/a	from .heap import BufferWrapper
377	n/a	wrapper = BufferWrapper(struct.calcsize('i') * 2)
378	n/a	cond = ctx.Condition()
379	n/a	self.__setstate__((parties, action, timeout, cond, wrapper))
380	n/a	self._state = 0
381	n/a	self._count = 0
382	n/a
383	n/a	def __setstate__(self, state):
384	n/a	(self._parties, self._action, self._timeout,
385	n/a	self._cond, self._wrapper) = state
386	n/a	self._array = self._wrapper.create_memoryview().cast('i')
387	n/a
388	n/a	def __getstate__(self):
389	n/a	return (self._parties, self._action, self._timeout,
390	n/a	self._cond, self._wrapper)
391	n/a
392	n/a	@property
393	n/a	def _state(self):
394	n/a	return self._array[0]
395	n/a
396	n/a	@_state.setter
397	n/a	def _state(self, value):
398	n/a	self._array[0] = value
399	n/a
400	n/a	@property
401	n/a	def _count(self):
402	n/a	return self._array[1]
403	n/a
404	n/a	@_count.setter
405	n/a	def _count(self, value):
406	n/a	self._array[1] = value