diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py index abd88adf7..f577186e9 100644 --- a/Lib/multiprocessing/connection.py +++ b/Lib/multiprocessing/connection.py @@ -11,13 +11,12 @@ __all__ = [ 'Client', 'Listener', 'Pipe', 'wait' ] import errno import io +import itertools import os import sys import socket import struct import time -import tempfile -import itertools from . import util @@ -39,11 +38,14 @@ except ImportError: # # -BUFSIZE = 8192 +# 64 KiB is the default PIPE buffer size of most POSIX platforms. +BUFSIZE = 64 * 1024 + # A very generous timeout when it comes to local connections... CONNECTION_TIMEOUT = 20. _mmap_counter = itertools.count() +_MAX_PIPE_ATTEMPTS = 100 default_family = 'AF_INET' families = ['AF_INET'] @@ -74,10 +76,14 @@ def arbitrary_address(family): if family == 'AF_INET': return ('localhost', 0) elif family == 'AF_UNIX': - return tempfile.mktemp(prefix='sock-', dir=util.get_temp_dir()) + # NOTE: util.get_temp_dir() is a 0o700 per-process directory. A + # mktemp-style ToC vs ToU concern is not important; bind() surfaces + # the extremely unlikely collision as EADDRINUSE. + return os.path.join(util.get_temp_dir(), + f'sock-{os.urandom(6).hex()}') elif family == 'AF_PIPE': - return tempfile.mktemp(prefix=r'\\.\pipe\pyc-%d-%d-' % - (os.getpid(), next(_mmap_counter)), dir="") + return (r'\\.\pipe\pyc-%d-%d-%s' % + (os.getpid(), next(_mmap_counter), os.urandom(8).hex())) else: raise ValueError('unrecognized family') @@ -179,6 +185,10 @@ class _ConnectionBase: finally: self._handle = None + def _detach(self): + """Stop managing the underlying file descriptor or handle.""" + self._handle = None + def send_bytes(self, buf, offset=0, size=None): """Send the bytes data from a bytes-like object""" self._check_closed() @@ -316,22 +326,32 @@ if _winapi: try: ov, err = _winapi.ReadFile(self._handle, bsize, overlapped=True) + + sentinel = object() + return_value = sentinel try: - if err == _winapi.ERROR_IO_PENDING: - waitres = _winapi.WaitForMultipleObjects( - [ov.event], False, INFINITE) - assert waitres == WAIT_OBJECT_0 + try: + if err == _winapi.ERROR_IO_PENDING: + waitres = _winapi.WaitForMultipleObjects( + [ov.event], False, INFINITE) + assert waitres == WAIT_OBJECT_0 + except: + ov.cancel() + raise + finally: + nread, err = ov.GetOverlappedResult(True) + if err == 0: + f = io.BytesIO() + f.write(ov.getbuffer()) + return_value = f + elif err == _winapi.ERROR_MORE_DATA: + return_value = self._get_more_data(ov, maxsize) except: - ov.cancel() - raise - finally: - nread, err = ov.GetOverlappedResult(True) - if err == 0: - f = io.BytesIO() - f.write(ov.getbuffer()) - return f - elif err == _winapi.ERROR_MORE_DATA: - return self._get_more_data(ov, maxsize) + if return_value is sentinel: + raise + + if return_value is not sentinel: + return return_value except OSError as e: if e.winerror == _winapi.ERROR_BROKEN_PIPE: raise EOFError @@ -392,7 +412,8 @@ class Connection(_ConnectionBase): handle = self._handle remaining = size while remaining > 0: - chunk = read(handle, remaining) + to_read = min(BUFSIZE, remaining) + chunk = read(handle, to_read) n = len(chunk) if n == 0: if remaining == size: @@ -455,17 +476,29 @@ class Listener(object): def __init__(self, address=None, family=None, backlog=1, authkey=None): family = family or (address and address_type(address)) \ or default_family - address = address or arbitrary_address(family) - _validate_family(family) - if family == 'AF_PIPE': - self._listener = PipeListener(address, backlog) - else: - self._listener = SocketListener(address, family, backlog) - if authkey is not None and not isinstance(authkey, bytes): raise TypeError('authkey should be a byte string') + if family == 'AF_PIPE': + if address: + self._listener = PipeListener(address, backlog) + else: + for attempts in itertools.count(): + address = arbitrary_address(family) + try: + self._listener = PipeListener(address, backlog) + break + except OSError as e: + if attempts >= _MAX_PIPE_ATTEMPTS: + raise + if e.winerror not in (_winapi.ERROR_PIPE_BUSY, + _winapi.ERROR_ACCESS_DENIED): + raise + else: + address = address or arbitrary_address(family) + self._listener = SocketListener(address, family, backlog) + self._authkey = authkey def accept(self): @@ -553,7 +586,6 @@ else: ''' Returns pair of connection objects at either end of a pipe ''' - address = arbitrary_address('AF_PIPE') if duplex: openmode = _winapi.PIPE_ACCESS_DUPLEX access = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE @@ -563,15 +595,25 @@ else: access = _winapi.GENERIC_WRITE obsize, ibsize = 0, BUFSIZE - h1 = _winapi.CreateNamedPipe( - address, openmode | _winapi.FILE_FLAG_OVERLAPPED | - _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE, - _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE | - _winapi.PIPE_WAIT, - 1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER, - # default security descriptor: the handle cannot be inherited - _winapi.NULL - ) + for attempts in itertools.count(): + address = arbitrary_address('AF_PIPE') + try: + h1 = _winapi.CreateNamedPipe( + address, openmode | _winapi.FILE_FLAG_OVERLAPPED | + _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE, + _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE | + _winapi.PIPE_WAIT, + 1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER, + # default security descriptor: the handle cannot be inherited + _winapi.NULL + ) + break + except OSError as e: + if attempts >= _MAX_PIPE_ATTEMPTS: + raise + if e.winerror not in (_winapi.ERROR_PIPE_BUSY, + _winapi.ERROR_ACCESS_DENIED): + raise h2 = _winapi.CreateFile( address, access, 0, _winapi.NULL, _winapi.OPEN_EXISTING, _winapi.FILE_FLAG_OVERLAPPED, _winapi.NULL diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index f395e8b04..5fa6d7e48 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -145,7 +145,13 @@ class BaseContext(object): '''Check whether this is a fake forked process in a frozen executable. If so then run code specified by commandline and exit. ''' - if self.get_start_method() == 'spawn' and getattr(sys, 'frozen', False): + # gh-140814: allow_none=True avoids locking in the default start + # method, which would cause a later set_start_method() to fail. + # None is safe to pass through: spawn.freeze_support() + # independently detects whether this process is a spawned + # child, so the start method check here is only an optimization. + if (getattr(sys, 'frozen', False) + and self.get_start_method(allow_none=True) in ('spawn', None)): from .spawn import freeze_support freeze_support() @@ -167,7 +173,7 @@ class BaseContext(object): ''' # This is undocumented. In previous versions of multiprocessing # its only effect was to make socket objects inheritable on Windows. - from . import connection + from . import connection # noqa: F401 def set_executable(self, executable): '''Sets the path to a python.exe or pythonw.exe binary used to run @@ -259,13 +265,12 @@ class DefaultContext(BaseContext): def get_all_start_methods(self): """Returns a list of the supported start methods, default first.""" - if sys.platform == 'win32': - return ['spawn'] - else: - methods = ['spawn', 'fork'] if sys.platform == 'darwin' else ['fork', 'spawn'] - if reduction.HAVE_SEND_HANDLE: - methods.append('forkserver') - return methods + default = self._default_context.get_start_method() + start_method_names = [default] + start_method_names.extend( + name for name in _concrete_contexts if name != default + ) + return start_method_names # @@ -320,14 +325,15 @@ if sys.platform != 'win32': 'spawn': SpawnContext(), 'forkserver': ForkServerContext(), } - if sys.platform == 'darwin': - # bpo-33725: running arbitrary code after fork() is no longer reliable - # on macOS since macOS 10.14 (Mojave). Use spawn by default instead. - _default_context = DefaultContext(_concrete_contexts['spawn']) + # bpo-33725: running arbitrary code after fork() is no longer reliable + # on macOS since macOS 10.14 (Mojave). Use spawn by default instead. + # gh-84559: We changed everyones default to a thread safeish one in 3.14. + if reduction.HAVE_SEND_HANDLE and sys.platform != 'darwin': + _default_context = DefaultContext(_concrete_contexts['forkserver']) else: - _default_context = DefaultContext(_concrete_contexts['fork']) + _default_context = DefaultContext(_concrete_contexts['spawn']) -else: +else: # Windows class SpawnProcess(process.BaseProcess): _start_method = 'spawn' diff --git a/Lib/multiprocessing/dummy/__init__.py b/Lib/multiprocessing/dummy/__init__.py index 6a1468609..7dc5d1c8d 100644 --- a/Lib/multiprocessing/dummy/__init__.py +++ b/Lib/multiprocessing/dummy/__init__.py @@ -33,7 +33,7 @@ from queue import Queue class DummyProcess(threading.Thread): - def __init__(self, group=None, target=None, name=None, args=(), kwargs={}): + def __init__(self, group=None, target=None, name=None, args=(), kwargs=None): threading.Thread.__init__(self, group, target, name, args, kwargs) self._pid = None self._children = weakref.WeakKeyDictionary() diff --git a/Lib/multiprocessing/forkserver.py b/Lib/multiprocessing/forkserver.py index e243442e7..e431b3f12 100644 --- a/Lib/multiprocessing/forkserver.py +++ b/Lib/multiprocessing/forkserver.py @@ -9,6 +9,7 @@ import sys import threading import warnings +from . import AuthenticationError from . import connection from . import process from .context import reduction @@ -25,6 +26,7 @@ __all__ = ['ensure_running', 'get_inherited_fds', 'connect_to_new_process', MAXFDS_TO_SEND = 256 SIGNED_STRUCT = struct.Struct('q') # large enough for pid_t +_AUTHKEY_LEN = 32 # <= PIPEBUF so it fits a single write to an empty pipe. # # Forkserver class @@ -33,6 +35,7 @@ SIGNED_STRUCT = struct.Struct('q') # large enough for pid_t class ForkServer(object): def __init__(self): + self._forkserver_authkey = None self._forkserver_address = None self._forkserver_alive_fd = None self._forkserver_pid = None @@ -59,6 +62,7 @@ class ForkServer(object): if not util.is_abstract_socket_namespace(self._forkserver_address): os.unlink(self._forkserver_address) self._forkserver_address = None + self._forkserver_authkey = None def set_forkserver_preload(self, modules_names): '''Set list of module names to try to load in forkserver process.''' @@ -83,6 +87,7 @@ class ForkServer(object): process data. ''' self.ensure_running() + assert self._forkserver_authkey if len(fds) + 4 >= MAXFDS_TO_SEND: raise ValueError('too many fds') with socket.socket(socket.AF_UNIX) as client: @@ -93,6 +98,18 @@ class ForkServer(object): resource_tracker.getfd()] allfds += fds try: + client.setblocking(True) + wrapped_client = connection.Connection(client.fileno()) + # The other side of this exchange happens in the child as + # implemented in main(). + try: + connection.answer_challenge( + wrapped_client, self._forkserver_authkey) + connection.deliver_challenge( + wrapped_client, self._forkserver_authkey) + finally: + wrapped_client._detach() + del wrapped_client reduction.sendfds(client, allfds) return parent_r, parent_w except: @@ -120,20 +137,30 @@ class ForkServer(object): return # dead, launch it again os.close(self._forkserver_alive_fd) + self._forkserver_authkey = None self._forkserver_address = None self._forkserver_alive_fd = None self._forkserver_pid = None - cmd = ('from multiprocessing.forkserver import main; ' + - 'main(%d, %d, %r, **%r)') + # gh-144503: sys_argv is passed as real argv elements after the + # ``-c cmd`` rather than repr'd into main_kws so that a large + # parent sys.argv cannot push the single ``-c`` command string + # over the OS per-argument length limit (MAX_ARG_STRLEN on Linux). + # The child sees them as sys.argv[1:]. + cmd = ('import sys; ' + 'from multiprocessing.forkserver import main; ' + 'main(%d, %d, %r, sys_argv=sys.argv[1:], **%r)') main_kws = {} + sys_argv = None if self._preload_modules: data = spawn.get_preparation_data('ignore') if 'sys_path' in data: main_kws['sys_path'] = data['sys_path'] if 'init_main_from_path' in data: main_kws['main_path'] = data['init_main_from_path'] + if 'sys_argv' in data: + sys_argv = data['sys_argv'] with socket.socket(socket.AF_UNIX) as listener: address = connection.arbitrary_address('AF_UNIX') @@ -145,19 +172,33 @@ class ForkServer(object): # all client processes own the write end of the "alive" pipe; # when they all terminate the read end becomes ready. alive_r, alive_w = os.pipe() + # A short lived pipe to initialize the forkserver authkey. + authkey_r, authkey_w = os.pipe() try: - fds_to_pass = [listener.fileno(), alive_r] + fds_to_pass = [listener.fileno(), alive_r, authkey_r] + main_kws['authkey_r'] = authkey_r cmd %= (listener.fileno(), alive_r, self._preload_modules, main_kws) exe = spawn.get_executable() args = [exe] + util._args_from_interpreter_flags() args += ['-c', cmd] + if sys_argv is not None: + args += sys_argv pid = util.spawnv_passfds(exe, args, fds_to_pass) except: os.close(alive_w) + os.close(authkey_w) raise finally: os.close(alive_r) + os.close(authkey_r) + # Authenticate our control socket to prevent access from + # processes we have not shared this key with. + try: + self._forkserver_authkey = os.urandom(_AUTHKEY_LEN) + os.write(authkey_w, self._forkserver_authkey) + finally: + os.close(authkey_w) self._forkserver_address = address self._forkserver_alive_fd = alive_w self._forkserver_pid = pid @@ -166,9 +207,21 @@ class ForkServer(object): # # -def main(listener_fd, alive_r, preload, main_path=None, sys_path=None): - '''Run forkserver.''' +def main(listener_fd, alive_r, preload, main_path=None, sys_path=None, + *, sys_argv=None, authkey_r=None): + """Run forkserver.""" + if authkey_r is not None: + try: + authkey = os.read(authkey_r, _AUTHKEY_LEN) + assert len(authkey) == _AUTHKEY_LEN, f'{len(authkey)} < {_AUTHKEY_LEN}' + finally: + os.close(authkey_r) + else: + authkey = b'' + if preload: + if sys_argv is not None: + sys.argv[:] = sys_argv if sys_path is not None: sys.path[:] = sys_path if '__main__' in preload and main_path is not None: @@ -262,8 +315,24 @@ def main(listener_fd, alive_r, preload, main_path=None, sys_path=None): if listener in rfds: # Incoming fork request with listener.accept()[0] as s: - # Receive fds from client - fds = reduction.recvfds(s, MAXFDS_TO_SEND + 1) + try: + if authkey: + wrapped_s = connection.Connection(s.fileno()) + # The other side of this exchange happens in + # in connect_to_new_process(). + try: + connection.deliver_challenge( + wrapped_s, authkey) + connection.answer_challenge( + wrapped_s, authkey) + finally: + wrapped_s._detach() + del wrapped_s + # Receive fds from client + fds = reduction.recvfds(s, MAXFDS_TO_SEND + 1) + except (EOFError, BrokenPipeError, AuthenticationError): + s.close() + continue if len(fds) > MAXFDS_TO_SEND: raise RuntimeError( "Too many ({0:n}) fds to send".format( @@ -331,13 +400,14 @@ def _serve_one(child_r, fds, unused_fds, handlers): # def read_signed(fd): - data = b'' - length = SIGNED_STRUCT.size - while len(data) < length: - s = os.read(fd, length - len(data)) - if not s: + data = bytearray(SIGNED_STRUCT.size) + unread = memoryview(data) + while unread: + count = os.readinto(fd, unread) + if count == 0: raise EOFError('unexpected EOF') - data += s + unread = unread[count:] + return SIGNED_STRUCT.unpack(data)[0] def write_signed(fd, n): diff --git a/Lib/multiprocessing/managers.py b/Lib/multiprocessing/managers.py index ef791c275..91bcf243e 100644 --- a/Lib/multiprocessing/managers.py +++ b/Lib/multiprocessing/managers.py @@ -18,6 +18,7 @@ import sys import threading import signal import array +import collections.abc import queue import time import types @@ -1058,12 +1059,14 @@ class IteratorProxy(BaseProxy): class AcquirerProxy(BaseProxy): - _exposed_ = ('acquire', 'release') + _exposed_ = ('acquire', 'release', 'locked') def acquire(self, blocking=True, timeout=None): args = (blocking,) if timeout is None else (blocking, timeout) return self._callmethod('acquire', args) def release(self): return self._callmethod('release') + def locked(self): + return self._callmethod('locked') def __enter__(self): return self._callmethod('acquire') def __exit__(self, exc_type, exc_val, exc_tb): @@ -1071,7 +1074,7 @@ class AcquirerProxy(BaseProxy): class ConditionProxy(AcquirerProxy): - _exposed_ = ('acquire', 'release', 'wait', 'notify', 'notify_all') + _exposed_ = ('acquire', 'release', 'locked', 'wait', 'notify', 'notify_all') def wait(self, timeout=None): return self._callmethod('wait', (timeout,)) def notify(self, n=1): @@ -1159,10 +1162,10 @@ class ValueProxy(BaseProxy): BaseListProxy = MakeProxyType('BaseListProxy', ( - '__add__', '__contains__', '__delitem__', '__getitem__', '__len__', - '__mul__', '__reversed__', '__rmul__', '__setitem__', - 'append', 'count', 'extend', 'index', 'insert', 'pop', 'remove', - 'reverse', 'sort', '__imul__' + '__add__', '__contains__', '__delitem__', '__getitem__', '__imul__', + '__len__', '__mul__', '__reversed__', '__rmul__', '__setitem__', + 'append', 'clear', 'copy', 'count', 'extend', 'index', 'insert', 'pop', + 'remove', 'reverse', 'sort', )) class ListProxy(BaseListProxy): def __iadd__(self, value): @@ -1174,18 +1177,55 @@ class ListProxy(BaseListProxy): __class_getitem__ = classmethod(types.GenericAlias) +collections.abc.MutableSequence.register(BaseListProxy) -_BaseDictProxy = MakeProxyType('DictProxy', ( - '__contains__', '__delitem__', '__getitem__', '__iter__', '__len__', - '__setitem__', 'clear', 'copy', 'get', 'items', +_BaseDictProxy = MakeProxyType('_BaseDictProxy', ( + '__contains__', '__delitem__', '__getitem__', '__ior__', '__iter__', + '__len__', '__or__', '__reversed__', '__ror__', + '__setitem__', 'clear', 'copy', 'fromkeys', 'get', 'items', 'keys', 'pop', 'popitem', 'setdefault', 'update', 'values' )) _BaseDictProxy._method_to_typeid_ = { '__iter__': 'Iterator', } class DictProxy(_BaseDictProxy): + def __ior__(self, value): + self._callmethod('__ior__', (value,)) + return self + __class_getitem__ = classmethod(types.GenericAlias) +collections.abc.MutableMapping.register(_BaseDictProxy) + +_BaseSetProxy = MakeProxyType("_BaseSetProxy", ( + '__and__', '__class_getitem__', '__contains__', '__iand__', '__ior__', + '__isub__', '__iter__', '__ixor__', '__len__', '__or__', '__rand__', + '__ror__', '__rsub__', '__rxor__', '__sub__', '__xor__', + '__ge__', '__gt__', '__le__', '__lt__', + 'add', 'clear', 'copy', 'difference', 'difference_update', 'discard', + 'intersection', 'intersection_update', 'isdisjoint', 'issubset', + 'issuperset', 'pop', 'remove', 'symmetric_difference', + 'symmetric_difference_update', 'union', 'update', +)) + +class SetProxy(_BaseSetProxy): + def __ior__(self, value): + self._callmethod('__ior__', (value,)) + return self + def __iand__(self, value): + self._callmethod('__iand__', (value,)) + return self + def __ixor__(self, value): + self._callmethod('__ixor__', (value,)) + return self + def __isub__(self, value): + self._callmethod('__isub__', (value,)) + return self + + __class_getitem__ = classmethod(types.GenericAlias) + +collections.abc.MutableMapping.register(_BaseSetProxy) + ArrayProxy = MakeProxyType('ArrayProxy', ( '__len__', '__getitem__', '__setitem__' @@ -1237,6 +1277,7 @@ SyncManager.register('Barrier', threading.Barrier, BarrierProxy) SyncManager.register('Pool', pool.Pool, PoolProxy) SyncManager.register('list', list, ListProxy) SyncManager.register('dict', dict, DictProxy) +SyncManager.register('set', set, SetProxy) SyncManager.register('Value', Value, ValueProxy) SyncManager.register('Array', Array, ArrayProxy) SyncManager.register('Namespace', Namespace, NamespaceProxy) diff --git a/Lib/multiprocessing/popen_fork.py b/Lib/multiprocessing/popen_fork.py index a57ef6bda..a02a53b6a 100644 --- a/Lib/multiprocessing/popen_fork.py +++ b/Lib/multiprocessing/popen_fork.py @@ -54,6 +54,9 @@ class Popen(object): if self.wait(timeout=0.1) is None: raise + def interrupt(self): + self._send_signal(signal.SIGINT) + def terminate(self): self._send_signal(signal.SIGTERM) @@ -64,7 +67,17 @@ class Popen(object): code = 1 parent_r, child_w = os.pipe() child_r, parent_w = os.pipe() - self.pid = os.fork() + # gh-146313: Tell the resource tracker's at-fork handler to keep + # the inherited pipe fd so this child reuses the parent's tracker + # (gh-80849) rather than closing it and launching its own. + from .resource_tracker import _fork_intent + _fork_intent.preserve_fd = True + try: + self.pid = os.fork() + finally: + # Reset in both parent and child so the flag does not leak + # into a subsequent raw os.fork() or nested Process launch. + _fork_intent.preserve_fd = False if self.pid == 0: try: atexit._clear() diff --git a/Lib/multiprocessing/process.py b/Lib/multiprocessing/process.py index b45f7df47..262513f29 100644 --- a/Lib/multiprocessing/process.py +++ b/Lib/multiprocessing/process.py @@ -77,7 +77,7 @@ class BaseProcess(object): def _Popen(self): raise NotImplementedError - def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, + def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None): assert group is None, 'group argument must be None for now' count = next(_process_counter) @@ -89,7 +89,7 @@ class BaseProcess(object): self._closed = False self._target = target self._args = tuple(args) - self._kwargs = dict(kwargs) + self._kwargs = dict(kwargs) if kwargs else {} self._name = name or type(self).__name__ + '-' + \ ':'.join(str(i) for i in self._identity) if daemon is not None: @@ -125,6 +125,13 @@ class BaseProcess(object): del self._target, self._args, self._kwargs _children.add(self) + def interrupt(self): + ''' + Terminate process; sends SIGINT signal + ''' + self._check_closed() + self._popen.interrupt() + def terminate(self): ''' Terminate process; sends SIGTERM signal or uses TerminateProcess() diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py index 925f04390..981599acf 100644 --- a/Lib/multiprocessing/queues.py +++ b/Lib/multiprocessing/queues.py @@ -121,7 +121,7 @@ class Queue(object): def qsize(self): # Raises NotImplementedError on Mac OSX because of broken sem_getvalue() - return self._maxsize - self._sem._semlock._get_value() + return self._maxsize - self._sem.get_value() def empty(self): return not self._poll() diff --git a/Lib/multiprocessing/reduction.py b/Lib/multiprocessing/reduction.py index 5593f0682..fcccd3eef 100644 --- a/Lib/multiprocessing/reduction.py +++ b/Lib/multiprocessing/reduction.py @@ -139,15 +139,12 @@ else: __all__ += ['DupFd', 'sendfds', 'recvfds'] import array - # On MacOSX we should acknowledge receipt of fds -- see Issue14669 - ACKNOWLEDGE = sys.platform == 'darwin' - def sendfds(sock, fds): '''Send an array of fds over an AF_UNIX socket.''' fds = array.array('i', fds) msg = bytes([len(fds) % 256]) sock.sendmsg([msg], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, fds)]) - if ACKNOWLEDGE and sock.recv(1) != b'A': + if sock.recv(1) != b'A': raise RuntimeError('did not receive acknowledgement of fd') def recvfds(sock, size): @@ -158,8 +155,11 @@ else: if not msg and not ancdata: raise EOFError try: - if ACKNOWLEDGE: - sock.send(b'A') + # We send/recv an Ack byte after the fds to work around an old + # macOS bug; it isn't clear if this is still required but it + # makes unit testing fd sending easier. + # See: https://github.com/python/cpython/issues/58874 + sock.send(b'A') # Acknowledge if len(ancdata) != 1: raise RuntimeError('received %d items of ancdata' % len(ancdata)) diff --git a/Lib/multiprocessing/resource_tracker.py b/Lib/multiprocessing/resource_tracker.py index 22e3bbcf2..dfb5a3dfe 100644 --- a/Lib/multiprocessing/resource_tracker.py +++ b/Lib/multiprocessing/resource_tracker.py @@ -20,6 +20,7 @@ import os import signal import sys import threading +import time import warnings from collections import deque @@ -51,12 +52,8 @@ if os.name == 'posix': # absence of POSIX named semaphores. In that case, no named semaphores were # ever opened, so no cleanup would be necessary. if hasattr(_multiprocessing, 'sem_unlink'): - _CLEANUP_FUNCS.update({ - 'semaphore': _multiprocessing.sem_unlink, - }) - _CLEANUP_FUNCS.update({ - 'shared_memory': _posixshmem.shm_unlink, - }) + _CLEANUP_FUNCS['semaphore'] = _multiprocessing.sem_unlink + _CLEANUP_FUNCS['shared_memory'] = _posixshmem.shm_unlink class ReentrantCallError(RuntimeError): @@ -79,6 +76,10 @@ class ResourceTracker(object): # The reader should understand all formats. self._use_simple_format = True + # Set to True by _stop_locked() if the waitpid polling loop ran to + # its timeout without reaping the tracker. Exposed for tests. + self._waitpid_timed_out = False + def _reentrant_call_error(self): # gh-109629: this happens if an explicit call to the ResourceTracker # gets interrupted by a garbage collection, invoking a finalizer (*) @@ -91,16 +92,51 @@ class ResourceTracker(object): # making sure child processess are cleaned before ResourceTracker # gets destructed. # see https://github.com/python/cpython/issues/88887 - self._stop(use_blocking_lock=False) + # gh-146313: use a timeout to avoid deadlocking if a forked child + # still holds the pipe's write end open. + self._stop(use_blocking_lock=False, wait_timeout=1.0) - def _stop(self, use_blocking_lock=True): + def _after_fork_in_child(self): + # gh-146313: Called in the child right after os.fork(). + # + # The tracker process is a child of the *parent*, not of us, so we + # could never waitpid() it anyway. Clearing _pid means our __del__ + # becomes a no-op (the early return for _pid is None). + # + # Whether we keep the inherited _fd depends on who forked us: + # + # - multiprocessing.Process with the 'fork' start method sets + # _fork_intent.preserve_fd before forking. The child keeps the + # fd and reuses the parent's tracker (gh-80849). This is safe + # because multiprocessing's atexit handler joins all children + # before the parent's __del__ runs, so by then the fd copies + # are gone and the parent can reap the tracker promptly. + # + # - A raw os.fork() leaves the flag unset. We close the fd in the child after forking so + # the parent's __del__ can reap the tracker without waiting + # for the child to exit. If we later need a tracker, ensure_running() + # will launch a fresh one. + self._lock._at_fork_reinit() + self._reentrant_messages.clear() + self._pid = None + self._exitcode = None + if (self._fd is not None and + not getattr(_fork_intent, 'preserve_fd', False)): + fd = self._fd + self._fd = None + try: + os.close(fd) + except OSError: + pass + + def _stop(self, use_blocking_lock=True, wait_timeout=None): if use_blocking_lock: with self._lock: - self._stop_locked() + self._stop_locked(wait_timeout=wait_timeout) else: acquired = self._lock.acquire(blocking=False) try: - self._stop_locked() + self._stop_locked(wait_timeout=wait_timeout) finally: if acquired: self._lock.release() @@ -110,6 +146,10 @@ class ResourceTracker(object): close=os.close, waitpid=os.waitpid, waitstatus_to_exitcode=os.waitstatus_to_exitcode, + monotonic=time.monotonic, + sleep=time.sleep, + WNOHANG=getattr(os, 'WNOHANG', None), + wait_timeout=None, ): # This shouldn't happen (it might when called by a finalizer) # so we check for it anyway. @@ -126,7 +166,30 @@ class ResourceTracker(object): self._fd = None try: - _, status = waitpid(self._pid, 0) + if wait_timeout is None: + _, status = waitpid(self._pid, 0) + else: + # gh-146313: A forked child may still hold the pipe's write + # end open, preventing the tracker from seeing EOF and + # exiting. Poll with WNOHANG to avoid blocking forever. + deadline = monotonic() + wait_timeout + delay = 0.001 + while True: + result_pid, status = waitpid(self._pid, WNOHANG) + if result_pid != 0: + break + remaining = deadline - monotonic() + if remaining <= 0: + # The tracker is still running; it will be + # reparented to PID 1 (or the nearest subreaper) + # when we exit, and reaped there once all pipe + # holders release their fd. + self._pid = None + self._exitcode = None + self._waitpid_timed_out = True + return + delay = min(delay * 2, remaining, 0.1) + sleep(delay) except ChildProcessError: self._pid = None self._exitcode = None @@ -312,12 +375,24 @@ class ResourceTracker(object): self._ensure_running_and_write(msg) +# gh-146313: Per-thread flag set by .popen_fork.Popen._launch() just before +# os.fork(), telling _after_fork_in_child() to keep the inherited pipe fd so +# the child can reuse this tracker (gh-80849). Unset for raw os.fork() calls, +# where the child instead closes the fd so the parent's __del__ can reap the +# tracker. Using threading.local() keeps multiple threads calling +# popen_fork.Popen._launch() at once from clobbering eachothers intent. +_fork_intent = threading.local() + _resource_tracker = ResourceTracker() ensure_running = _resource_tracker.ensure_running register = _resource_tracker.register unregister = _resource_tracker.unregister getfd = _resource_tracker.getfd +# gh-146313: See _after_fork_in_child docstring. +if hasattr(os, 'register_at_fork'): + os.register_at_fork(after_in_child=_resource_tracker._after_fork_in_child) + def _decode_message(line): if line.startswith(b'{'): diff --git a/Lib/multiprocessing/shared_memory.py b/Lib/multiprocessing/shared_memory.py index 67e70fdc2..99a8ce332 100644 --- a/Lib/multiprocessing/shared_memory.py +++ b/Lib/multiprocessing/shared_memory.py @@ -539,6 +539,6 @@ class ShareableList: if value == entry: return position else: - raise ValueError(f"{value!r} not in this container") + raise ValueError("ShareableList.index(x): x not in list") __class_getitem__ = classmethod(types.GenericAlias) diff --git a/Lib/multiprocessing/spawn.py b/Lib/multiprocessing/spawn.py index daac1ecc3..d43864c93 100644 --- a/Lib/multiprocessing/spawn.py +++ b/Lib/multiprocessing/spawn.py @@ -184,7 +184,7 @@ def get_preparation_data(name): sys_argv=sys.argv, orig_dir=process.ORIGINAL_DIR, dir=os.getcwd(), - start_method=get_start_method(), + start_method=get_start_method(allow_none=True), ) # Figure out whether to initialise main in the subprocess as a module diff --git a/Lib/multiprocessing/synchronize.py b/Lib/multiprocessing/synchronize.py index 870c91349..9188114ae 100644 --- a/Lib/multiprocessing/synchronize.py +++ b/Lib/multiprocessing/synchronize.py @@ -21,22 +21,21 @@ from . import context from . import process from . import util -# Try to import the mp.synchronize module cleanly, if it fails -# raise ImportError for platforms lacking a working sem_open implementation. -# See issue 3770 +# TODO: Do any platforms still lack a functioning sem_open? try: from _multiprocessing import SemLock, sem_unlink -except (ImportError): +except ImportError: raise ImportError("This platform lacks a functioning sem_open" + - " implementation, therefore, the required" + - " synchronization primitives needed will not" + - " function, see issue 3770.") + " implementation. https://github.com/python/cpython/issues/48020.") # # Constants # -RECURSIVE_MUTEX, SEMAPHORE = list(range(2)) +# These match the enum in Modules/_multiprocessing/semaphore.c +RECURSIVE_MUTEX = 0 +SEMAPHORE = 1 + SEM_VALUE_MAX = _multiprocessing.SemLock.SEM_VALUE_MAX # @@ -91,6 +90,9 @@ class SemLock(object): self.acquire = self._semlock.acquire self.release = self._semlock.release + def locked(self): + return self._semlock._is_zero() + def __enter__(self): return self._semlock.__enter__() @@ -133,11 +135,16 @@ class Semaphore(SemLock): SemLock.__init__(self, SEMAPHORE, value, SEM_VALUE_MAX, ctx=ctx) def get_value(self): + '''Returns current value of Semaphore. + + Raises NotImplementedError on Mac OSX + because of broken sem_getvalue(). + ''' return self._semlock._get_value() def __repr__(self): try: - value = self._semlock._get_value() + value = self.get_value() except Exception: value = 'unknown' return '<%s(value=%s)>' % (self.__class__.__name__, value) @@ -153,7 +160,7 @@ class BoundedSemaphore(Semaphore): def __repr__(self): try: - value = self._semlock._get_value() + value = self.get_value() except Exception: value = 'unknown' return '<%s(value=%s, maxvalue=%s)>' % \ @@ -245,8 +252,8 @@ class Condition(object): def __repr__(self): try: - num_waiters = (self._sleeping_count._semlock._get_value() - - self._woken_count._semlock._get_value()) + num_waiters = (self._sleeping_count.get_value() - + self._woken_count.get_value()) except Exception: num_waiters = 'unknown' return '<%s(%s, %s)>' % (self.__class__.__name__, self._lock, num_waiters) diff --git a/Lib/multiprocessing/util.py b/Lib/multiprocessing/util.py index 4c8425064..549fb07c2 100644 --- a/Lib/multiprocessing/util.py +++ b/Lib/multiprocessing/util.py @@ -14,12 +14,12 @@ import weakref import atexit import threading # we want threading to install it's # cleanup function before multiprocessing does -from subprocess import _args_from_interpreter_flags +from subprocess import _args_from_interpreter_flags # noqa: F401 from . import process __all__ = [ - 'sub_debug', 'debug', 'info', 'sub_warning', 'get_logger', + 'sub_debug', 'debug', 'info', 'sub_warning', 'warn', 'get_logger', 'log_to_stderr', 'get_temp_dir', 'register_after_fork', 'is_exiting', 'Finalize', 'ForkAwareThreadLock', 'ForkAwareLocal', 'close_all_fds_except', 'SUBDEBUG', 'SUBWARNING', @@ -54,7 +54,7 @@ def info(msg, *args): if _logger: _logger.log(INFO, msg, *args, stacklevel=2) -def _warn(msg, *args): +def warn(msg, *args): if _logger: _logger.log(WARNING, msg, *args, stacklevel=2) @@ -196,14 +196,14 @@ def _get_base_temp_dir(tempfile): try: base_system_tempdir = tempfile._get_default_tempdir(dirlist) except FileNotFoundError: - _warn("Process-wide temporary directory %s will not be usable for " - "creating socket files and no usable system-wide temporary " - "directory was found in %s", base_tempdir, dirlist) + warn("Process-wide temporary directory %s will not be usable for " + "creating socket files and no usable system-wide temporary " + "directory was found in %s", base_tempdir, dirlist) # At this point, the system-wide temporary directory is not usable # but we may assume that the user-defined one is, even if we will # not be able to write socket files out there. return base_tempdir - _warn("Ignoring user-defined temporary directory: %s", base_tempdir) + warn("Ignoring user-defined temporary directory: %s", base_tempdir) # at most max(map(len, dirlist)) + 14 + 14 = 36 characters assert len(base_system_tempdir) + 14 + 14 < _SUN_PATH_MAX return base_system_tempdir diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index a009cb8ca..c3a980598 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -17,6 +17,7 @@ import errno import functools import signal import array +import collections.abc import socket import random import logging @@ -205,10 +206,38 @@ def only_run_in_spawn_testsuite(reason): return decorator +def only_run_in_forkserver_testsuite(reason): + """Returns a decorator: raises SkipTest unless fork is supported + and the current start method is forkserver. + + Combines @support.requires_fork() with the single-run semantics of + only_run_in_spawn_testsuite(), but uses the forkserver testsuite as + the single-run target. Appropriate for tests that exercise + os.fork() directly (raw fork or mp.set_start_method("fork") in a + subprocess) and don't vary by start method, since forkserver is + only available on platforms that support fork. + """ + + def decorator(test_item): + + @functools.wraps(test_item) + def forkserver_check_wrapper(*args, **kwargs): + if not support.has_fork_support: + raise unittest.SkipTest("requires working os.fork()") + if (start_method := multiprocessing.get_start_method()) != "forkserver": + raise unittest.SkipTest( + f"{start_method=}, not 'forkserver'; {reason}") + return test_item(*args, **kwargs) + + return forkserver_check_wrapper + + return decorator + + class TestInternalDecorators(unittest.TestCase): """Logic within a test suite that could errantly skip tests? Test it!""" - @unittest.skipIf(sys.platform == "win32", "test requires that fork exists.") + @support.requires_fork() def test_only_run_in_spawn_testsuite(self): if multiprocessing.get_start_method() != "spawn": raise unittest.SkipTest("only run in test_multiprocessing_spawn.") @@ -232,6 +261,30 @@ class TestInternalDecorators(unittest.TestCase): finally: multiprocessing.set_start_method(orig_start_method, force=True) + @support.requires_fork() + def test_only_run_in_forkserver_testsuite(self): + if multiprocessing.get_start_method() != "forkserver": + raise unittest.SkipTest("only run in test_multiprocessing_forkserver.") + + try: + @only_run_in_forkserver_testsuite("testing this decorator") + def return_four_if_forkserver(): + return 4 + except Exception as err: + self.fail(f"expected decorated `def` not to raise; caught {err}") + + orig_start_method = multiprocessing.get_start_method(allow_none=True) + try: + multiprocessing.set_start_method("forkserver", force=True) + self.assertEqual(return_four_if_forkserver(), 4) + multiprocessing.set_start_method("spawn", force=True) + with self.assertRaises(unittest.SkipTest) as ctx: + return_four_if_forkserver() + self.assertIn("testing this decorator", str(ctx.exception)) + self.assertIn("start_method=", str(ctx.exception)) + finally: + multiprocessing.set_start_method(orig_start_method, force=True) + # # Creates a wrapper for a function which records the time it takes to finish @@ -318,7 +371,7 @@ class _TestProcess(BaseTestCase): authkey = current.authkey self.assertTrue(current.is_alive()) - self.assertTrue(not current.daemon) + self.assertFalse(current.daemon) self.assertIsInstance(authkey, bytes) self.assertTrue(len(authkey) > 0) self.assertEqual(current.ident, os.getpid()) @@ -462,7 +515,7 @@ class _TestProcess(BaseTestCase): self.assertEqual(p.is_alive(), False) self.assertEqual(p.daemon, True) self.assertNotIn(p, self.active_children()) - self.assertTrue(type(self.active_children()) is list) + self.assertIs(type(self.active_children()), list) self.assertEqual(p.exitcode, None) p.start() @@ -516,16 +569,23 @@ class _TestProcess(BaseTestCase): event.set() time.sleep(100) + @classmethod + def _sleep_no_int_handler(cls, event): + signal.signal(signal.SIGINT, signal.SIG_DFL) + cls._sleep_some_event(event) + @classmethod def _test_sleep(cls, delay): time.sleep(delay) - def _kill_process(self, meth): + def _kill_process(self, meth, target=None): if self.TYPE == 'threads': self.skipTest('test not appropriate for {}'.format(self.TYPE)) event = self.Event() - p = self.Process(target=self._sleep_some_event, args=(event,)) + if not target: + target = self._sleep_some_event + p = self.Process(target=target, args=(event,)) p.daemon = True p.start() @@ -575,6 +635,19 @@ class _TestProcess(BaseTestCase): return p.exitcode + @unittest.skipIf(os.name == 'nt', "POSIX only") + def test_interrupt(self): + exitcode = self._kill_process(multiprocessing.Process.interrupt) + self.assertEqual(exitcode, 1) + # exit code 1 is hard-coded for uncaught exceptions + # (KeyboardInterrupt in this case) + # in multiprocessing.BaseProcess._bootstrap + + @unittest.skipIf(os.name == 'nt', "POSIX only") + def test_interrupt_no_handler(self): + exitcode = self._kill_process(multiprocessing.Process.interrupt, target=self._sleep_no_int_handler) + self.assertEqual(exitcode, -signal.SIGINT) + def test_terminate(self): exitcode = self._kill_process(multiprocessing.Process.terminate) self.assertEqual(exitcode, -signal.SIGTERM) @@ -591,8 +664,8 @@ class _TestProcess(BaseTestCase): cpus = multiprocessing.cpu_count() except NotImplementedError: cpus = 1 - self.assertTrue(type(cpus) is int) - self.assertTrue(cpus >= 1) + self.assertIsInstance(cpus, int) + self.assertGreaterEqual(cpus, 1) def test_active_children(self): self.assertEqual(type(self.active_children()), list) @@ -858,8 +931,8 @@ class _TestProcess(BaseTestCase): finally: setattr(sys, stream_name, old_stream) - @classmethod - def _sleep_and_set_event(self, evt, delay=0.0): + @staticmethod + def _sleep_and_set_event(evt, delay=0.0): time.sleep(delay) evt.set() @@ -910,6 +983,56 @@ class _TestProcess(BaseTestCase): if os.name != 'nt': self.check_forkserver_death(signal.SIGKILL) + def test_forkserver_auth_is_enabled(self): + if self.TYPE == "threads": + self.skipTest(f"test not appropriate for {self.TYPE}") + if multiprocessing.get_start_method() != "forkserver": + self.skipTest("forkserver start method specific") + + forkserver = multiprocessing.forkserver._forkserver + forkserver.ensure_running() + self.assertTrue(forkserver._forkserver_pid) + authkey = forkserver._forkserver_authkey + self.assertTrue(authkey) + self.assertGreater(len(authkey), 15) + addr = forkserver._forkserver_address + self.assertTrue(addr) + + # Demonstrate that a raw auth handshake, as Client performs, does not + # raise an error. + client = multiprocessing.connection.Client(addr, authkey=authkey) + client.close() + + # That worked, now launch a quick process. + proc = self.Process(target=sys.exit) + proc.start() + proc.join() + self.assertEqual(proc.exitcode, 0) + + def test_forkserver_without_auth_fails(self): + if self.TYPE == "threads": + self.skipTest(f"test not appropriate for {self.TYPE}") + if multiprocessing.get_start_method() != "forkserver": + self.skipTest("forkserver start method specific") + + forkserver = multiprocessing.forkserver._forkserver + forkserver.ensure_running() + self.assertTrue(forkserver._forkserver_pid) + authkey_len = len(forkserver._forkserver_authkey) + with unittest.mock.patch.object( + forkserver, '_forkserver_authkey', None): + # With an incorrect authkey we should get an auth rejection + # rather than the above protocol error. + forkserver._forkserver_authkey = b'T' * authkey_len + proc = self.Process(target=sys.exit) + with self.assertRaises(multiprocessing.AuthenticationError): + proc.start() + del proc + + # authkey restored, launching processes should work again. + proc = self.Process(target=sys.exit) + proc.start() + proc.join() # # @@ -1445,10 +1568,34 @@ class _TestLock(BaseTestCase): def test_lock(self): lock = self.Lock() self.assertEqual(lock.acquire(), True) + self.assertTrue(lock.locked()) self.assertEqual(lock.acquire(False), False) self.assertEqual(lock.release(), None) + self.assertFalse(lock.locked()) self.assertRaises((ValueError, threading.ThreadError), lock.release) + @classmethod + def _test_lock_locked_2processes(cls, lock, event, res): + lock.acquire() + res.value = lock.locked() + event.set() + + @unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes') + def test_lock_locked_2processes(self): + if self.TYPE != 'processes': + self.skipTest('test not appropriate for {}'.format(self.TYPE)) + + lock = self.Lock() + event = self.Event() + res = self.Value('b', 0) + p = self.Process(target=self._test_lock_locked_2processes, + args=(lock, event, res)) + p.start() + event.wait() + self.assertTrue(lock.locked()) + self.assertTrue(res.value) + p.join() + @staticmethod def _acquire_release(lock, timeout, l=None, n=1): for _ in range(n): @@ -1485,14 +1632,11 @@ class _TestLock(BaseTestCase): for i in range(n): self.assertIn(f'', l) - - t = threading.Thread(target=self._acquire_release, - args=(lock, 0.2), - name=f'T1') + rlock = self.RLock() + t = threading.Thread(target=rlock.acquire) t.start() - time.sleep(0.1) - self.assertEqual('', repr(lock)) - time.sleep(0.2) + t.join() + self.assertEqual('', repr(rlock)) pname = 'P1' l = multiprocessing.Manager().list() @@ -1503,28 +1647,49 @@ class _TestLock(BaseTestCase): p.join() self.assertEqual(f'', l[0]) - event = self.Event() - lock = self.RLock() - p = self.Process(target=self._acquire_event, - args=(lock, event)) + rlock = self.RLock() + p = self.Process(target=self._acquire, args=(rlock,)) p.start() - event.wait() - self.assertEqual('', repr(lock)) p.join() + self.assertEqual('', repr(rlock)) def test_rlock(self): lock = self.RLock() self.assertEqual(lock.acquire(), True) + self.assertTrue(lock.locked()) self.assertEqual(lock.acquire(), True) self.assertEqual(lock.acquire(), True) self.assertEqual(lock.release(), None) + self.assertTrue(lock.locked()) self.assertEqual(lock.release(), None) self.assertEqual(lock.release(), None) + self.assertFalse(lock.locked()) self.assertRaises((AssertionError, RuntimeError), lock.release) + @unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes') + def test_rlock_locked_2processes(self): + if self.TYPE != 'processes': + self.skipTest('test not appropriate for {}'.format(self.TYPE)) + + rlock = self.RLock() + event = self.Event() + res = self.Value('b', 0) + # target is the same as for the test_lock_locked_2processes test. + p = self.Process(target=self._test_lock_locked_2processes, + args=(rlock, event, res)) + p.start() + event.wait() + self.assertTrue(rlock.locked()) + self.assertTrue(res.value) + p.join() + def test_lock_context(self): - with self.Lock(): - pass + with self.Lock() as locked: + self.assertTrue(locked) + + def test_rlock_context(self): + with self.RLock() as locked: + self.assertTrue(locked) class _TestSemaphore(BaseTestCase): @@ -2354,14 +2519,14 @@ class _TestValue(BaseTestCase): self.assertEqual(lock, lock3) arr4 = self.Value('i', 5, lock=False) - self.assertFalse(hasattr(arr4, 'get_lock')) - self.assertFalse(hasattr(arr4, 'get_obj')) + self.assertNotHasAttr(arr4, 'get_lock') + self.assertNotHasAttr(arr4, 'get_obj') self.assertRaises(AttributeError, self.Value, 'i', 5, lock='navalue') arr5 = self.RawValue('i', 5) - self.assertFalse(hasattr(arr5, 'get_lock')) - self.assertFalse(hasattr(arr5, 'get_obj')) + self.assertNotHasAttr(arr5, 'get_lock') + self.assertNotHasAttr(arr5, 'get_obj') class _TestArray(BaseTestCase): @@ -2434,14 +2599,14 @@ class _TestArray(BaseTestCase): self.assertEqual(lock, lock3) arr4 = self.Array('i', range(10), lock=False) - self.assertFalse(hasattr(arr4, 'get_lock')) - self.assertFalse(hasattr(arr4, 'get_obj')) + self.assertNotHasAttr(arr4, 'get_lock') + self.assertNotHasAttr(arr4, 'get_obj') self.assertRaises(AttributeError, self.Array, 'i', range(10), lock='notalock') arr5 = self.RawArray('i', range(10)) - self.assertFalse(hasattr(arr5, 'get_lock')) - self.assertFalse(hasattr(arr5, 'get_obj')) + self.assertNotHasAttr(arr5, 'get_lock') + self.assertNotHasAttr(arr5, 'get_obj') # # @@ -2482,6 +2647,23 @@ class _TestContainers(BaseTestCase): a.append('hello') self.assertEqual(f[0][:], [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'hello']) + def test_list_isinstance(self): + a = self.list() + self.assertIsInstance(a, collections.abc.MutableSequence) + + # MutableSequence also has __iter__, but we can iterate over + # ListProxy using __getitem__ instead. Adding __iter__ to ListProxy + # would change the behavior of a list modified during iteration. + mutable_sequence_methods = ( + '__contains__', '__delitem__', '__getitem__', '__iadd__', + '__len__', '__reversed__', '__setitem__', 'append', + 'clear', 'count', 'extend', 'index', 'insert', 'pop', 'remove', + 'reverse', + ) + for name in mutable_sequence_methods: + with self.subTest(name=name): + self.assertTrue(callable(getattr(a, name))) + def test_list_iter(self): a = self.list(list(range(10))) it = iter(a) @@ -2522,6 +2704,19 @@ class _TestContainers(BaseTestCase): self.assertEqual(sorted(d.values()), [chr(i) for i in indices]) self.assertEqual(sorted(d.items()), [(i, chr(i)) for i in indices]) + def test_dict_isinstance(self): + a = self.dict() + self.assertIsInstance(a, collections.abc.MutableMapping) + + mutable_mapping_methods = ( + '__contains__', '__delitem__', '__eq__', '__getitem__', '__iter__', + '__len__', '__ne__', '__setitem__', 'clear', 'get', 'items', + 'keys', 'pop', 'popitem', 'setdefault', 'update', 'values', + ) + for name in mutable_mapping_methods: + with self.subTest(name=name): + self.assertTrue(callable(getattr(a, name))) + def test_dict_iter(self): d = self.dict() indices = list(range(65, 70)) @@ -2599,8 +2794,8 @@ class _TestContainers(BaseTestCase): self.assertEqual((n.name, n.job), ('Bob', 'Builder')) del n.job self.assertEqual(str(n), "Namespace(name='Bob')") - self.assertTrue(hasattr(n, 'name')) - self.assertTrue(not hasattr(n, 'job')) + self.assertHasAttr(n, 'name') + self.assertNotHasAttr(n, 'job') # # @@ -2972,8 +3167,7 @@ class _TestPool(BaseTestCase): # check that we indeed waited for all jobs self.assertGreater(time.monotonic() - t_start, 0.9) - # TODO: RUSTPYTHON - reference counting differences - @unittest.skip("TODO: RUSTPYTHON") + @unittest.skip("TODO: RUSTPYTHON; reference counting differences") def test_release_task_refs(self): # Issue #29861: task arguments and results should not be kept # alive after we are done with them. @@ -3172,6 +3366,7 @@ class _TestMyManager(BaseTestCase): ALLOWED_TYPES = ('manager',) + @support.skip_if_sanitizer('TSan: leaks threads', thread=True) def test_mymanager(self): manager = MyManager(shutdown_timeout=SHUTDOWN_TIMEOUT) manager.start() @@ -3183,6 +3378,7 @@ class _TestMyManager(BaseTestCase): # which happens on slow buildbots. self.assertIn(manager._process.exitcode, (0, -signal.SIGTERM)) + @support.skip_if_sanitizer('TSan: leaks threads', thread=True) def test_mymanager_context(self): manager = MyManager(shutdown_timeout=SHUTDOWN_TIMEOUT) with manager: @@ -3192,6 +3388,7 @@ class _TestMyManager(BaseTestCase): # which happens on slow buildbots. self.assertIn(manager._process.exitcode, (0, -signal.SIGTERM)) + @support.skip_if_sanitizer('TSan: leaks threads', thread=True) def test_mymanager_context_prestarted(self): manager = MyManager(shutdown_timeout=SHUTDOWN_TIMEOUT) manager.start() @@ -3262,6 +3459,7 @@ class _TestRemoteManager(BaseTestCase): # Note that xmlrpclib will deserialize object as a list not a tuple queue.put(tuple(cls.values)) + @support.skip_if_sanitizer('TSan: leaks threads', thread=True) def test_remote(self): authkey = os.urandom(32) @@ -3303,6 +3501,7 @@ class _TestManagerRestart(BaseTestCase): queue = manager.get_queue() queue.put('hello world') + @support.skip_if_sanitizer("TSan: leaks threads", thread=True) def test_rapid_restart(self): authkey = os.urandom(32) manager = QueueManager( @@ -3886,8 +4085,7 @@ class _TestPicklingConnections(BaseTestCase): conn.close() - # TODO: RUSTPYTHON - hangs - @unittest.skip("TODO: RUSTPYTHON") + @unittest.skip("TODO: RUSTPYTHON; hangs") def test_pickling(self): families = self.connection.families @@ -4109,8 +4307,7 @@ class _TestSharedCTypes(BaseTestCase): for i in range(len(arr)): arr[i] *= 2 - # TODO: RUSTPYTHON - ctypes Structure shared memory not working - @unittest.expectedFailure + @unittest.expectedFailure # TODO: RUSTPYTHON; - ctypes Structure shared memory not working def test_sharedctypes(self, lock=False): x = Value('i', 7, lock=lock) y = Value(c_double, 1.0/3.0, lock=lock) @@ -4134,8 +4331,7 @@ class _TestSharedCTypes(BaseTestCase): self.assertAlmostEqual(arr[i], i*2) self.assertEqual(string.value, latin('hellohello')) - # TODO: RUSTPYTHON - calls test_sharedctypes which fails - @unittest.expectedFailure + @unittest.expectedFailure # TODO: RUSTPYTHON; - calls test_sharedctypes which fails def test_synchronize(self): self.test_sharedctypes(lock=True) @@ -4813,8 +5009,7 @@ class _TestFinalize(BaseTestCase): result = [obj for obj in iter(conn.recv, 'STOP')] self.assertEqual(result, ['a', 'b', 'd10', 'd03', 'd02', 'd01', 'e']) - # TODO: RUSTPYTHON; SIGSEGV due to dict thread-safety issue under aggressive GC - @unittest.skip("TODO: RUSTPYTHON") + @unittest.skip("TODO: RUSTPYTHON; SIGSEGV due to dict thread-safety issue under aggressive GC") @support.requires_resource('cpu') def test_thread_safety(self): # bpo-24484: _run_finalizers() should be thread-safe @@ -4906,13 +5101,9 @@ class _TestImportStar(unittest.TestCase): for name in modules: __import__(name) mod = sys.modules[name] - self.assertTrue(hasattr(mod, '__all__'), name) - + self.assertHasAttr(mod, '__all__', name) for attr in mod.__all__: - self.assertTrue( - hasattr(mod, attr), - '%r does not have attribute %r' % (mod, attr) - ) + self.assertHasAttr(mod, attr) # # Quick test that logging works -- does not test logging output @@ -4925,7 +5116,7 @@ class _TestLogging(BaseTestCase): def test_enable_logging(self): logger = multiprocessing.get_logger() logger.setLevel(util.SUBWARNING) - self.assertTrue(logger is not None) + self.assertIsNotNone(logger) logger.debug('this will not be printed') logger.info('nor will this') logger.setLevel(LOG_LEVEL) @@ -5062,6 +5253,23 @@ class TestInvalidHandle(unittest.TestCase): multiprocessing.connection.Connection, -1) +# +# Regression tests for BaseProcess kwargs handling +# + +class TestBaseProcessKwargs(unittest.TestCase): + def test_default_kwargs_not_shared_between_instances(self): + # Creating multiple Process instances without passing kwargs + # must create independent empty dicts (no shared state). + p1 = multiprocessing.Process(target=lambda: None) + p2 = multiprocessing.Process(target=lambda: None) + self.assertIsInstance(p1._kwargs, dict) + self.assertIsInstance(p2._kwargs, dict) + self.assertIsNot(p1._kwargs, p2._kwargs) + # Mutating one should not affect the other + p1._kwargs['x'] = 1 + self.assertNotIn('x', p2._kwargs) + @hashlib_helper.requires_hashdigest('sha256') class OtherTest(unittest.TestCase): @@ -5700,6 +5908,39 @@ class TestStartMethod(unittest.TestCase): self.assertRaises(ValueError, ctx.set_start_method, None) self.check_context(ctx) + @staticmethod + def _dummy_func(): + pass + + def test_spawn_dont_set_context(self): + # Run a process with spawn or forkserver context may change + # the global start method, see gh-109263. + for method in ('fork', 'spawn', 'forkserver'): + multiprocessing.set_start_method(None, force=True) + + try: + ctx = multiprocessing.get_context(method) + except ValueError: + continue + process = ctx.Process(target=self._dummy_func) + process.start() + process.join() + self.assertIsNone(multiprocessing.get_start_method(allow_none=True)) + + @only_run_in_spawn_testsuite("freeze_support is not start method specific") + def test_freeze_support_dont_set_context(self): + # gh-140814: freeze_support() should not set the start method + # as a side effect, so a later set_start_method() still works. + multiprocessing.set_start_method(None, force=True) + try: + multiprocessing.freeze_support() + self.assertIsNone( + multiprocessing.get_start_method(allow_none=True)) + # Should not raise "context has already been set" + multiprocessing.set_start_method('spawn') + finally: + multiprocessing.set_start_method(None, force=True) + def test_context_check_module_types(self): try: ctx = multiprocessing.get_context('forkserver') @@ -5721,24 +5962,37 @@ class TestStartMethod(unittest.TestCase): self.assertEqual(multiprocessing.get_start_method(), method) ctx = multiprocessing.get_context() self.assertEqual(ctx.get_start_method(), method) - self.assertTrue(type(ctx).__name__.lower().startswith(method)) - self.assertTrue( - ctx.Process.__name__.lower().startswith(method)) + self.assertStartsWith(type(ctx).__name__.lower(), method) + self.assertStartsWith(ctx.Process.__name__.lower(), method) self.check_context(multiprocessing) count += 1 finally: multiprocessing.set_start_method(old_method, force=True) self.assertGreaterEqual(count, 1) - def test_get_all(self): + def test_get_all_start_methods(self): methods = multiprocessing.get_all_start_methods() + self.assertIn('spawn', methods) if sys.platform == 'win32': self.assertEqual(methods, ['spawn']) + elif sys.platform == 'darwin': + self.assertEqual(methods[0], 'spawn') # The default is first. + # Whether these work or not, they remain available on macOS. + self.assertIn('fork', methods) + self.assertIn('forkserver', methods) else: - self.assertTrue(methods == ['fork', 'spawn'] or - methods == ['spawn', 'fork'] or - methods == ['fork', 'spawn', 'forkserver'] or - methods == ['spawn', 'fork', 'forkserver']) + # POSIX + self.assertIn('fork', methods) + if other_methods := set(methods) - {'fork', 'spawn'}: + # If there are more than those two, forkserver must be one. + self.assertEqual({'forkserver'}, other_methods) + # The default is the first method in the list. + self.assertIn(methods[0], {'forkserver', 'spawn'}, + msg='3.14+ default must not be fork') + if methods[0] == 'spawn': + # Confirm that the current default selection logic prefers + # forkserver vs spawn when available. + self.assertNotIn('forkserver', methods) def test_preload_resources(self): if multiprocessing.get_start_method() != 'forkserver': @@ -5910,9 +6164,9 @@ class TestResourceTracker(unittest.TestCase): if should_die: self.assertEqual(len(all_warn), 1) the_warn = all_warn[0] - self.assertTrue(issubclass(the_warn.category, UserWarning)) - self.assertTrue("resource_tracker: process died" - in str(the_warn.message)) + self.assertIsSubclass(the_warn.category, UserWarning) + self.assertIn("resource_tracker: process died", + str(the_warn.message)) else: self.assertEqual(len(all_warn), 0) @@ -5934,8 +6188,9 @@ class TestResourceTracker(unittest.TestCase): def _is_resource_tracker_reused(conn, pid): from multiprocessing.resource_tracker import _resource_tracker _resource_tracker.ensure_running() - # The pid should be None in the child process, expect for the fork - # context. It should not be a new value. + # The pid should be None in the child (the at-fork handler clears + # it for fork; spawn/forkserver children never had it set). It + # should not be a new value. reused = _resource_tracker._pid in (None, pid) reused &= _resource_tracker._check_alive() conn.send(reused) @@ -6020,6 +6275,184 @@ class TestResourceTracker(unittest.TestCase): # restore sigmask to what it was before executing test signal.pthread_sigmask(signal.SIG_SETMASK, orig_sigmask) + @unittest.expectedFailureIf(not hasattr(os, "killpg"), "TODO: RUSTPYTHON") + @only_run_in_forkserver_testsuite("avoids redundant testing.") + def test_resource_tracker_fork_deadlock(self): + # gh-146313: ResourceTracker.__del__ used to deadlock if a forked + # child still held the pipe's write end open when the parent + # exited, because the parent would block in waitpid() waiting for + # the tracker to exit, but the tracker would never see EOF. + cmd = '''if 1: + import os, signal + from multiprocessing.resource_tracker import ensure_running + ensure_running() + if os.fork() == 0: + signal.pause() + os._exit(0) + # parent falls through and exits, triggering __del__ + ''' + proc = subprocess.Popen([sys.executable, '-c', cmd], + start_new_session=True) + try: + try: + proc.wait(timeout=support.SHORT_TIMEOUT) + except subprocess.TimeoutExpired: + self.fail( + "Parent process deadlocked in ResourceTracker.__del__" + ) + self.assertEqual(proc.returncode, 0) + finally: + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + proc.wait() + + @only_run_in_forkserver_testsuite("avoids redundant testing.") + def test_resource_tracker_mp_fork_reuse_and_prompt_reap(self): + # gh-146313 / gh-80849: A child started via multiprocessing.Process + # with the 'fork' start method should reuse the parent's resource + # tracker (the at-fork handler preserves the inherited pipe fd), + # *and* the parent should be able to reap the tracker promptly + # after joining the child, without hitting the waitpid timeout. + cmd = textwrap.dedent(''' + import multiprocessing as mp + from multiprocessing.resource_tracker import _resource_tracker + + def child(conn): + # Prove we can talk to the parent's tracker by registering + # and unregistering a dummy resource over the inherited fd. + # If the fd were closed, ensure_running would launch a new + # tracker and _pid would be non-None. + _resource_tracker.register("x", "dummy") + _resource_tracker.unregister("x", "dummy") + conn.send((_resource_tracker._fd is not None, + _resource_tracker._pid is None, + _resource_tracker._check_alive())) + + if __name__ == "__main__": + mp.set_start_method("fork") + _resource_tracker.ensure_running() + r, w = mp.Pipe(duplex=False) + p = mp.Process(target=child, args=(w,)) + p.start() + child_has_fd, child_pid_none, child_alive = r.recv() + p.join() + w.close(); r.close() + + # Now simulate __del__: the child has exited and released + # its fd copy, so the tracker should see EOF and exit + # promptly -- no timeout. + _resource_tracker._stop(wait_timeout=5.0) + print(child_has_fd, child_pid_none, child_alive, + _resource_tracker._waitpid_timed_out, + _resource_tracker._exitcode) + ''') + rc, out, err = script_helper.assert_python_ok('-c', cmd) + parts = out.decode().split() + self.assertEqual(parts, ['True', 'True', 'True', 'False', '0'], + f"unexpected: {parts!r} stderr={err!r}") + + @only_run_in_forkserver_testsuite("avoids redundant testing.") + def test_resource_tracker_raw_fork_prompt_reap(self): + # gh-146313: After a raw os.fork() the at-fork handler closes the + # child's inherited fd, so the parent can reap the tracker + # immediately -- even while the child is still alive -- rather + # than waiting out the 1s timeout. + cmd = textwrap.dedent(''' + import os, signal + from multiprocessing.resource_tracker import _resource_tracker + + _resource_tracker.ensure_running() + r, w = os.pipe() + pid = os.fork() + if pid == 0: + os.close(r) + # Report whether our fd was closed by the at-fork handler. + os.write(w, b"1" if _resource_tracker._fd is None else b"0") + os.close(w) + signal.pause() # stay alive so parent's reap is meaningful + os._exit(0) + os.close(w) + child_fd_closed = os.read(r, 1) == b"1" + os.close(r) + + # Child is still alive and paused. Because it closed its fd + # copy, our close below is the last one and the tracker exits. + _resource_tracker._stop(wait_timeout=5.0) + + os.kill(pid, signal.SIGKILL) + os.waitpid(pid, 0) + print(child_fd_closed, + _resource_tracker._waitpid_timed_out, + _resource_tracker._exitcode) + ''') + rc, out, err = script_helper.assert_python_ok('-c', cmd) + parts = out.decode().split() + self.assertEqual(parts, ['True', 'False', '0'], + f"unexpected: {parts!r} stderr={err!r}") + + @only_run_in_forkserver_testsuite("avoids redundant testing.") + def test_resource_tracker_lock_reinit_after_fork(self): + # gh-146313: If a parent thread held the tracker's lock at fork + # time, the child would inherit the held lock and deadlock on + # its next ensure_running(). The at-fork handler reinits it. + cmd = textwrap.dedent(''' + import os, threading + from multiprocessing.resource_tracker import _resource_tracker + + held = threading.Event() + release = threading.Event() + def hold(): + with _resource_tracker._lock: + held.set() + release.wait() + t = threading.Thread(target=hold) + t.start() + held.wait() + + pid = os.fork() + if pid == 0: + ok = _resource_tracker._lock.acquire(timeout=5.0) + os._exit(0 if ok else 1) + + release.set() + t.join() + _, status = os.waitpid(pid, 0) + print(os.waitstatus_to_exitcode(status)) + ''') + rc, out, err = script_helper.assert_python_ok( + '-W', 'ignore::DeprecationWarning', '-c', cmd) + self.assertEqual(out.strip(), b'0', + f"child failed to acquire lock: stderr={err!r}") + + @only_run_in_forkserver_testsuite("avoids redundant testing.") + def test_resource_tracker_safety_net_timeout(self): + # gh-146313: When an mp.Process(fork) child holds the preserved + # fd and the parent calls _stop() without joining (simulating + # abnormal shutdown), the safety-net timeout should fire rather + # than deadlocking. + cmd = textwrap.dedent(''' + import multiprocessing as mp + import signal + from multiprocessing.resource_tracker import _resource_tracker + + if __name__ == "__main__": + mp.set_start_method("fork") + _resource_tracker.ensure_running() + p = mp.Process(target=signal.pause) + p.start() + # Stop WITHOUT joining -- child still holds preserved fd + _resource_tracker._stop(wait_timeout=0.5) + print(_resource_tracker._waitpid_timed_out) + p.terminate() + p.join() + ''') + rc, out, err = script_helper.assert_python_ok('-c', cmd) + self.assertEqual(out.strip(), b'True', + f"safety-net timeout did not fire: stderr={err!r}") + + class TestSimpleQueue(unittest.TestCase): @classmethod @@ -6117,8 +6550,8 @@ class TestPoolNotLeakOnFailure(unittest.TestCase): Process=FailingForkProcess)) p.close() p.join() - self.assertFalse( - any(process.is_alive() for process in forked_processes)) + for process in forked_processes: + self.assertFalse(process.is_alive(), process) @hashlib_helper.requires_hashdigest('sha256') @@ -6207,6 +6640,7 @@ class TestSyncManagerTypes(unittest.TestCase): @classmethod def _test_lock(cls, obj): obj.acquire() + obj.locked() def test_lock(self, lname="Lock"): o = getattr(self.manager, lname)() @@ -6218,8 +6652,9 @@ class TestSyncManagerTypes(unittest.TestCase): def _test_rlock(cls, obj): obj.acquire() obj.release() + obj.locked() - def test_rlock(self, lname="Lock"): + def test_rlock(self, lname="RLock"): o = getattr(self.manager, lname)() self.run_worker(self._test_rlock, o) @@ -6290,12 +6725,30 @@ class TestSyncManagerTypes(unittest.TestCase): case.assertEqual(obj[0], 5) case.assertEqual(obj.count(5), 1) case.assertEqual(obj.index(5), 0) + obj += [7] + case.assertIsInstance(obj, multiprocessing.managers.ListProxy) + case.assertListEqual(list(obj), [5, 7]) + obj *= 2 + case.assertIsInstance(obj, multiprocessing.managers.ListProxy) + case.assertListEqual(list(obj), [5, 7, 5, 7]) + double_obj = obj * 2 + case.assertIsInstance(double_obj, list) + case.assertListEqual(list(double_obj), [5, 7, 5, 7, 5, 7, 5, 7]) + double_obj = 2 * obj + case.assertIsInstance(double_obj, list) + case.assertListEqual(list(double_obj), [5, 7, 5, 7, 5, 7, 5, 7]) + copied_obj = obj.copy() + case.assertIsInstance(copied_obj, list) + case.assertListEqual(list(copied_obj), [5, 7, 5, 7]) + obj.extend(double_obj + copied_obj) obj.sort() obj.reverse() for x in obj: pass - case.assertEqual(len(obj), 1) - case.assertEqual(obj.pop(0), 5) + case.assertEqual(len(obj), 16) + case.assertEqual(obj.pop(0), 7) + obj.clear() + case.assertEqual(len(obj), 0) def test_list(self): o = self.manager.list() @@ -6314,8 +6767,31 @@ class TestSyncManagerTypes(unittest.TestCase): case.assertListEqual(list(obj.keys()), ['foo']) case.assertListEqual(list(obj.values()), [5]) case.assertDictEqual(obj.copy(), {'foo': 5}) - case.assertTupleEqual(obj.popitem(), ('foo', 5)) + obj |= {'bar': 6} + case.assertIsInstance(obj, multiprocessing.managers.DictProxy) + case.assertDictEqual(dict(obj), {'foo': 5, 'bar': 6}) + x = reversed(obj) + case.assertIsInstance(x, type(iter([]))) + case.assertListEqual(list(x), ['bar', 'foo']) + x = {'bar': 7, 'baz': 7} | obj + case.assertIsInstance(x, dict) + case.assertDictEqual(dict(x), {'foo': 5, 'bar': 6, 'baz': 7}) + x = obj | {'bar': 7, 'baz': 7} + case.assertIsInstance(x, dict) + case.assertDictEqual(dict(x), {'foo': 5, 'bar': 7, 'baz': 7}) + x = obj.fromkeys(['bar'], 6) + case.assertIsInstance(x, dict) + case.assertDictEqual(x, {'bar': 6}) + x = obj.popitem() + case.assertIsInstance(x, tuple) + case.assertTupleEqual(x, ('bar', 6)) + obj.setdefault('bar', 0) + obj.update({'bar': 7}) + case.assertEqual(obj.pop('bar'), 7) + obj.clear() + case.assertEqual(len(obj), 0) + @unittest.expectedFailure # TODO: RUSTPYTHON def test_dict(self): o = self.manager.dict() o['foo'] = 5 @@ -6360,6 +6836,150 @@ class TestSyncManagerTypes(unittest.TestCase): o.y = 1 self.run_worker(self._test_namespace, o) + @classmethod + def _test_set_operator_symbols(cls, obj): + case = unittest.TestCase() + obj.update(['a', 'b', 'c']) + case.assertEqual(len(obj), 3) + case.assertIn('a', obj) + case.assertNotIn('d', obj) + result = obj | {'d', 'e'} + case.assertSetEqual(result, {'a', 'b', 'c', 'd', 'e'}) + result = {'d', 'e'} | obj + case.assertSetEqual(result, {'a', 'b', 'c', 'd', 'e'}) + obj |= {'d', 'e'} + case.assertSetEqual(obj, {'a', 'b', 'c', 'd', 'e'}) + case.assertIsInstance(obj, multiprocessing.managers.SetProxy) + + obj.clear() + obj.update(['a', 'b', 'c']) + result = {'a', 'b', 'd'} - obj + case.assertSetEqual(result, {'d'}) + result = obj - {'a', 'b'} + case.assertSetEqual(result, {'c'}) + obj -= {'a', 'b'} + case.assertSetEqual(obj, {'c'}) + case.assertIsInstance(obj, multiprocessing.managers.SetProxy) + + obj.clear() + obj.update(['a', 'b', 'c']) + result = {'b', 'c', 'd'} ^ obj + case.assertSetEqual(result, {'a', 'd'}) + result = obj ^ {'b', 'c', 'd'} + case.assertSetEqual(result, {'a', 'd'}) + obj ^= {'b', 'c', 'd'} + case.assertSetEqual(obj, {'a', 'd'}) + case.assertIsInstance(obj, multiprocessing.managers.SetProxy) + + obj.clear() + obj.update(['a', 'b', 'c']) + result = obj & {'b', 'c', 'd'} + case.assertSetEqual(result, {'b', 'c'}) + result = {'b', 'c', 'd'} & obj + case.assertSetEqual(result, {'b', 'c'}) + obj &= {'b', 'c', 'd'} + case.assertSetEqual(obj, {'b', 'c'}) + case.assertIsInstance(obj, multiprocessing.managers.SetProxy) + + obj.clear() + obj.update(['a', 'b', 'c']) + case.assertSetEqual(set(obj), {'a', 'b', 'c'}) + + @classmethod + def _test_set_operator_methods(cls, obj): + case = unittest.TestCase() + obj.add('d') + case.assertIn('d', obj) + + obj.clear() + obj.update(['a', 'b', 'c']) + copy_obj = obj.copy() + case.assertSetEqual(copy_obj, obj) + obj.remove('a') + case.assertNotIn('a', obj) + case.assertRaises(KeyError, obj.remove, 'a') + + obj.clear() + obj.update(['a']) + obj.discard('a') + case.assertNotIn('a', obj) + obj.discard('a') + case.assertNotIn('a', obj) + obj.update(['a']) + popped = obj.pop() + case.assertNotIn(popped, obj) + + obj.clear() + obj.update(['a', 'b', 'c']) + result = obj.intersection({'b', 'c', 'd'}) + case.assertSetEqual(result, {'b', 'c'}) + obj.intersection_update({'b', 'c', 'd'}) + case.assertSetEqual(obj, {'b', 'c'}) + + obj.clear() + obj.update(['a', 'b', 'c']) + result = obj.difference({'a', 'b'}) + case.assertSetEqual(result, {'c'}) + obj.difference_update({'a', 'b'}) + case.assertSetEqual(obj, {'c'}) + + obj.clear() + obj.update(['a', 'b', 'c']) + result = obj.symmetric_difference({'b', 'c', 'd'}) + case.assertSetEqual(result, {'a', 'd'}) + obj.symmetric_difference_update({'b', 'c', 'd'}) + case.assertSetEqual(obj, {'a', 'd'}) + + @classmethod + def _test_set_comparisons(cls, obj): + case = unittest.TestCase() + obj.update(['a', 'b', 'c']) + result = obj.union({'d', 'e'}) + case.assertSetEqual(result, {'a', 'b', 'c', 'd', 'e'}) + case.assertTrue(obj.isdisjoint({'d', 'e'})) + case.assertFalse(obj.isdisjoint({'a', 'd'})) + + case.assertTrue(obj.issubset({'a', 'b', 'c', 'd'})) + case.assertFalse(obj.issubset({'a', 'b'})) + case.assertLess(obj, {'a', 'b', 'c', 'd'}) + case.assertLessEqual(obj, {'a', 'b', 'c'}) + + case.assertTrue(obj.issuperset({'a', 'b'})) + case.assertFalse(obj.issuperset({'a', 'b', 'd'})) + case.assertGreater(obj, {'a'}) + case.assertGreaterEqual(obj, {'a', 'b'}) + + def test_set(self): + o = self.manager.set() + self.run_worker(self._test_set_operator_symbols, o) + o = self.manager.set() + self.run_worker(self._test_set_operator_methods, o) + o = self.manager.set() + self.run_worker(self._test_set_comparisons, o) + + def test_set_init(self): + o = self.manager.set({'a', 'b', 'c'}) + self.assertSetEqual(o, {'a', 'b', 'c'}) + o = self.manager.set(["a", "b", "c"]) + self.assertSetEqual(o, {"a", "b", "c"}) + o = self.manager.set({"a": 1, "b": 2, "c": 3}) + self.assertSetEqual(o, {"a", "b", "c"}) + self.assertRaises(RemoteError, self.manager.set, 1234) + + def test_set_contain_all_method(self): + o = self.manager.set() + set_methods = { + '__and__', '__class_getitem__', '__contains__', '__iand__', '__ior__', + '__isub__', '__iter__', '__ixor__', '__len__', '__or__', '__rand__', + '__ror__', '__rsub__', '__rxor__', '__sub__', '__xor__', + '__ge__', '__gt__', '__le__', '__lt__', + 'add', 'clear', 'copy', 'difference', 'difference_update', 'discard', + 'intersection', 'intersection_update', 'isdisjoint', 'issubset', + 'issuperset', 'pop', 'remove', 'symmetric_difference', + 'symmetric_difference_update', 'union', 'update', + } + self.assertLessEqual(set_methods, set(dir(o))) + class TestNamedResource(unittest.TestCase): @only_run_in_spawn_testsuite("spawn specific test.") @@ -6571,6 +7191,43 @@ class MiscTestCase(unittest.TestCase): out = out.decode().split("\n") self.assertEqual(out, ['__main__', '__mp_main__', 'f', 'f', '']) + def test_preload_main_sys_argv(self): + # gh-143706: Check that sys.argv is set before __main__ is pre-loaded + if multiprocessing.get_start_method() != "forkserver": + self.skipTest("forkserver specific test") + + name = os.path.join(os.path.dirname(__file__), 'mp_preload_sysargv.py') + _, out, err = test.support.script_helper.assert_python_ok( + name, 'foo', 'bar') + self.assertEqual(err, b'') + + out = out.decode().split("\n") + expected_argv = "['foo', 'bar']" + self.assertEqual(out, [ + f"module:{expected_argv}", + f"fun:{expected_argv}", + f"module:{expected_argv}", + f"fun:{expected_argv}", + '', + ]) + + @only_run_in_forkserver_testsuite("forkserver specific test.") + def test_preload_main_large_sys_argv(self): + # gh-144503: a very large parent sys.argv must not prevent the + # forkserver from starting (it previously overflowed the OS + # per-argument length limit when repr'd into the -c command string). + name = os.path.join(os.path.dirname(__file__), + 'mp_preload_large_sysargv.py') + _, out, err = test.support.script_helper.assert_python_ok(name) + self.assertEqual(err, b'') + + out = out.decode().split("\n") + self.assertEqual(out, [ + 'preload:5002:sentinel', + 'worker:5002:sentinel', + '', + ]) + # # Mixins # @@ -6587,15 +7244,13 @@ class BaseMixin(object): # cycles. Trigger a garbage collection to break these cycles. test.support.gc_collect() - # TODO: RUSTPYTHON: Filter out stopped processes since gc.collect() is a no-op - processes = {p for p in multiprocessing.process._dangling if p.is_alive()} - {p for p in cls.dangling[0] if p.is_alive()} + processes = set(multiprocessing.process._dangling) - set(cls.dangling[0]) if processes: test.support.environment_altered = True support.print_warning(f'Dangling processes: {processes}') processes = None - # TODO: RUSTPYTHON: Filter out stopped threads since gc.collect() is a no-op - threads = {t for t in threading._dangling if t.is_alive()} - {t for t in cls.dangling[1] if t.is_alive()} + threads = set(threading._dangling) - set(cls.dangling[1]) if threads: test.support.environment_altered = True support.print_warning(f'Dangling threads: {threads}') @@ -6785,16 +7440,14 @@ def install_tests_in_module_dict(remote_globs, start_method, multiprocessing.set_start_method(old_start_method[0], force=True) # pause a bit so we don't get warning about dangling threads/processes - # TODO: RUSTPYTHON: Filter out stopped processes since gc.collect() is a no-op - processes = {p for p in multiprocessing.process._dangling if p.is_alive()} - {p for p in dangling[0] if p.is_alive()} + processes = set(multiprocessing.process._dangling) - set(dangling[0]) if processes: need_sleep = True test.support.environment_altered = True support.print_warning(f'Dangling processes: {processes}') processes = None - # TODO: RUSTPYTHON: Filter out stopped threads since gc.collect() is a no-op - threads = {t for t in threading._dangling if t.is_alive()} - {t for t in dangling[1] if t.is_alive()} + threads = set(threading._dangling) - set(dangling[1]) if threads: need_sleep = True test.support.environment_altered = True diff --git a/Lib/test/mp_preload_large_sysargv.py b/Lib/test/mp_preload_large_sysargv.py new file mode 100644 index 000000000..790fcd76e --- /dev/null +++ b/Lib/test/mp_preload_large_sysargv.py @@ -0,0 +1,30 @@ +# gh-144503: Test that the forkserver can start when the parent process has +# a very large sys.argv. Prior to the fix, sys.argv was repr'd into the +# forkserver ``-c`` command string which could exceed the OS limit on the +# length of a single argv element (MAX_ARG_STRLEN on Linux, ~128 KiB), +# causing posix_spawn to fail and the parent to see a BrokenPipeError. + +import multiprocessing +import sys + +EXPECTED_LEN = 5002 # argv[0] + 5000 padding entries + sentinel + + +def fun(): + print(f"worker:{len(sys.argv)}:{sys.argv[-1]}") + + +if __name__ == "__main__": + # Inflate sys.argv well past 128 KiB before the forkserver is started. + sys.argv[1:] = ["x" * 50] * 5000 + ["sentinel"] + assert len(sys.argv) == EXPECTED_LEN + + ctx = multiprocessing.get_context("forkserver") + p = ctx.Process(target=fun) + p.start() + p.join() + sys.exit(p.exitcode) +else: + # This branch runs when the forkserver preloads this module as + # __mp_main__; confirm the large argv was propagated intact. + print(f"preload:{len(sys.argv)}:{sys.argv[-1]}") diff --git a/Lib/test/mp_preload_sysargv.py b/Lib/test/mp_preload_sysargv.py new file mode 100644 index 000000000..5ad38cda8 --- /dev/null +++ b/Lib/test/mp_preload_sysargv.py @@ -0,0 +1,22 @@ +# gh-143706: Test that sys.argv is correctly set during main module import +# when using forkserver with __main__ preloading. + +import multiprocessing +import sys + +# This will be printed during module import - sys.argv should be correct here +print(f"module:{sys.argv[1:]}") + +def fun(): + # This will be printed when the function is called + print(f"fun:{sys.argv[1:]}") + +if __name__ == "__main__": + ctx = multiprocessing.get_context("forkserver") + ctx.set_forkserver_preload(['__main__']) + + fun() + + p = ctx.Process(target=fun) + p.start() + p.join() diff --git a/Lib/test/test_multiprocessing_fork/test_processes.py b/Lib/test/test_multiprocessing_fork/test_processes.py index 987d5d7e3..e64e9afc0 100644 --- a/Lib/test/test_multiprocessing_fork/test_processes.py +++ b/Lib/test/test_multiprocessing_fork/test_processes.py @@ -3,6 +3,5 @@ from test._test_multiprocessing import install_tests_in_module_dict install_tests_in_module_dict(globals(), 'fork', only_type="processes") - if __name__ == '__main__': unittest.main() diff --git a/crates/stdlib/src/_queue.rs b/crates/stdlib/src/_queue.rs new file mode 100644 index 000000000..65456bd35 --- /dev/null +++ b/crates/stdlib/src/_queue.rs @@ -0,0 +1,309 @@ +pub(crate) use _queue::module_def; + +#[pymodule] +mod _queue { + use alloc::collections::VecDeque; + use core::time::Duration; + use std::time::Instant; + + use crate::vm::{ + AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, VirtualMachine, + builtins::{PyBaseExceptionRef, PyException, PyGenericAlias, PyStr, PyType, PyTypeRef}, + function::{PyComparisonValue, TimeoutSeconds}, + protocol::PyNumberMethods, + types::{AsNumber, Comparable, Constructor, PyComparisonOp, Representable}, + }; + + type BufInner = VecDeque; + + cfg_select! { + feature = "threading" => { + use parking_lot::{Condvar, Mutex, MutexGuard}; + + type Buf = Mutex; + }, + _ => { + use crate::common::lock::PyMutex; + + type Buf = PyMutex; + } + } + + const INITIAL_RING_BUF_CAPACITY: usize = 8; + + #[pyattr] + #[pyclass(module = "_queue", name = "Empty", base = PyException)] + #[repr(transparent)] + pub(crate) struct PyEmptyError(PyException); + + #[pyclass(flags(HAS_WEAKREF))] + impl PyEmptyError {} + + /// ## See Also + /// + /// [`empty_error`](https://github.com/python/cpython/blob/v3.14.5/Modules/_queuemodule.c#L347-L355). + fn empty_error(vm: &VirtualMachine) -> PyBaseExceptionRef { + vm.new_exception_empty(PyEmptyError::class(&vm.ctx).to_owned()) + } + + #[cfg(feature = "threading")] + #[derive(Debug)] + struct Semaphore { + mutex: Mutex, + cond: Condvar, + } + + #[cfg(feature = "threading")] + impl Semaphore { + #[must_use] + fn new() -> Self { + Self { + mutex: Mutex::new(0), + cond: Condvar::new(), + } + } + + fn release(&self) { + { + let mut count = self.mutex.lock(); + *count += 1; + } // lock dropped. now we can notify a waiting thread + + self.cond.notify_one(); + } + + /// Returns `true` if the semaphore was acquired, `false` on timeout. + #[must_use] + fn acquire(&self, block: bool, deadline: Option, vm: &VirtualMachine) -> bool { + let mut count = self.mutex.lock(); + loop { + if *count > 0 { + *count -= 1; + return true; + } + + if !block { + return false; + } + + match deadline { + Some(dl) => { + let result = vm.allow_threads(|| self.cond.wait_until(&mut count, dl)); + if result.timed_out() && *count == 0 { + return false; + } + } + None => { + vm.allow_threads(|| self.cond.wait(&mut count)); + } + } + } + } + } + + #[pyattr] + #[pyclass(module = "_queue", name = "SimpleQueue", unhashable = true)] + #[derive(Debug, PyPayload)] + struct PySimpleQueue { + buf: Buf, + #[cfg(feature = "threading")] + sem: Semaphore, + } + + impl Default for PySimpleQueue { + fn default() -> Self { + Self { + buf: Buf::new(VecDeque::with_capacity(INITIAL_RING_BUF_CAPACITY)), + #[cfg(feature = "threading")] + sem: Semaphore::new(), + } + } + } + + impl PySimpleQueue { + fn push(&self, item: PyObjectRef) { + self.buf.lock().push_back(item); + + #[cfg(feature = "threading")] + self.sem.release(); + } + + /// Returns a strong reference from the head of the buffer. + /// + /// ## See Also + /// + /// [`RingBuf_Get`](https://github.com/python/cpython/blob/v3.14.5/Modules/_queuemodule.c#L133-L154). + fn get_inner( + #[cfg(feature = "threading")] buf: &mut MutexGuard<'_, BufInner>, + #[cfg(not(feature = "threading"))] buf: &mut BufInner, + ) -> Option { + let cap = buf.capacity(); + + if buf.len() < (cap / 4) { + // Items is less than 25% occupied, shrink it by 50%. This allows for + // growth without immediately needing to resize the underlying items array + buf.shrink_to(cap / 2) + } + + buf.pop_front() + } + } + + #[derive(FromArgs)] + struct PutArgs { + #[pyarg(positional)] + item: PyObjectRef, + #[expect( + dead_code, + reason = "Intentional. Provide compatibility with the Queue class" + )] + #[pyarg(any, optional, default = true)] + block: bool, + #[expect( + dead_code, + reason = "Intentional. Provide compatibility with the Queue class" + )] + #[pyarg(any, optional)] + timeout: Option, + } + + #[derive(FromArgs)] + struct GetArgs { + #[pyarg(any, optional, default = true)] + block: bool, + #[pyarg(any, optional)] + timeout: Option, + } + + #[pyclass( + with(Constructor, Comparable, Representable), + flags(BASETYPE, HAS_WEAKREF, IMMUTABLETYPE) + )] + impl PySimpleQueue { + #[pymethod] + fn empty(&self) -> bool { + self.buf.lock().is_empty() + } + + #[pymethod] + fn qsize(&self) -> usize { + self.buf.lock().len() + } + + #[pymethod] + fn put(&self, args: PutArgs) { + let PutArgs { item, .. } = args; + self.push(item); + } + + #[pymethod] + fn put_nowait(&self, item: PyObjectRef) { + self.push(item); + } + + #[pymethod] + fn get(&self, args: GetArgs, vm: &VirtualMachine) -> PyResult { + let GetArgs { block, timeout } = args; + + // Non-blocking: just try once + if !block { + return Self::get_inner(&mut self.buf.lock()).ok_or_else(|| empty_error(vm)); + } + + #[cfg_attr( + not(feature = "threading"), + expect( + unused_variables, + reason = "We are still validating the 'timeout' arg even if we don't have threading" + ) + )] + let deadline = match timeout.map(|v| v.to_secs_f64()) { + Some(v) if v < 0.0 => { + return Err(vm.new_value_error("'timeout' must be a non-negative number")); + } + Some(v) => Some(Instant::now() + Duration::from_secs_f64(v)), + None => None, + }; + + #[cfg(feature = "threading")] + { + if !self.sem.acquire(block, deadline, vm) { + return Err(empty_error(vm)); + } + } + + Self::get_inner(&mut self.buf.lock()).ok_or_else(|| empty_error(vm)) + } + + #[pymethod] + fn get_nowait(&self, vm: &VirtualMachine) -> PyResult { + #[cfg(feature = "threading")] + { + if !self.sem.acquire(false, None, vm) { + return Err(empty_error(vm)); + } + } + + Self::get_inner(&mut self.buf.lock()).ok_or_else(|| empty_error(vm)) + } + + #[pyclassmethod] + fn __class_getitem__( + cls: PyTypeRef, + args: PyObjectRef, + vm: &VirtualMachine, + ) -> PyGenericAlias { + PyGenericAlias::from_args(cls, args, vm) + } + } + + impl Constructor for PySimpleQueue { + type Args = (); + + fn py_new(_cls: &Py, _args: Self::Args, _vm: &VirtualMachine) -> PyResult { + Ok(Self::default()) + } + } + + impl AsNumber for PySimpleQueue { + fn as_number() -> &'static PyNumberMethods { + static AS_NUMBER: PyNumberMethods = PyNumberMethods { + boolean: Some(|number, _vm| { + let zelf = number.obj.downcast_ref::().unwrap(); + Ok(!zelf.buf.lock().is_empty()) + }), + ..PyNumberMethods::NOT_IMPLEMENTED + }; + &AS_NUMBER + } + } + + impl Comparable for PySimpleQueue { + fn cmp( + zelf: &Py, + other: &PyObject, + op: PyComparisonOp, + _vm: &VirtualMachine, + ) -> PyResult { + Ok(if let Some(res) = op.identical_optimization(zelf, other) { + res.into() + } else { + PyComparisonValue::NotImplemented + }) + } + } + + impl Representable for PySimpleQueue { + fn repr(zelf: &Py, vm: &VirtualMachine) -> PyResult> { + Ok(vm.ctx.new_str(format!( + "<{} at {:#x}>", + Self::class(&vm.ctx).slot_name(), + zelf.get_id() + ))) + } + + fn repr_str(_zelf: &Py, _vm: &VirtualMachine) -> PyResult { + unreachable!("repr() is overridden directly") + } + } +} diff --git a/crates/stdlib/src/lib.rs b/crates/stdlib/src/lib.rs index b1ae53af8..119307874 100644 --- a/crates/stdlib/src/lib.rs +++ b/crates/stdlib/src/lib.rs @@ -53,6 +53,7 @@ mod math; #[cfg(all(feature = "host_env", any(unix, windows)))] mod mmap; +mod _queue; mod pyexpat; mod pystruct; mod random; @@ -225,6 +226,7 @@ pub fn stdlib_module_defs(ctx: &Context) -> Vec<&'static builtins::PyModuleDef> posixshmem::module_def(ctx), pyexpat::module_def(ctx), pystruct::module_def(ctx), + _queue::module_def(ctx), random::module_def(ctx), #[cfg(all(feature = "host_env", unix, not(target_os = "redox")))] resource::module_def(ctx),