Python code coverage for Lib/multiprocessing/semaphore_tracker.py

#	count	content
1	n/a	#
2	n/a	# On Unix we run a server process which keeps track of unlinked
3	n/a	# semaphores. The server ignores SIGINT and SIGTERM and reads from a
4	n/a	# pipe. Every other process of the program has a copy of the writable
5	n/a	# end of the pipe, so we get EOF when all other processes have exited.
6	n/a	# Then the server process unlinks any remaining semaphore names.
7	n/a	#
8	n/a	# This is important because the system only supports a limited number
9	n/a	# of named semaphores, and they will not be automatically removed till
10	n/a	# the next reboot. Without this semaphore tracker process, "killall
11	n/a	# python" would probably leave unlinked semaphores.
12	n/a	#
13	n/a
14	n/a	import os
15	n/a	import signal
16	n/a	import sys
17	n/a	import threading
18	n/a	import warnings
19	n/a	import _multiprocessing
20	n/a
21	n/a	from . import spawn
22	n/a	from . import util
23	n/a
24	n/a	__all__ = ['ensure_running', 'register', 'unregister']
25	n/a
26	n/a
27	n/a	class SemaphoreTracker(object):
28	n/a
29	n/a	def __init__(self):
30	n/a	self._lock = threading.Lock()
31	n/a	self._fd = None
32	n/a
33	n/a	def getfd(self):
34	n/a	self.ensure_running()
35	n/a	return self._fd
36	n/a
37	n/a	def ensure_running(self):
38	n/a	'''Make sure that semaphore tracker process is running.
39	n/a
40	n/a	This can be run from any process. Usually a child process will use
41	n/a	the semaphore created by its parent.'''
42	n/a	with self._lock:
43	n/a	if self._fd is not None:
44	n/a	return
45	n/a	fds_to_pass = []
46	n/a	try:
47	n/a	fds_to_pass.append(sys.stderr.fileno())
48	n/a	except Exception:
49	n/a	pass
50	n/a	cmd = 'from multiprocessing.semaphore_tracker import main;main(%d)'
51	n/a	r, w = os.pipe()
52	n/a	try:
53	n/a	fds_to_pass.append(r)
54	n/a	# process will out live us, so no need to wait on pid
55	n/a	exe = spawn.get_executable()
56	n/a	args = [exe] + util._args_from_interpreter_flags()
57	n/a	args += ['-c', cmd % r]
58	n/a	util.spawnv_passfds(exe, args, fds_to_pass)
59	n/a	except:
60	n/a	os.close(w)
61	n/a	raise
62	n/a	else:
63	n/a	self._fd = w
64	n/a	finally:
65	n/a	os.close(r)
66	n/a
67	n/a	def register(self, name):
68	n/a	'''Register name of semaphore with semaphore tracker.'''
69	n/a	self._send('REGISTER', name)
70	n/a
71	n/a	def unregister(self, name):
72	n/a	'''Unregister name of semaphore with semaphore tracker.'''
73	n/a	self._send('UNREGISTER', name)
74	n/a
75	n/a	def _send(self, cmd, name):
76	n/a	self.ensure_running()
77	n/a	msg = '{0}:{1}\n'.format(cmd, name).encode('ascii')
78	n/a	if len(name) > 512:
79	n/a	# posix guarantees that writes to a pipe of less than PIPE_BUF
80	n/a	# bytes are atomic, and that PIPE_BUF >= 512
81	n/a	raise ValueError('name too long')
82	n/a	nbytes = os.write(self._fd, msg)
83	n/a	assert nbytes == len(msg)
84	n/a
85	n/a
86	n/a	_semaphore_tracker = SemaphoreTracker()
87	n/a	ensure_running = _semaphore_tracker.ensure_running
88	n/a	register = _semaphore_tracker.register
89	n/a	unregister = _semaphore_tracker.unregister
90	n/a	getfd = _semaphore_tracker.getfd
91	n/a
92	n/a
93	n/a	def main(fd):
94	n/a	'''Run semaphore tracker.'''
95	n/a	# protect the process from ^C and "killall python" etc
96	n/a	signal.signal(signal.SIGINT, signal.SIG_IGN)
97	n/a	signal.signal(signal.SIGTERM, signal.SIG_IGN)
98	n/a
99	n/a	for f in (sys.stdin, sys.stdout):
100	n/a	try:
101	n/a	f.close()
102	n/a	except Exception:
103	n/a	pass
104	n/a
105	n/a	cache = set()
106	n/a	try:
107	n/a	# keep track of registered/unregistered semaphores
108	n/a	with open(fd, 'rb') as f:
109	n/a	for line in f:
110	n/a	try:
111	n/a	cmd, name = line.strip().split(b':')
112	n/a	if cmd == b'REGISTER':
113	n/a	cache.add(name)
114	n/a	elif cmd == b'UNREGISTER':
115	n/a	cache.remove(name)
116	n/a	else:
117	n/a	raise RuntimeError('unrecognized command %r' % cmd)
118	n/a	except Exception:
119	n/a	try:
120	n/a	sys.excepthook(*sys.exc_info())
121	n/a	except:
122	n/a	pass
123	n/a	finally:
124	n/a	# all processes have terminated; cleanup any remaining semaphores
125	n/a	if cache:
126	n/a	try:
127	n/a	warnings.warn('semaphore_tracker: There appear to be %d '
128	n/a	'leaked semaphores to clean up at shutdown' %
129	n/a	len(cache))
130	n/a	except Exception:
131	n/a	pass
132	n/a	for name in cache:
133	n/a	# For some reason the process which created and registered this
134	n/a	# semaphore has failed to unregister it. Presumably it has died.
135	n/a	# We therefore unlink it.
136	n/a	try:
137	n/a	name = name.decode('ascii')
138	n/a	try:
139	n/a	_multiprocessing.sem_unlink(name)
140	n/a	except Exception as e:
141	n/a	warnings.warn('semaphore_tracker: %r: %s' % (name, e))
142	n/a	finally:
143	n/a	pass