2 # Module providing various facilities to other parts of the package
4 # multiprocessing/util.py
6 # Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
12 import threading # we want threading to install it's
13 # cleanup function before multiprocessing does
15 from multiprocessing.process import current_process, active_children
18 'sub_debug', 'debug', 'info', 'sub_warning', 'get_logger',
19 'log_to_stderr', 'get_temp_dir', 'register_after_fork',
20 'is_exiting', 'Finalize', 'ForkAwareThreadLock', 'ForkAwareLocal',
21 'SUBDEBUG', 'SUBWARNING',
34 LOGGER_NAME = 'multiprocessing'
35 DEFAULT_LOGGING_FORMAT = '[%(levelname)s/%(processName)s] %(message)s'
38 _log_to_stderr = False
40 def sub_debug(msg, *args):
42 _logger.log(SUBDEBUG, msg, *args)
44 def debug(msg, *args):
46 _logger.log(DEBUG, msg, *args)
50 _logger.log(INFO, msg, *args)
52 def sub_warning(msg, *args):
54 _logger.log(SUBWARNING, msg, *args)
58 Returns logger used by multiprocessing
61 import logging, atexit
63 logging._acquireLock()
67 _logger = logging.getLogger(LOGGER_NAME)
69 logging.addLevelName(SUBDEBUG, 'SUBDEBUG')
70 logging.addLevelName(SUBWARNING, 'SUBWARNING')
72 # XXX multiprocessing should cleanup before logging
73 if hasattr(atexit, 'unregister'):
74 atexit.unregister(_exit_function)
75 atexit.register(_exit_function)
77 atexit._exithandlers.remove((_exit_function, (), {}))
78 atexit._exithandlers.append((_exit_function, (), {}))
81 logging._releaseLock()
85 def log_to_stderr(level=None):
87 Turn on logging and add a handler which prints to stderr
93 formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
94 handler = logging.StreamHandler()
95 handler.setFormatter(formatter)
96 logger.addHandler(handler)
99 logger.setLevel(level)
100 _log_to_stderr = True
104 # Function returning a temp directory which will be removed on exit
108 # get name of a temp directory which will be automatically cleaned up
109 if current_process()._tempdir is None:
110 import shutil, tempfile
111 tempdir = tempfile.mkdtemp(prefix='pymp-')
112 info('created temp directory %s', tempdir)
113 Finalize(None, shutil.rmtree, args=[tempdir], exitpriority=-100)
114 current_process()._tempdir = tempdir
115 return current_process()._tempdir
118 # Support for reinitialization of objects when bootstrapping a child process
121 _afterfork_registry = weakref.WeakValueDictionary()
122 _afterfork_counter = itertools.count()
124 def _run_after_forkers():
125 items = list(_afterfork_registry.items())
127 for (index, ident, func), obj in items:
131 info('after forker raised exception %s', e)
133 def register_after_fork(obj, func):
134 _afterfork_registry[(_afterfork_counter.next(), id(obj), func)] = obj
137 # Finalization using weakrefs
140 _finalizer_registry = {}
141 _finalizer_counter = itertools.count()
144 class Finalize(object):
146 Class which supports object finalization using weakrefs
148 def __init__(self, obj, callback, args=(), kwargs=None, exitpriority=None):
149 assert exitpriority is None or type(exitpriority) is int
152 self._weakref = weakref.ref(obj, self)
154 assert exitpriority is not None
156 self._callback = callback
158 self._kwargs = kwargs or {}
159 self._key = (exitpriority, _finalizer_counter.next())
161 _finalizer_registry[self._key] = self
163 def __call__(self, wr=None):
165 Run the callback unless it has already been called or cancelled
168 del _finalizer_registry[self._key]
170 sub_debug('finalizer no longer registered')
172 sub_debug('finalizer calling %s with args %s and kwargs %s',
173 self._callback, self._args, self._kwargs)
174 res = self._callback(*self._args, **self._kwargs)
175 self._weakref = self._callback = self._args = \
176 self._kwargs = self._key = None
181 Cancel finalization of the object
184 del _finalizer_registry[self._key]
188 self._weakref = self._callback = self._args = \
189 self._kwargs = self._key = None
191 def still_active(self):
193 Return whether this finalizer is still waiting to invoke callback
195 return self._key in _finalizer_registry
199 obj = self._weakref()
200 except (AttributeError, TypeError):
204 return '<Finalize object, dead>'
206 x = '<Finalize object, callback=%s' % \
207 getattr(self._callback, '__name__', self._callback)
209 x += ', args=' + str(self._args)
211 x += ', kwargs=' + str(self._kwargs)
212 if self._key[0] is not None:
213 x += ', exitprority=' + str(self._key[0])
217 def _run_finalizers(minpriority=None):
219 Run all finalizers whose exit priority is not None and at least minpriority
221 Finalizers with highest priority are called first; finalizers with
222 the same priority will be called in reverse order of creation.
224 if minpriority is None:
225 f = lambda p : p[0][0] is not None
227 f = lambda p : p[0][0] is not None and p[0][0] >= minpriority
229 items = [x for x in _finalizer_registry.items() if f(x)]
230 items.sort(reverse=True)
232 for key, finalizer in items:
233 sub_debug('calling %s', finalizer)
238 traceback.print_exc()
240 if minpriority is None:
241 _finalizer_registry.clear()
249 Returns true if the process is shutting down
251 return _exiting or _exiting is None
255 def _exit_function():
258 info('process shutting down')
259 debug('running all "atexit" finalizers with priority >= 0')
262 for p in active_children():
264 info('calling terminate() for daemon %s', p.name)
267 for p in active_children():
268 info('calling join() for process %s', p.name)
271 debug('running the remaining "atexit" finalizers')
274 atexit.register(_exit_function)
277 # Some fork aware types
280 class ForkAwareThreadLock(object):
282 self._lock = threading.Lock()
283 self.acquire = self._lock.acquire
284 self.release = self._lock.release
285 register_after_fork(self, ForkAwareThreadLock.__init__)
287 class ForkAwareLocal(threading.local):
289 register_after_fork(self, lambda obj : obj.__dict__.clear())
290 def __reduce__(self):
291 return type(self), ()