Python code coverage for Lib/multiprocessing/util.py

#	count	content
1	n/a	#
2	n/a	# Module providing various facilities to other parts of the package
3	n/a	#
4	n/a	# multiprocessing/util.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	import os
11	n/a	import itertools
12	n/a	import sys
13	n/a	import weakref
14	n/a	import atexit
15	n/a	import threading # we want threading to install it's
16	n/a	# cleanup function before multiprocessing does
17	n/a	from subprocess import _args_from_interpreter_flags
18	n/a
19	n/a	from . import process
20	n/a
21	n/a	__all__ = [
22	n/a	'sub_debug', 'debug', 'info', 'sub_warning', 'get_logger',
23	n/a	'log_to_stderr', 'get_temp_dir', 'register_after_fork',
24	n/a	'is_exiting', 'Finalize', 'ForkAwareThreadLock', 'ForkAwareLocal',
25	n/a	'close_all_fds_except', 'SUBDEBUG', 'SUBWARNING',
26	n/a	]
27	n/a
28	n/a	#
29	n/a	# Logging
30	n/a	#
31	n/a
32	n/a	NOTSET = 0
33	n/a	SUBDEBUG = 5
34	n/a	DEBUG = 10
35	n/a	INFO = 20
36	n/a	SUBWARNING = 25
37	n/a
38	n/a	LOGGER_NAME = 'multiprocessing'
39	n/a	DEFAULT_LOGGING_FORMAT = '[%(levelname)s/%(processName)s] %(message)s'
40	n/a
41	n/a	_logger = None
42	n/a	_log_to_stderr = False
43	n/a
44	n/a	def sub_debug(msg, *args):
45	n/a	if _logger:
46	n/a	_logger.log(SUBDEBUG, msg, *args)
47	n/a
48	n/a	def debug(msg, *args):
49	n/a	if _logger:
50	n/a	_logger.log(DEBUG, msg, *args)
51	n/a
52	n/a	def info(msg, *args):
53	n/a	if _logger:
54	n/a	_logger.log(INFO, msg, *args)
55	n/a
56	n/a	def sub_warning(msg, *args):
57	n/a	if _logger:
58	n/a	_logger.log(SUBWARNING, msg, *args)
59	n/a
60	n/a	def get_logger():
61	n/a	'''
62	n/a	Returns logger used by multiprocessing
63	n/a	'''
64	n/a	global _logger
65	n/a	import logging
66	n/a
67	n/a	logging._acquireLock()
68	n/a	try:
69	n/a	if not _logger:
70	n/a
71	n/a	_logger = logging.getLogger(LOGGER_NAME)
72	n/a	_logger.propagate = 0
73	n/a
74	n/a	# XXX multiprocessing should cleanup before logging
75	n/a	if hasattr(atexit, 'unregister'):
76	n/a	atexit.unregister(_exit_function)
77	n/a	atexit.register(_exit_function)
78	n/a	else:
79	n/a	atexit._exithandlers.remove((_exit_function, (), {}))
80	n/a	atexit._exithandlers.append((_exit_function, (), {}))
81	n/a
82	n/a	finally:
83	n/a	logging._releaseLock()
84	n/a
85	n/a	return _logger
86	n/a
87	n/a	def log_to_stderr(level=None):
88	n/a	'''
89	n/a	Turn on logging and add a handler which prints to stderr
90	n/a	'''
91	n/a	global _log_to_stderr
92	n/a	import logging
93	n/a
94	n/a	logger = get_logger()
95	n/a	formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
96	n/a	handler = logging.StreamHandler()
97	n/a	handler.setFormatter(formatter)
98	n/a	logger.addHandler(handler)
99	n/a
100	n/a	if level:
101	n/a	logger.setLevel(level)
102	n/a	_log_to_stderr = True
103	n/a	return _logger
104	n/a
105	n/a	#
106	n/a	# Function returning a temp directory which will be removed on exit
107	n/a	#
108	n/a
109	n/a	def get_temp_dir():
110	n/a	# get name of a temp directory which will be automatically cleaned up
111	n/a	tempdir = process.current_process()._config.get('tempdir')
112	n/a	if tempdir is None:
113	n/a	import shutil, tempfile
114	n/a	tempdir = tempfile.mkdtemp(prefix='pymp-')
115	n/a	info('created temp directory %s', tempdir)
116	n/a	Finalize(None, shutil.rmtree, args=[tempdir], exitpriority=-100)
117	n/a	process.current_process()._config['tempdir'] = tempdir
118	n/a	return tempdir
119	n/a
120	n/a	#
121	n/a	# Support for reinitialization of objects when bootstrapping a child process
122	n/a	#
123	n/a
124	n/a	_afterfork_registry = weakref.WeakValueDictionary()
125	n/a	_afterfork_counter = itertools.count()
126	n/a
127	n/a	def _run_after_forkers():
128	n/a	items = list(_afterfork_registry.items())
129	n/a	items.sort()
130	n/a	for (index, ident, func), obj in items:
131	n/a	try:
132	n/a	func(obj)
133	n/a	except Exception as e:
134	n/a	info('after forker raised exception %s', e)
135	n/a
136	n/a	def register_after_fork(obj, func):
137	n/a	_afterfork_registry[(next(_afterfork_counter), id(obj), func)] = obj
138	n/a
139	n/a	#
140	n/a	# Finalization using weakrefs
141	n/a	#
142	n/a
143	n/a	_finalizer_registry = {}
144	n/a	_finalizer_counter = itertools.count()
145	n/a
146	n/a
147	n/a	class Finalize(object):
148	n/a	'''
149	n/a	Class which supports object finalization using weakrefs
150	n/a	'''
151	n/a	def __init__(self, obj, callback, args=(), kwargs=None, exitpriority=None):
152	n/a	assert exitpriority is None or type(exitpriority) is int
153	n/a
154	n/a	if obj is not None:
155	n/a	self._weakref = weakref.ref(obj, self)
156	n/a	else:
157	n/a	assert exitpriority is not None
158	n/a
159	n/a	self._callback = callback
160	n/a	self._args = args
161	n/a	self._kwargs = kwargs or {}
162	n/a	self._key = (exitpriority, next(_finalizer_counter))
163	n/a	self._pid = os.getpid()
164	n/a
165	n/a	_finalizer_registry[self._key] = self
166	n/a
167	n/a	def __call__(self, wr=None,
168	n/a	# Need to bind these locally because the globals can have
169	n/a	# been cleared at shutdown
170	n/a	_finalizer_registry=_finalizer_registry,
171	n/a	sub_debug=sub_debug, getpid=os.getpid):
172	n/a	'''
173	n/a	Run the callback unless it has already been called or cancelled
174	n/a	'''
175	n/a	try:
176	n/a	del _finalizer_registry[self._key]
177	n/a	except KeyError:
178	n/a	sub_debug('finalizer no longer registered')
179	n/a	else:
180	n/a	if self._pid != getpid():
181	n/a	sub_debug('finalizer ignored because different process')
182	n/a	res = None
183	n/a	else:
184	n/a	sub_debug('finalizer calling %s with args %s and kwargs %s',
185	n/a	self._callback, self._args, self._kwargs)
186	n/a	res = self._callback(self._args, *self._kwargs)
187	n/a	self._weakref = self._callback = self._args = \
188	n/a	self._kwargs = self._key = None
189	n/a	return res
190	n/a
191	n/a	def cancel(self):
192	n/a	'''
193	n/a	Cancel finalization of the object
194	n/a	'''
195	n/a	try:
196	n/a	del _finalizer_registry[self._key]
197	n/a	except KeyError:
198	n/a	pass
199	n/a	else:
200	n/a	self._weakref = self._callback = self._args = \
201	n/a	self._kwargs = self._key = None
202	n/a
203	n/a	def still_active(self):
204	n/a	'''
205	n/a	Return whether this finalizer is still waiting to invoke callback
206	n/a	'''
207	n/a	return self._key in _finalizer_registry
208	n/a
209	n/a	def __repr__(self):
210	n/a	try:
211	n/a	obj = self._weakref()
212	n/a	except (AttributeError, TypeError):
213	n/a	obj = None
214	n/a
215	n/a	if obj is None:
216	n/a	return '<%s object, dead>' % self.__class__.__name__
217	n/a
218	n/a	x = '<%s object, callback=%s' % (
219	n/a	self.__class__.__name__,
220	n/a	getattr(self._callback, '__name__', self._callback))
221	n/a	if self._args:
222	n/a	x += ', args=' + str(self._args)
223	n/a	if self._kwargs:
224	n/a	x += ', kwargs=' + str(self._kwargs)
225	n/a	if self._key[0] is not None:
226	n/a	x += ', exitprority=' + str(self._key[0])
227	n/a	return x + '>'
228	n/a
229	n/a
230	n/a	def _run_finalizers(minpriority=None):
231	n/a	'''
232	n/a	Run all finalizers whose exit priority is not None and at least minpriority
233	n/a
234	n/a	Finalizers with highest priority are called first; finalizers with
235	n/a	the same priority will be called in reverse order of creation.
236	n/a	'''
237	n/a	if _finalizer_registry is None:
238	n/a	# This function may be called after this module's globals are
239	n/a	# destroyed. See the _exit_function function in this module for more
240	n/a	# notes.
241	n/a	return
242	n/a
243	n/a	if minpriority is None:
244	n/a	f = lambda p : p[0][0] is not None
245	n/a	else:
246	n/a	f = lambda p : p[0][0] is not None and p[0][0] >= minpriority
247	n/a
248	n/a	items = [x for x in list(_finalizer_registry.items()) if f(x)]
249	n/a	items.sort(reverse=True)
250	n/a
251	n/a	for key, finalizer in items:
252	n/a	sub_debug('calling %s', finalizer)
253	n/a	try:
254	n/a	finalizer()
255	n/a	except Exception:
256	n/a	import traceback
257	n/a	traceback.print_exc()
258	n/a
259	n/a	if minpriority is None:
260	n/a	_finalizer_registry.clear()
261	n/a
262	n/a	#
263	n/a	# Clean up on exit
264	n/a	#
265	n/a
266	n/a	def is_exiting():
267	n/a	'''
268	n/a	Returns true if the process is shutting down
269	n/a	'''
270	n/a	return _exiting or _exiting is None
271	n/a
272	n/a	_exiting = False
273	n/a
274	n/a	def _exit_function(info=info, debug=debug, _run_finalizers=_run_finalizers,
275	n/a	active_children=process.active_children,
276	n/a	current_process=process.current_process):
277	n/a	# We hold on to references to functions in the arglist due to the
278	n/a	# situation described below, where this function is called after this
279	n/a	# module's globals are destroyed.
280	n/a
281	n/a	global _exiting
282	n/a
283	n/a	if not _exiting:
284	n/a	_exiting = True
285	n/a
286	n/a	info('process shutting down')
287	n/a	debug('running all "atexit" finalizers with priority >= 0')
288	n/a	_run_finalizers(0)
289	n/a
290	n/a	if current_process() is not None:
291	n/a	# We check if the current process is None here because if
292	n/a	# it's None, any call to ``active_children()`` will raise
293	n/a	# an AttributeError (active_children winds up trying to
294	n/a	# get attributes from util._current_process). One
295	n/a	# situation where this can happen is if someone has
296	n/a	# manipulated sys.modules, causing this module to be
297	n/a	# garbage collected. The destructor for the module type
298	n/a	# then replaces all values in the module dict with None.
299	n/a	# For instance, after setuptools runs a test it replaces
300	n/a	# sys.modules with a copy created earlier. See issues
301	n/a	# #9775 and #15881. Also related: #4106, #9205, and
302	n/a	# #9207.
303	n/a
304	n/a	for p in active_children():
305	n/a	if p.daemon:
306	n/a	info('calling terminate() for daemon %s', p.name)
307	n/a	p._popen.terminate()
308	n/a
309	n/a	for p in active_children():
310	n/a	info('calling join() for process %s', p.name)
311	n/a	p.join()
312	n/a
313	n/a	debug('running the remaining "atexit" finalizers')
314	n/a	_run_finalizers()
315	n/a
316	n/a	atexit.register(_exit_function)
317	n/a
318	n/a	#
319	n/a	# Some fork aware types
320	n/a	#
321	n/a
322	n/a	class ForkAwareThreadLock(object):
323	n/a	def __init__(self):
324	n/a	self._reset()
325	n/a	register_after_fork(self, ForkAwareThreadLock._reset)
326	n/a
327	n/a	def _reset(self):
328	n/a	self._lock = threading.Lock()
329	n/a	self.acquire = self._lock.acquire
330	n/a	self.release = self._lock.release
331	n/a
332	n/a	def __enter__(self):
333	n/a	return self._lock.__enter__()
334	n/a
335	n/a	def __exit__(self, *args):
336	n/a	return self._lock.__exit__(*args)
337	n/a
338	n/a
339	n/a	class ForkAwareLocal(threading.local):
340	n/a	def __init__(self):
341	n/a	register_after_fork(self, lambda obj : obj.__dict__.clear())
342	n/a	def __reduce__(self):
343	n/a	return type(self), ()
344	n/a
345	n/a	#
346	n/a	# Close fds except those specified
347	n/a	#
348	n/a
349	n/a	try:
350	n/a	MAXFD = os.sysconf("SC_OPEN_MAX")
351	n/a	except Exception:
352	n/a	MAXFD = 256
353	n/a
354	n/a	def close_all_fds_except(fds):
355	n/a	fds = list(fds) + [-1, MAXFD]
356	n/a	fds.sort()
357	n/a	assert fds[-1] == MAXFD, 'fd too large'
358	n/a	for i in range(len(fds) - 1):
359	n/a	os.closerange(fds[i]+1, fds[i+1])
360	n/a	#
361	n/a	# Close sys.stdin and replace stdin with os.devnull
362	n/a	#
363	n/a
364	n/a	def _close_stdin():
365	n/a	if sys.stdin is None:
366	n/a	return
367	n/a
368	n/a	try:
369	n/a	sys.stdin.close()
370	n/a	except (OSError, ValueError):
371	n/a	pass
372	n/a
373	n/a	try:
374	n/a	fd = os.open(os.devnull, os.O_RDONLY)
375	n/a	try:
376	n/a	sys.stdin = open(fd, closefd=False)
377	n/a	except:
378	n/a	os.close(fd)
379	n/a	raise
380	n/a	except (OSError, ValueError):
381	n/a	pass
382	n/a
383	n/a	#
384	n/a	# Start a program with only specified fds kept open
385	n/a	#
386	n/a
387	n/a	def spawnv_passfds(path, args, passfds):
388	n/a	import _posixsubprocess
389	n/a	passfds = sorted(passfds)
390	n/a	errpipe_read, errpipe_write = os.pipe()
391	n/a	try:
392	n/a	return _posixsubprocess.fork_exec(
393	n/a	args, [os.fsencode(path)], True, passfds, None, None,
394	n/a	-1, -1, -1, -1, -1, -1, errpipe_read, errpipe_write,
395	n/a	False, False, None)
396	n/a	finally:
397	n/a	os.close(errpipe_read)
398	n/a	os.close(errpipe_write)