Mercurial > repos > rliterman > csp2

annotate CSP2/CSP2_env/env-d9b9114564458d9d-741b3de822f2aaca6c6caa4325c4afce/lib/python3.8/multiprocessing/process.py @ 69:33d812a61356

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