Mercurial > repos > rliterman > csp2

annotate CSP2/CSP2_env/env-d9b9114564458d9d-741b3de822f2aaca6c6caa4325c4afce/lib/python3.8/multiprocessing/process.py @ 68:5028fdace37b

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
planemo upload commit 2e9511a184a1ca667c7be0c6321a36dc4e3d116d
author jpayne
date Tue, 18 Mar 2025 16:23:26 -0400
parents
children
rev   line source
jpayne@68 1 #
jpayne@68 2 # Module providing the `Process` class which emulates `threading.Thread`
jpayne@68 3 #
jpayne@68 4 # multiprocessing/process.py
jpayne@68 5 #
jpayne@68 6 # Copyright (c) 2006-2008, R Oudkerk
jpayne@68 7 # Licensed to PSF under a Contributor Agreement.
jpayne@68 8 #
jpayne@68 9
jpayne@68 10 __all__ = ['BaseProcess', 'current_process', 'active_children',
jpayne@68 11 'parent_process']
jpayne@68 12
jpayne@68 13 #
jpayne@68 14 # Imports
jpayne@68 15 #
jpayne@68 16
jpayne@68 17 import os
jpayne@68 18 import sys
jpayne@68 19 import signal
jpayne@68 20 import itertools
jpayne@68 21 import threading
jpayne@68 22 from _weakrefset import WeakSet
jpayne@68 23
jpayne@68 24 #
jpayne@68 25 #
jpayne@68 26 #
jpayne@68 27
jpayne@68 28 try:
jpayne@68 29 ORIGINAL_DIR = os.path.abspath(os.getcwd())
jpayne@68 30 except OSError:
jpayne@68 31 ORIGINAL_DIR = None
jpayne@68 32
jpayne@68 33 #
jpayne@68 34 # Public functions
jpayne@68 35 #
jpayne@68 36
jpayne@68 37 def current_process():
jpayne@68 38 '''
jpayne@68 39 Return process object representing the current process
jpayne@68 40 '''
jpayne@68 41 return _current_process
jpayne@68 42
jpayne@68 43 def active_children():
jpayne@68 44 '''
jpayne@68 45 Return list of process objects corresponding to live child processes
jpayne@68 46 '''
jpayne@68 47 _cleanup()
jpayne@68 48 return list(_children)
jpayne@68 49
jpayne@68 50
jpayne@68 51 def parent_process():
jpayne@68 52 '''
jpayne@68 53 Return process object representing the parent process
jpayne@68 54 '''
jpayne@68 55 return _parent_process
jpayne@68 56
jpayne@68 57 #
jpayne@68 58 #
jpayne@68 59 #
jpayne@68 60
jpayne@68 61 def _cleanup():
jpayne@68 62 # check for processes which have finished
jpayne@68 63 for p in list(_children):
jpayne@68 64 if p._popen.poll() is not None:
jpayne@68 65 _children.discard(p)
jpayne@68 66
jpayne@68 67 #
jpayne@68 68 # The `Process` class
jpayne@68 69 #
jpayne@68 70
jpayne@68 71 class BaseProcess(object):
jpayne@68 72 '''
jpayne@68 73 Process objects represent activity that is run in a separate process
jpayne@68 74
jpayne@68 75 The class is analogous to `threading.Thread`
jpayne@68 76 '''
jpayne@68 77 def _Popen(self):
jpayne@68 78 raise NotImplementedError
jpayne@68 79
jpayne@68 80 def __init__(self, group=None, target=None, name=None, args=(), kwargs={},
jpayne@68 81 *, daemon=None):
jpayne@68 82 assert group is None, 'group argument must be None for now'
jpayne@68 83 count = next(_process_counter)
jpayne@68 84 self._identity = _current_process._identity + (count,)
jpayne@68 85 self._config = _current_process._config.copy()
jpayne@68 86 self._parent_pid = os.getpid()
jpayne@68 87 self._parent_name = _current_process.name
jpayne@68 88 self._popen = None
jpayne@68 89 self._closed = False
jpayne@68 90 self._target = target
jpayne@68 91 self._args = tuple(args)
jpayne@68 92 self._kwargs = dict(kwargs)
jpayne@68 93 self._name = name or type(self).__name__ + '-' + \
jpayne@68 94 ':'.join(str(i) for i in self._identity)
jpayne@68 95 if daemon is not None:
jpayne@68 96 self.daemon = daemon
jpayne@68 97 _dangling.add(self)
jpayne@68 98
jpayne@68 99 def _check_closed(self):
jpayne@68 100 if self._closed:
jpayne@68 101 raise ValueError("process object is closed")
jpayne@68 102
jpayne@68 103 def run(self):
jpayne@68 104 '''
jpayne@68 105 Method to be run in sub-process; can be overridden in sub-class
jpayne@68 106 '''
jpayne@68 107 if self._target:
jpayne@68 108 self._target(*self._args, **self._kwargs)
jpayne@68 109
jpayne@68 110 def start(self):
jpayne@68 111 '''
jpayne@68 112 Start child process
jpayne@68 113 '''
jpayne@68 114 self._check_closed()
jpayne@68 115 assert self._popen is None, 'cannot start a process twice'
jpayne@68 116 assert self._parent_pid == os.getpid(), \
jpayne@68 117 'can only start a process object created by current process'
jpayne@68 118 assert not _current_process._config.get('daemon'), \
jpayne@68 119 'daemonic processes are not allowed to have children'
jpayne@68 120 _cleanup()
jpayne@68 121 self._popen = self._Popen(self)
jpayne@68 122 self._sentinel = self._popen.sentinel
jpayne@68 123 # Avoid a refcycle if the target function holds an indirect
jpayne@68 124 # reference to the process object (see bpo-30775)
jpayne@68 125 del self._target, self._args, self._kwargs
jpayne@68 126 _children.add(self)
jpayne@68 127
jpayne@68 128 def terminate(self):
jpayne@68 129 '''
jpayne@68 130 Terminate process; sends SIGTERM signal or uses TerminateProcess()
jpayne@68 131 '''
jpayne@68 132 self._check_closed()
jpayne@68 133 self._popen.terminate()
jpayne@68 134
jpayne@68 135 def kill(self):
jpayne@68 136 '''
jpayne@68 137 Terminate process; sends SIGKILL signal or uses TerminateProcess()
jpayne@68 138 '''
jpayne@68 139 self._check_closed()
jpayne@68 140 self._popen.kill()
jpayne@68 141
jpayne@68 142 def join(self, timeout=None):
jpayne@68 143 '''
jpayne@68 144 Wait until child process terminates
jpayne@68 145 '''
jpayne@68 146 self._check_closed()
jpayne@68 147 assert self._parent_pid == os.getpid(), 'can only join a child process'
jpayne@68 148 assert self._popen is not None, 'can only join a started process'
jpayne@68 149 res = self._popen.wait(timeout)
jpayne@68 150 if res is not None:
jpayne@68 151 _children.discard(self)
jpayne@68 152
jpayne@68 153 def is_alive(self):
jpayne@68 154 '''
jpayne@68 155 Return whether process is alive
jpayne@68 156 '''
jpayne@68 157 self._check_closed()
jpayne@68 158 if self is _current_process:
jpayne@68 159 return True
jpayne@68 160 assert self._parent_pid == os.getpid(), 'can only test a child process'
jpayne@68 161
jpayne@68 162 if self._popen is None:
jpayne@68 163 return False
jpayne@68 164
jpayne@68 165 returncode = self._popen.poll()
jpayne@68 166 if returncode is None:
jpayne@68 167 return True
jpayne@68 168 else:
jpayne@68 169 _children.discard(self)
jpayne@68 170 return False
jpayne@68 171
jpayne@68 172 def close(self):
jpayne@68 173 '''
jpayne@68 174 Close the Process object.
jpayne@68 175
jpayne@68 176 This method releases resources held by the Process object. It is
jpayne@68 177 an error to call this method if the child process is still running.
jpayne@68 178 '''
jpayne@68 179 if self._popen is not None:
jpayne@68 180 if self._popen.poll() is None:
jpayne@68 181 raise ValueError("Cannot close a process while it is still running. "
jpayne@68 182 "You should first call join() or terminate().")
jpayne@68 183 self._popen.close()
jpayne@68 184 self._popen = None
jpayne@68 185 del self._sentinel
jpayne@68 186 _children.discard(self)
jpayne@68 187 self._closed = True
jpayne@68 188
jpayne@68 189 @property
jpayne@68 190 def name(self):
jpayne@68 191 return self._name
jpayne@68 192
jpayne@68 193 @name.setter
jpayne@68 194 def name(self, name):
jpayne@68 195 assert isinstance(name, str), 'name must be a string'
jpayne@68 196 self._name = name
jpayne@68 197
jpayne@68 198 @property
jpayne@68 199 def daemon(self):
jpayne@68 200 '''
jpayne@68 201 Return whether process is a daemon
jpayne@68 202 '''
jpayne@68 203 return self._config.get('daemon', False)
jpayne@68 204
jpayne@68 205 @daemon.setter
jpayne@68 206 def daemon(self, daemonic):
jpayne@68 207 '''
jpayne@68 208 Set whether process is a daemon
jpayne@68 209 '''
jpayne@68 210 assert self._popen is None, 'process has already started'
jpayne@68 211 self._config['daemon'] = daemonic
jpayne@68 212
jpayne@68 213 @property
jpayne@68 214 def authkey(self):
jpayne@68 215 return self._config['authkey']
jpayne@68 216
jpayne@68 217 @authkey.setter
jpayne@68 218 def authkey(self, authkey):
jpayne@68 219 '''
jpayne@68 220 Set authorization key of process
jpayne@68 221 '''
jpayne@68 222 self._config['authkey'] = AuthenticationString(authkey)
jpayne@68 223
jpayne@68 224 @property
jpayne@68 225 def exitcode(self):
jpayne@68 226 '''
jpayne@68 227 Return exit code of process or `None` if it has yet to stop
jpayne@68 228 '''
jpayne@68 229 self._check_closed()
jpayne@68 230 if self._popen is None:
jpayne@68 231 return self._popen
jpayne@68 232 return self._popen.poll()
jpayne@68 233
jpayne@68 234 @property
jpayne@68 235 def ident(self):
jpayne@68 236 '''
jpayne@68 237 Return identifier (PID) of process or `None` if it has yet to start
jpayne@68 238 '''
jpayne@68 239 self._check_closed()
jpayne@68 240 if self is _current_process:
jpayne@68 241 return os.getpid()
jpayne@68 242 else:
jpayne@68 243 return self._popen and self._popen.pid
jpayne@68 244
jpayne@68 245 pid = ident
jpayne@68 246
jpayne@68 247 @property
jpayne@68 248 def sentinel(self):
jpayne@68 249 '''
jpayne@68 250 Return a file descriptor (Unix) or handle (Windows) suitable for
jpayne@68 251 waiting for process termination.
jpayne@68 252 '''
jpayne@68 253 self._check_closed()
jpayne@68 254 try:
jpayne@68 255 return self._sentinel
jpayne@68 256 except AttributeError:
jpayne@68 257 raise ValueError("process not started") from None
jpayne@68 258
jpayne@68 259 def __repr__(self):
jpayne@68 260 exitcode = None
jpayne@68 261 if self is _current_process:
jpayne@68 262 status = 'started'
jpayne@68 263 elif self._closed:
jpayne@68 264 status = 'closed'
jpayne@68 265 elif self._parent_pid != os.getpid():
jpayne@68 266 status = 'unknown'
jpayne@68 267 elif self._popen is None:
jpayne@68 268 status = 'initial'
jpayne@68 269 else:
jpayne@68 270 exitcode = self._popen.poll()
jpayne@68 271 if exitcode is not None:
jpayne@68 272 status = 'stopped'
jpayne@68 273 else:
jpayne@68 274 status = 'started'
jpayne@68 275
jpayne@68 276 info = [type(self).__name__, 'name=%r' % self._name]
jpayne@68 277 if self._popen is not None:
jpayne@68 278 info.append('pid=%s' % self._popen.pid)
jpayne@68 279 info.append('parent=%s' % self._parent_pid)
jpayne@68 280 info.append(status)
jpayne@68 281 if exitcode is not None:
jpayne@68 282 exitcode = _exitcode_to_name.get(exitcode, exitcode)
jpayne@68 283 info.append('exitcode=%s' % exitcode)
jpayne@68 284 if self.daemon:
jpayne@68 285 info.append('daemon')
jpayne@68 286 return '<%s>' % ' '.join(info)
jpayne@68 287
jpayne@68 288 ##
jpayne@68 289
jpayne@68 290 def _bootstrap(self, parent_sentinel=None):
jpayne@68 291 from . import util, context
jpayne@68 292 global _current_process, _parent_process, _process_counter, _children
jpayne@68 293
jpayne@68 294 try:
jpayne@68 295 if self._start_method is not None:
jpayne@68 296 context._force_start_method(self._start_method)
jpayne@68 297 _process_counter = itertools.count(1)
jpayne@68 298 _children = set()
jpayne@68 299 util._close_stdin()
jpayne@68 300 old_process = _current_process
jpayne@68 301 _current_process = self
jpayne@68 302 _parent_process = _ParentProcess(
jpayne@68 303 self._parent_name, self._parent_pid, parent_sentinel)
jpayne@68 304 if threading._HAVE_THREAD_NATIVE_ID:
jpayne@68 305 threading.main_thread()._set_native_id()
jpayne@68 306 try:
jpayne@68 307 util._finalizer_registry.clear()
jpayne@68 308 util._run_after_forkers()
jpayne@68 309 finally:
jpayne@68 310 # delay finalization of the old process object until after
jpayne@68 311 # _run_after_forkers() is executed
jpayne@68 312 del old_process
jpayne@68 313 util.info('child process calling self.run()')
jpayne@68 314 try:
jpayne@68 315 self.run()
jpayne@68 316 exitcode = 0
jpayne@68 317 finally:
jpayne@68 318 util._exit_function()
jpayne@68 319 except SystemExit as e:
jpayne@68 320 if not e.args:
jpayne@68 321 exitcode = 1
jpayne@68 322 elif isinstance(e.args[0], int):
jpayne@68 323 exitcode = e.args[0]
jpayne@68 324 else:
jpayne@68 325 sys.stderr.write(str(e.args[0]) + '\n')
jpayne@68 326 exitcode = 1
jpayne@68 327 except:
jpayne@68 328 exitcode = 1
jpayne@68 329 import traceback
jpayne@68 330 sys.stderr.write('Process %s:\n' % self.name)
jpayne@68 331 traceback.print_exc()
jpayne@68 332 finally:
jpayne@68 333 threading._shutdown()
jpayne@68 334 util.info('process exiting with exitcode %d' % exitcode)
jpayne@68 335 util._flush_std_streams()
jpayne@68 336
jpayne@68 337 return exitcode
jpayne@68 338
jpayne@68 339 #
jpayne@68 340 # We subclass bytes to avoid accidental transmission of auth keys over network
jpayne@68 341 #
jpayne@68 342
jpayne@68 343 class AuthenticationString(bytes):
jpayne@68 344 def __reduce__(self):
jpayne@68 345 from .context import get_spawning_popen
jpayne@68 346 if get_spawning_popen() is None:
jpayne@68 347 raise TypeError(
jpayne@68 348 'Pickling an AuthenticationString object is '
jpayne@68 349 'disallowed for security reasons'
jpayne@68 350 )
jpayne@68 351 return AuthenticationString, (bytes(self),)
jpayne@68 352
jpayne@68 353
jpayne@68 354 #
jpayne@68 355 # Create object representing the parent process
jpayne@68 356 #
jpayne@68 357
jpayne@68 358 class _ParentProcess(BaseProcess):
jpayne@68 359
jpayne@68 360 def __init__(self, name, pid, sentinel):
jpayne@68 361 self._identity = ()
jpayne@68 362 self._name = name
jpayne@68 363 self._pid = pid
jpayne@68 364 self._parent_pid = None
jpayne@68 365 self._popen = None
jpayne@68 366 self._closed = False
jpayne@68 367 self._sentinel = sentinel
jpayne@68 368 self._config = {}
jpayne@68 369
jpayne@68 370 def is_alive(self):
jpayne@68 371 from multiprocessing.connection import wait
jpayne@68 372 return not wait([self._sentinel], timeout=0)
jpayne@68 373
jpayne@68 374 @property
jpayne@68 375 def ident(self):
jpayne@68 376 return self._pid
jpayne@68 377
jpayne@68 378 def join(self, timeout=None):
jpayne@68 379 '''
jpayne@68 380 Wait until parent process terminates
jpayne@68 381 '''
jpayne@68 382 from multiprocessing.connection import wait
jpayne@68 383 wait([self._sentinel], timeout=timeout)
jpayne@68 384
jpayne@68 385 pid = ident
jpayne@68 386
jpayne@68 387 #
jpayne@68 388 # Create object representing the main process
jpayne@68 389 #
jpayne@68 390
jpayne@68 391 class _MainProcess(BaseProcess):
jpayne@68 392
jpayne@68 393 def __init__(self):
jpayne@68 394 self._identity = ()
jpayne@68 395 self._name = 'MainProcess'
jpayne@68 396 self._parent_pid = None
jpayne@68 397 self._popen = None
jpayne@68 398 self._closed = False
jpayne@68 399 self._config = {'authkey': AuthenticationString(os.urandom(32)),
jpayne@68 400 'semprefix': '/mp'}
jpayne@68 401 # Note that some versions of FreeBSD only allow named
jpayne@68 402 # semaphores to have names of up to 14 characters. Therefore
jpayne@68 403 # we choose a short prefix.
jpayne@68 404 #
jpayne@68 405 # On MacOSX in a sandbox it may be necessary to use a
jpayne@68 406 # different prefix -- see #19478.
jpayne@68 407 #
jpayne@68 408 # Everything in self._config will be inherited by descendant
jpayne@68 409 # processes.
jpayne@68 410
jpayne@68 411 def close(self):
jpayne@68 412 pass
jpayne@68 413
jpayne@68 414
jpayne@68 415 _parent_process = None
jpayne@68 416 _current_process = _MainProcess()
jpayne@68 417 _process_counter = itertools.count(1)
jpayne@68 418 _children = set()
jpayne@68 419 del _MainProcess
jpayne@68 420
jpayne@68 421 #
jpayne@68 422 # Give names to some return codes
jpayne@68 423 #
jpayne@68 424
jpayne@68 425 _exitcode_to_name = {}
jpayne@68 426
jpayne@68 427 for name, signum in list(signal.__dict__.items()):
jpayne@68 428 if name[:3]=='SIG' and '_' not in name:
jpayne@68 429 _exitcode_to_name[-signum] = f'-{name}'
jpayne@68 430
jpayne@68 431 # For debug and leak testing
jpayne@68 432 _dangling = WeakSet()