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 sharedmemory 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 use semunlink to clean up named semaphores semunlink may be missing if the python build process detected the absence of posix named semaphores in that case no named semaphores were ever opened so no cleanup would be necessary gh109629 this happens if an explicit call to the resourcetracker gets interrupted by a garbage collection invoking a finalizer that itself calls back into resourcetracker for example the semlock finalizer this should not happen stop isn t called by a finalizer but we check for it anyway not running closing the alive file descriptor stops main make sure that resource tracker process is running this can be run from any process usually a child process will use the resource its parent with self lock if self lock recursioncount 1 the code below is certainly not reentrantsafe so bail out return self reentrantcallerror if self fd is not none resource tracker was launched before is it still running if self checkalive still alive return dead launch it again os closeself fd cleanup to avoid dangling processes try pid can be none if this process is a child from another python process which has started the resourcetracker if self pid is not none os waitpidself pid 0 except childprocesserror the resourcetracker has already been terminated pass self fd none self pid none warnings warn resourcetracker process died unexpectedly relaunching some resources might leak fdstopass try fdstopass appendsys stderr fileno except exception pass cmd from multiprocessing resourcetracker import main maind r w os pipe try fdstopass appendr process will out live us so no need to wait on pid exe spawn getexecutable args exe util argsfrominterpreterflags args c cmd r bpo33613 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 havesigmask signal pthreadsigmasksignal sigblock ignoredsignals pid util spawnvpassfdsexe args fdstopass finally if havesigmask signal pthreadsigmasksignal sigunblock ignoredsignals except os closew raise else self fd w self pid pid finally os closer def checkaliveself we cannot use send here as it calls ensurerunning creating a cycle register name of resource with resource tracker self send register name rtype def unregisterself name rtype the code below might or might not work depending on whether the resource tracker was already running and still alive better warn the user xxx is warnings warn itself reentrantsafe posix guarantees that writes to a pipe of less than pipebuf bytes are atomic and that pipebuf 512 run resource tracker protect the process from c and killall python etc signal signalsignal sigint signal sigign signal signalsignal sigterm signal sigign if havesigmask signal pthreadsigmasksignal sigunblock ignoredsignals for f in sys stdin sys stdout try f close except exception pass cache rtype set for rtype in cleanupfuncs keys try keep track of registeredunregistered resources with openfd rb as f for line in f try cmd name rtype line strip decode ascii split cleanupfunc cleanupfuncs getrtype none if cleanupfunc is none raise valueerror f cannot register name for automatic cleanup f unknown resource type rtype if cmd register cachertype addname elif cmd unregister cachertype removename elif cmd probe pass else raise runtimeerror unrecognized command r cmd except exception try sys excepthooksys excinfo except pass finally all processes have terminated cleanup any remaining resources for rtype rtypecache in cache items if rtypecache try warnings warn f resourcetracker there appear to be lenrtypecache f leaked rtype objects to clean up at shutdown rtypecache except exception pass for name in rtypecache 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 cleanupfuncsrtypename except exception as e warnings warn resourcetracker r s name e finally pass 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 use sem_unlink to clean up named semaphores sem_unlink may be missing if the python build process detected the absence of posix named semaphores in that case no named semaphores were ever opened so no cleanup would be necessary gh 109629 this happens if an explicit call to the resourcetracker gets interrupted by a garbage collection invoking a finalizer that itself calls back into resourcetracker for example the semlock finalizer this should not happen _stop isn t called by a finalizer but we check for it anyway not running closing the alive file descriptor stops main make sure that resource tracker process is running this can be run from any process usually a child process will use the resource its parent the code below is certainly not reentrant safe so bail out resource tracker was launched before is it still running still alive dead launch it again clean up to avoid dangling processes _pid can be none if this process is a child from another python process which has started the resource_tracker the resource_tracker has already been terminated process will out live us so no need to wait on pid 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 check that the pipe has not been closed by sending a probe we cannot use send here as it calls ensure_running creating a cycle register name of resource with resource tracker unregister name of resource with resource tracker the code below might or might not work depending on whether the resource tracker was already running and still alive better warn the user xxx is warnings warn itself reentrant safe posix guarantees that writes to a pipe of less than pipe_buf bytes are atomic and that pipe_buf 512 run resource tracker protect the process from c and killall python etc keep track of registered unregistered resources all processes have terminated cleanup any remaining resources for some reason the process which created and registered this resource has failed to unregister it presumably it has died we therefore unlink it
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

    
    
    
    
    
    if hasattr(_multiprocessing, 'sem_unlink'):
        _CLEANUP_FUNCS.update({
            'semaphore': _multiprocessing.sem_unlink,
        })
    _CLEANUP_FUNCS.update({
        'shared_memory': _posixshmem.shm_unlink,
    })


class ReentrantCallError(RuntimeError):
    pass


class ResourceTracker(object):

    def __init__(self):
        self._lock = threading.RLock()
        self._fd = None
        self._pid = None

    def _reentrant_call_error(self):
        
        
        
        
        raise ReentrantCallError(
            "Reentrant call into the multiprocessing resource tracker")

    def _stop(self):
        with self._lock:
            
            
            if self._lock._recursion_count() > 1:
                return self._reentrant_call_error()
            if self._fd is None:
                
                return

            
            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):
        
        with self._lock:
            if self._lock._recursion_count() > 1:
                
                return self._reentrant_call_error()
            if self._fd is not None:
                
                if self._check_alive():
                    
                    return
                
                os.close(self._fd)

                
                try:
                    
                    
                    if self._pid is not None:
                        os.waitpid(self._pid, 0)
                except ChildProcessError:
                    
                    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)
                
                exe = spawn.get_executable()
                args = [exe] + util._args_from_interpreter_flags()
                args += ['-c', cmd % r]
                
                
                
                
                
                
                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):
        
        try:
            
            
            os.write(self._fd, b'PROBE:0:noop\n')
        except OSError:
            return False
        else:
            return True

    def register(self, name, rtype):
        
        self._send('REGISTER', name, rtype)

    def unregister(self, name, rtype):
        
        self._send('UNREGISTER', name, rtype)

    def _send(self, cmd, name, rtype):
        try:
            self.ensure_running()
        except ReentrantCallError:
            
            
            
            
            warnings.warn(
                f"ResourceTracker called reentrantly for resource cleanup, "
                f"which is unsupported. "
                f"The {rtype} object {name!r} might leak.")
        msg = '{0}:{1}:{2}\n'.format(cmd, name, rtype).encode('ascii')
        if len(msg) > 512:
            
            
            raise ValueError('msg 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):
    
    
    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:
        
        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:
        
        for rtype, rtype_cache in cache.items():
            if rtype_cache:
                try:
                    warnings.warn(
                        f'resource_tracker: There appear to be {len(rtype_cache)} '
                        f'leaked {rtype} objects to clean up at shutdown: {rtype_cache}'
                    )
                except Exception:
                    pass
            for name in rtype_cache:
                
                
                
                try:
                    try:
                        _CLEANUP_FUNCS[rtype](name)
                    except Exception as e:
                        warnings.warn('resource_tracker: %r: %s' % (name, e))
                finally:
                    pass