jpayne@68: #
jpayne@68: # Module providing the `Process` class which emulates `threading.Thread`
jpayne@68: #
jpayne@68: # multiprocessing/process.py
jpayne@68: #
jpayne@68: # Copyright (c) 2006-2008, R Oudkerk
jpayne@68: # Licensed to PSF under a Contributor Agreement.
jpayne@68: #
jpayne@68: 
jpayne@68: __all__ = ['BaseProcess', 'current_process', 'active_children',
jpayne@68:            'parent_process']
jpayne@68: 
jpayne@68: #
jpayne@68: # Imports
jpayne@68: #
jpayne@68: 
jpayne@68: import os
jpayne@68: import sys
jpayne@68: import signal
jpayne@68: import itertools
jpayne@68: import threading
jpayne@68: from _weakrefset import WeakSet
jpayne@68: 
jpayne@68: #
jpayne@68: #
jpayne@68: #
jpayne@68: 
jpayne@68: try:
jpayne@68:     ORIGINAL_DIR = os.path.abspath(os.getcwd())
jpayne@68: except OSError:
jpayne@68:     ORIGINAL_DIR = None
jpayne@68: 
jpayne@68: #
jpayne@68: # Public functions
jpayne@68: #
jpayne@68: 
jpayne@68: def current_process():
jpayne@68:     '''
jpayne@68:     Return process object representing the current process
jpayne@68:     '''
jpayne@68:     return _current_process
jpayne@68: 
jpayne@68: def active_children():
jpayne@68:     '''
jpayne@68:     Return list of process objects corresponding to live child processes
jpayne@68:     '''
jpayne@68:     _cleanup()
jpayne@68:     return list(_children)
jpayne@68: 
jpayne@68: 
jpayne@68: def parent_process():
jpayne@68:     '''
jpayne@68:     Return process object representing the parent process
jpayne@68:     '''
jpayne@68:     return _parent_process
jpayne@68: 
jpayne@68: #
jpayne@68: #
jpayne@68: #
jpayne@68: 
jpayne@68: def _cleanup():
jpayne@68:     # check for processes which have finished
jpayne@68:     for p in list(_children):
jpayne@68:         if p._popen.poll() is not None:
jpayne@68:             _children.discard(p)
jpayne@68: 
jpayne@68: #
jpayne@68: # The `Process` class
jpayne@68: #
jpayne@68: 
jpayne@68: class BaseProcess(object):
jpayne@68:     '''
jpayne@68:     Process objects represent activity that is run in a separate process
jpayne@68: 
jpayne@68:     The class is analogous to `threading.Thread`
jpayne@68:     '''
jpayne@68:     def _Popen(self):
jpayne@68:         raise NotImplementedError
jpayne@68: 
jpayne@68:     def __init__(self, group=None, target=None, name=None, args=(), kwargs={},
jpayne@68:                  *, daemon=None):
jpayne@68:         assert group is None, 'group argument must be None for now'
jpayne@68:         count = next(_process_counter)
jpayne@68:         self._identity = _current_process._identity + (count,)
jpayne@68:         self._config = _current_process._config.copy()
jpayne@68:         self._parent_pid = os.getpid()
jpayne@68:         self._parent_name = _current_process.name
jpayne@68:         self._popen = None
jpayne@68:         self._closed = False
jpayne@68:         self._target = target
jpayne@68:         self._args = tuple(args)
jpayne@68:         self._kwargs = dict(kwargs)
jpayne@68:         self._name = name or type(self).__name__ + '-' + \
jpayne@68:                      ':'.join(str(i) for i in self._identity)
jpayne@68:         if daemon is not None:
jpayne@68:             self.daemon = daemon
jpayne@68:         _dangling.add(self)
jpayne@68: 
jpayne@68:     def _check_closed(self):
jpayne@68:         if self._closed:
jpayne@68:             raise ValueError("process object is closed")
jpayne@68: 
jpayne@68:     def run(self):
jpayne@68:         '''
jpayne@68:         Method to be run in sub-process; can be overridden in sub-class
jpayne@68:         '''
jpayne@68:         if self._target:
jpayne@68:             self._target(*self._args, **self._kwargs)
jpayne@68: 
jpayne@68:     def start(self):
jpayne@68:         '''
jpayne@68:         Start child process
jpayne@68:         '''
jpayne@68:         self._check_closed()
jpayne@68:         assert self._popen is None, 'cannot start a process twice'
jpayne@68:         assert self._parent_pid == os.getpid(), \
jpayne@68:                'can only start a process object created by current process'
jpayne@68:         assert not _current_process._config.get('daemon'), \
jpayne@68:                'daemonic processes are not allowed to have children'
jpayne@68:         _cleanup()
jpayne@68:         self._popen = self._Popen(self)
jpayne@68:         self._sentinel = self._popen.sentinel
jpayne@68:         # Avoid a refcycle if the target function holds an indirect
jpayne@68:         # reference to the process object (see bpo-30775)
jpayne@68:         del self._target, self._args, self._kwargs
jpayne@68:         _children.add(self)
jpayne@68: 
jpayne@68:     def terminate(self):
jpayne@68:         '''
jpayne@68:         Terminate process; sends SIGTERM signal or uses TerminateProcess()
jpayne@68:         '''
jpayne@68:         self._check_closed()
jpayne@68:         self._popen.terminate()
jpayne@68: 
jpayne@68:     def kill(self):
jpayne@68:         '''
jpayne@68:         Terminate process; sends SIGKILL signal or uses TerminateProcess()
jpayne@68:         '''
jpayne@68:         self._check_closed()
jpayne@68:         self._popen.kill()
jpayne@68: 
jpayne@68:     def join(self, timeout=None):
jpayne@68:         '''
jpayne@68:         Wait until child process terminates
jpayne@68:         '''
jpayne@68:         self._check_closed()
jpayne@68:         assert self._parent_pid == os.getpid(), 'can only join a child process'
jpayne@68:         assert self._popen is not None, 'can only join a started process'
jpayne@68:         res = self._popen.wait(timeout)
jpayne@68:         if res is not None:
jpayne@68:             _children.discard(self)
jpayne@68: 
jpayne@68:     def is_alive(self):
jpayne@68:         '''
jpayne@68:         Return whether process is alive
jpayne@68:         '''
jpayne@68:         self._check_closed()
jpayne@68:         if self is _current_process:
jpayne@68:             return True
jpayne@68:         assert self._parent_pid == os.getpid(), 'can only test a child process'
jpayne@68: 
jpayne@68:         if self._popen is None:
jpayne@68:             return False
jpayne@68: 
jpayne@68:         returncode = self._popen.poll()
jpayne@68:         if returncode is None:
jpayne@68:             return True
jpayne@68:         else:
jpayne@68:             _children.discard(self)
jpayne@68:             return False
jpayne@68: 
jpayne@68:     def close(self):
jpayne@68:         '''
jpayne@68:         Close the Process object.
jpayne@68: 
jpayne@68:         This method releases resources held by the Process object.  It is
jpayne@68:         an error to call this method if the child process is still running.
jpayne@68:         '''
jpayne@68:         if self._popen is not None:
jpayne@68:             if self._popen.poll() is None:
jpayne@68:                 raise ValueError("Cannot close a process while it is still running. "
jpayne@68:                                  "You should first call join() or terminate().")
jpayne@68:             self._popen.close()
jpayne@68:             self._popen = None
jpayne@68:             del self._sentinel
jpayne@68:             _children.discard(self)
jpayne@68:         self._closed = True
jpayne@68: 
jpayne@68:     @property
jpayne@68:     def name(self):
jpayne@68:         return self._name
jpayne@68: 
jpayne@68:     @name.setter
jpayne@68:     def name(self, name):
jpayne@68:         assert isinstance(name, str), 'name must be a string'
jpayne@68:         self._name = name
jpayne@68: 
jpayne@68:     @property
jpayne@68:     def daemon(self):
jpayne@68:         '''
jpayne@68:         Return whether process is a daemon
jpayne@68:         '''
jpayne@68:         return self._config.get('daemon', False)
jpayne@68: 
jpayne@68:     @daemon.setter
jpayne@68:     def daemon(self, daemonic):
jpayne@68:         '''
jpayne@68:         Set whether process is a daemon
jpayne@68:         '''
jpayne@68:         assert self._popen is None, 'process has already started'
jpayne@68:         self._config['daemon'] = daemonic
jpayne@68: 
jpayne@68:     @property
jpayne@68:     def authkey(self):
jpayne@68:         return self._config['authkey']
jpayne@68: 
jpayne@68:     @authkey.setter
jpayne@68:     def authkey(self, authkey):
jpayne@68:         '''
jpayne@68:         Set authorization key of process
jpayne@68:         '''
jpayne@68:         self._config['authkey'] = AuthenticationString(authkey)
jpayne@68: 
jpayne@68:     @property
jpayne@68:     def exitcode(self):
jpayne@68:         '''
jpayne@68:         Return exit code of process or `None` if it has yet to stop
jpayne@68:         '''
jpayne@68:         self._check_closed()
jpayne@68:         if self._popen is None:
jpayne@68:             return self._popen
jpayne@68:         return self._popen.poll()
jpayne@68: 
jpayne@68:     @property
jpayne@68:     def ident(self):
jpayne@68:         '''
jpayne@68:         Return identifier (PID) of process or `None` if it has yet to start
jpayne@68:         '''
jpayne@68:         self._check_closed()
jpayne@68:         if self is _current_process:
jpayne@68:             return os.getpid()
jpayne@68:         else:
jpayne@68:             return self._popen and self._popen.pid
jpayne@68: 
jpayne@68:     pid = ident
jpayne@68: 
jpayne@68:     @property
jpayne@68:     def sentinel(self):
jpayne@68:         '''
jpayne@68:         Return a file descriptor (Unix) or handle (Windows) suitable for
jpayne@68:         waiting for process termination.
jpayne@68:         '''
jpayne@68:         self._check_closed()
jpayne@68:         try:
jpayne@68:             return self._sentinel
jpayne@68:         except AttributeError:
jpayne@68:             raise ValueError("process not started") from None
jpayne@68: 
jpayne@68:     def __repr__(self):
jpayne@68:         exitcode = None
jpayne@68:         if self is _current_process:
jpayne@68:             status = 'started'
jpayne@68:         elif self._closed:
jpayne@68:             status = 'closed'
jpayne@68:         elif self._parent_pid != os.getpid():
jpayne@68:             status = 'unknown'
jpayne@68:         elif self._popen is None:
jpayne@68:             status = 'initial'
jpayne@68:         else:
jpayne@68:             exitcode = self._popen.poll()
jpayne@68:             if exitcode is not None:
jpayne@68:                 status = 'stopped'
jpayne@68:             else:
jpayne@68:                 status = 'started'
jpayne@68: 
jpayne@68:         info = [type(self).__name__, 'name=%r' % self._name]
jpayne@68:         if self._popen is not None:
jpayne@68:             info.append('pid=%s' % self._popen.pid)
jpayne@68:         info.append('parent=%s' % self._parent_pid)
jpayne@68:         info.append(status)
jpayne@68:         if exitcode is not None:
jpayne@68:             exitcode = _exitcode_to_name.get(exitcode, exitcode)
jpayne@68:             info.append('exitcode=%s' % exitcode)
jpayne@68:         if self.daemon:
jpayne@68:             info.append('daemon')
jpayne@68:         return '<%s>' % ' '.join(info)
jpayne@68: 
jpayne@68:     ##
jpayne@68: 
jpayne@68:     def _bootstrap(self, parent_sentinel=None):
jpayne@68:         from . import util, context
jpayne@68:         global _current_process, _parent_process, _process_counter, _children
jpayne@68: 
jpayne@68:         try:
jpayne@68:             if self._start_method is not None:
jpayne@68:                 context._force_start_method(self._start_method)
jpayne@68:             _process_counter = itertools.count(1)
jpayne@68:             _children = set()
jpayne@68:             util._close_stdin()
jpayne@68:             old_process = _current_process
jpayne@68:             _current_process = self
jpayne@68:             _parent_process = _ParentProcess(
jpayne@68:                 self._parent_name, self._parent_pid, parent_sentinel)
jpayne@68:             if threading._HAVE_THREAD_NATIVE_ID:
jpayne@68:                 threading.main_thread()._set_native_id()
jpayne@68:             try:
jpayne@68:                 util._finalizer_registry.clear()
jpayne@68:                 util._run_after_forkers()
jpayne@68:             finally:
jpayne@68:                 # delay finalization of the old process object until after
jpayne@68:                 # _run_after_forkers() is executed
jpayne@68:                 del old_process
jpayne@68:             util.info('child process calling self.run()')
jpayne@68:             try:
jpayne@68:                 self.run()
jpayne@68:                 exitcode = 0
jpayne@68:             finally:
jpayne@68:                 util._exit_function()
jpayne@68:         except SystemExit as e:
jpayne@68:             if not e.args:
jpayne@68:                 exitcode = 1
jpayne@68:             elif isinstance(e.args[0], int):
jpayne@68:                 exitcode = e.args[0]
jpayne@68:             else:
jpayne@68:                 sys.stderr.write(str(e.args[0]) + '\n')
jpayne@68:                 exitcode = 1
jpayne@68:         except:
jpayne@68:             exitcode = 1
jpayne@68:             import traceback
jpayne@68:             sys.stderr.write('Process %s:\n' % self.name)
jpayne@68:             traceback.print_exc()
jpayne@68:         finally:
jpayne@68:             threading._shutdown()
jpayne@68:             util.info('process exiting with exitcode %d' % exitcode)
jpayne@68:             util._flush_std_streams()
jpayne@68: 
jpayne@68:         return exitcode
jpayne@68: 
jpayne@68: #
jpayne@68: # We subclass bytes to avoid accidental transmission of auth keys over network
jpayne@68: #
jpayne@68: 
jpayne@68: class AuthenticationString(bytes):
jpayne@68:     def __reduce__(self):
jpayne@68:         from .context import get_spawning_popen
jpayne@68:         if get_spawning_popen() is None:
jpayne@68:             raise TypeError(
jpayne@68:                 'Pickling an AuthenticationString object is '
jpayne@68:                 'disallowed for security reasons'
jpayne@68:                 )
jpayne@68:         return AuthenticationString, (bytes(self),)
jpayne@68: 
jpayne@68: 
jpayne@68: #
jpayne@68: # Create object representing the parent process
jpayne@68: #
jpayne@68: 
jpayne@68: class _ParentProcess(BaseProcess):
jpayne@68: 
jpayne@68:     def __init__(self, name, pid, sentinel):
jpayne@68:         self._identity = ()
jpayne@68:         self._name = name
jpayne@68:         self._pid = pid
jpayne@68:         self._parent_pid = None
jpayne@68:         self._popen = None
jpayne@68:         self._closed = False
jpayne@68:         self._sentinel = sentinel
jpayne@68:         self._config = {}
jpayne@68: 
jpayne@68:     def is_alive(self):
jpayne@68:         from multiprocessing.connection import wait
jpayne@68:         return not wait([self._sentinel], timeout=0)
jpayne@68: 
jpayne@68:     @property
jpayne@68:     def ident(self):
jpayne@68:         return self._pid
jpayne@68: 
jpayne@68:     def join(self, timeout=None):
jpayne@68:         '''
jpayne@68:         Wait until parent process terminates
jpayne@68:         '''
jpayne@68:         from multiprocessing.connection import wait
jpayne@68:         wait([self._sentinel], timeout=timeout)
jpayne@68: 
jpayne@68:     pid = ident
jpayne@68: 
jpayne@68: #
jpayne@68: # Create object representing the main process
jpayne@68: #
jpayne@68: 
jpayne@68: class _MainProcess(BaseProcess):
jpayne@68: 
jpayne@68:     def __init__(self):
jpayne@68:         self._identity = ()
jpayne@68:         self._name = 'MainProcess'
jpayne@68:         self._parent_pid = None
jpayne@68:         self._popen = None
jpayne@68:         self._closed = False
jpayne@68:         self._config = {'authkey': AuthenticationString(os.urandom(32)),
jpayne@68:                         'semprefix': '/mp'}
jpayne@68:         # Note that some versions of FreeBSD only allow named
jpayne@68:         # semaphores to have names of up to 14 characters.  Therefore
jpayne@68:         # we choose a short prefix.
jpayne@68:         #
jpayne@68:         # On MacOSX in a sandbox it may be necessary to use a
jpayne@68:         # different prefix -- see #19478.
jpayne@68:         #
jpayne@68:         # Everything in self._config will be inherited by descendant
jpayne@68:         # processes.
jpayne@68: 
jpayne@68:     def close(self):
jpayne@68:         pass
jpayne@68: 
jpayne@68: 
jpayne@68: _parent_process = None
jpayne@68: _current_process = _MainProcess()
jpayne@68: _process_counter = itertools.count(1)
jpayne@68: _children = set()
jpayne@68: del _MainProcess
jpayne@68: 
jpayne@68: #
jpayne@68: # Give names to some return codes
jpayne@68: #
jpayne@68: 
jpayne@68: _exitcode_to_name = {}
jpayne@68: 
jpayne@68: for name, signum in list(signal.__dict__.items()):
jpayne@68:     if name[:3]=='SIG' and '_' not in name:
jpayne@68:         _exitcode_to_name[-signum] = f'-{name}'
jpayne@68: 
jpayne@68: # For debug and leak testing
jpayne@68: _dangling = WeakSet()