mirror of
https://github.com/python/cpython.git
synced 2025-10-31 05:31:20 +00:00
9707e8e22d
Multiprocessing and concurrent.futures tests now stop the resource tracker process when tests complete. Add ResourceTracker._stop() method to multiprocessing.resource_tracker. Add _cleanup_tests() helper function to multiprocessing.util: share code between multiprocessing and concurrent.futures tests.
231 lines
8.4 KiB
Python
231 lines
8.4 KiB
Python
###############################################################################
|
|
# Server process to keep track of unlinked resources (like shared memory
|
|
# segments, semaphores etc.) and clean them.
|
|
#
|
|
# On Unix we run a server process which keeps track of unlinked
|
|
# resources. The server ignores SIGINT and SIGTERM and reads from a
|
|
# pipe. Every other process of the program has a copy of the writable
|
|
# end of the pipe, so we get EOF when all other processes have exited.
|
|
# Then the server process unlinks any remaining resource names.
|
|
#
|
|
# This is important because there may be system limits for such resources: for
|
|
# instance, the system only supports a limited number of named semaphores, and
|
|
# shared-memory segments live in the RAM. If a python process leaks such a
|
|
# resource, this resource will not be removed till the next reboot. Without
|
|
# this resource tracker process, "killall python" would probably leave unlinked
|
|
# resources.
|
|
|
|
import os
|
|
import signal
|
|
import sys
|
|
import threading
|
|
import warnings
|
|
|
|
from . import spawn
|
|
from . import util
|
|
|
|
__all__ = ['ensure_running', 'register', 'unregister']
|
|
|
|
_HAVE_SIGMASK = hasattr(signal, 'pthread_sigmask')
|
|
_IGNORED_SIGNALS = (signal.SIGINT, signal.SIGTERM)
|
|
|
|
_CLEANUP_FUNCS = {
|
|
'noop': lambda: None,
|
|
}
|
|
|
|
if os.name == 'posix':
|
|
import _multiprocessing
|
|
import _posixshmem
|
|
|
|
_CLEANUP_FUNCS.update({
|
|
'semaphore': _multiprocessing.sem_unlink,
|
|
'shared_memory': _posixshmem.shm_unlink,
|
|
})
|
|
|
|
|
|
class ResourceTracker(object):
|
|
|
|
def __init__(self):
|
|
self._lock = threading.Lock()
|
|
self._fd = None
|
|
self._pid = None
|
|
|
|
def _stop(self):
|
|
with self._lock:
|
|
if self._fd is None:
|
|
# not running
|
|
return
|
|
|
|
# closing the "alive" file descriptor stops main()
|
|
os.close(self._fd)
|
|
self._fd = None
|
|
|
|
os.waitpid(self._pid, 0)
|
|
self._pid = None
|
|
|
|
def getfd(self):
|
|
self.ensure_running()
|
|
return self._fd
|
|
|
|
def ensure_running(self):
|
|
'''Make sure that resource tracker process is running.
|
|
|
|
This can be run from any process. Usually a child process will use
|
|
the resource created by its parent.'''
|
|
with self._lock:
|
|
if self._fd is not None:
|
|
# resource tracker was launched before, is it still running?
|
|
if self._check_alive():
|
|
# => still alive
|
|
return
|
|
# => dead, launch it again
|
|
os.close(self._fd)
|
|
|
|
# Clean-up to avoid dangling processes.
|
|
try:
|
|
# _pid can be None if this process is a child from another
|
|
# python process, which has started the resource_tracker.
|
|
if self._pid is not None:
|
|
os.waitpid(self._pid, 0)
|
|
except ChildProcessError:
|
|
# The resource_tracker has already been terminated.
|
|
pass
|
|
self._fd = None
|
|
self._pid = None
|
|
|
|
warnings.warn('resource_tracker: process died unexpectedly, '
|
|
'relaunching. Some resources might leak.')
|
|
|
|
fds_to_pass = []
|
|
try:
|
|
fds_to_pass.append(sys.stderr.fileno())
|
|
except Exception:
|
|
pass
|
|
cmd = 'from multiprocessing.resource_tracker import main;main(%d)'
|
|
r, w = os.pipe()
|
|
try:
|
|
fds_to_pass.append(r)
|
|
# process will out live us, so no need to wait on pid
|
|
exe = spawn.get_executable()
|
|
args = [exe] + util._args_from_interpreter_flags()
|
|
args += ['-c', cmd % r]
|
|
# bpo-33613: Register a signal mask that will block the signals.
|
|
# This signal mask will be inherited by the child that is going
|
|
# to be spawned and will protect the child from a race condition
|
|
# that can make the child die before it registers signal handlers
|
|
# for SIGINT and SIGTERM. The mask is unregistered after spawning
|
|
# the child.
|
|
try:
|
|
if _HAVE_SIGMASK:
|
|
signal.pthread_sigmask(signal.SIG_BLOCK, _IGNORED_SIGNALS)
|
|
pid = util.spawnv_passfds(exe, args, fds_to_pass)
|
|
finally:
|
|
if _HAVE_SIGMASK:
|
|
signal.pthread_sigmask(signal.SIG_UNBLOCK, _IGNORED_SIGNALS)
|
|
except:
|
|
os.close(w)
|
|
raise
|
|
else:
|
|
self._fd = w
|
|
self._pid = pid
|
|
finally:
|
|
os.close(r)
|
|
|
|
def _check_alive(self):
|
|
'''Check that the pipe has not been closed by sending a probe.'''
|
|
try:
|
|
# We cannot use send here as it calls ensure_running, creating
|
|
# a cycle.
|
|
os.write(self._fd, b'PROBE:0:noop\n')
|
|
except OSError:
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
def register(self, name, rtype):
|
|
'''Register name of resource with resource tracker.'''
|
|
self._send('REGISTER', name, rtype)
|
|
|
|
def unregister(self, name, rtype):
|
|
'''Unregister name of resource with resource tracker.'''
|
|
self._send('UNREGISTER', name, rtype)
|
|
|
|
def _send(self, cmd, name, rtype):
|
|
self.ensure_running()
|
|
msg = '{0}:{1}:{2}\n'.format(cmd, name, rtype).encode('ascii')
|
|
if len(name) > 512:
|
|
# posix guarantees that writes to a pipe of less than PIPE_BUF
|
|
# bytes are atomic, and that PIPE_BUF >= 512
|
|
raise ValueError('name too long')
|
|
nbytes = os.write(self._fd, msg)
|
|
assert nbytes == len(msg), "nbytes {0:n} but len(msg) {1:n}".format(
|
|
nbytes, len(msg))
|
|
|
|
|
|
_resource_tracker = ResourceTracker()
|
|
ensure_running = _resource_tracker.ensure_running
|
|
register = _resource_tracker.register
|
|
unregister = _resource_tracker.unregister
|
|
getfd = _resource_tracker.getfd
|
|
|
|
def main(fd):
|
|
'''Run resource tracker.'''
|
|
# protect the process from ^C and "killall python" etc
|
|
signal.signal(signal.SIGINT, signal.SIG_IGN)
|
|
signal.signal(signal.SIGTERM, signal.SIG_IGN)
|
|
if _HAVE_SIGMASK:
|
|
signal.pthread_sigmask(signal.SIG_UNBLOCK, _IGNORED_SIGNALS)
|
|
|
|
for f in (sys.stdin, sys.stdout):
|
|
try:
|
|
f.close()
|
|
except Exception:
|
|
pass
|
|
|
|
cache = {rtype: set() for rtype in _CLEANUP_FUNCS.keys()}
|
|
try:
|
|
# keep track of registered/unregistered resources
|
|
with open(fd, 'rb') as f:
|
|
for line in f:
|
|
try:
|
|
cmd, name, rtype = line.strip().decode('ascii').split(':')
|
|
cleanup_func = _CLEANUP_FUNCS.get(rtype, None)
|
|
if cleanup_func is None:
|
|
raise ValueError(
|
|
f'Cannot register {name} for automatic cleanup: '
|
|
f'unknown resource type {rtype}')
|
|
|
|
if cmd == 'REGISTER':
|
|
cache[rtype].add(name)
|
|
elif cmd == 'UNREGISTER':
|
|
cache[rtype].remove(name)
|
|
elif cmd == 'PROBE':
|
|
pass
|
|
else:
|
|
raise RuntimeError('unrecognized command %r' % cmd)
|
|
except Exception:
|
|
try:
|
|
sys.excepthook(*sys.exc_info())
|
|
except:
|
|
pass
|
|
finally:
|
|
# all processes have terminated; cleanup any remaining resources
|
|
for rtype, rtype_cache in cache.items():
|
|
if rtype_cache:
|
|
try:
|
|
warnings.warn('resource_tracker: There appear to be %d '
|
|
'leaked %s objects to clean up at shutdown' %
|
|
(len(rtype_cache), rtype))
|
|
except Exception:
|
|
pass
|
|
for name in rtype_cache:
|
|
# For some reason the process which created and registered this
|
|
# resource has failed to unregister it. Presumably it has
|
|
# died. We therefore unlink it.
|
|
try:
|
|
try:
|
|
_CLEANUP_FUNCS[rtype](name)
|
|
except Exception as e:
|
|
warnings.warn('resource_tracker: %r: %s' % (name, e))
|
|
finally:
|
|
pass
|