impl _queue module + Update multiprocessing to 3.14.5 (#7994)

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

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'

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()

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):

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)

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()

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()

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()

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))

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'{'):

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)

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

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)

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

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]}")

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()

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()

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<PyObjectRef>;
+
+    cfg_select! {
+        feature = "threading" => {
+            use parking_lot::{Condvar, Mutex, MutexGuard};
+
+            type Buf = Mutex<BufInner>;
+        },
+        _ => {
+            use crate::common::lock::PyMutex;
+
+            type Buf = PyMutex<BufInner>;
+        }
+    }
+
+    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<usize>,
+        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<Instant>, 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<PyObjectRef> {
+            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<PyObjectRef>,
+    }
+
+    #[derive(FromArgs)]
+    struct GetArgs {
+        #[pyarg(any, optional, default = true)]
+        block: bool,
+        #[pyarg(any, optional)]
+        timeout: Option<TimeoutSeconds>,
+    }
+
+    #[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<PyObjectRef> {
+            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<PyObjectRef> {
+            #[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<PyType>, _args: Self::Args, _vm: &VirtualMachine) -> PyResult<Self> {
+            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::<PySimpleQueue>().unwrap();
+                    Ok(!zelf.buf.lock().is_empty())
+                }),
+                ..PyNumberMethods::NOT_IMPLEMENTED
+            };
+            &AS_NUMBER
+        }
+    }
+
+    impl Comparable for PySimpleQueue {
+        fn cmp(
+            zelf: &Py<Self>,
+            other: &PyObject,
+            op: PyComparisonOp,
+            _vm: &VirtualMachine,
+        ) -> PyResult<PyComparisonValue> {
+            Ok(if let Some(res) = op.identical_optimization(zelf, other) {
+                res.into()
+            } else {
+                PyComparisonValue::NotImplemented
+            })
+        }
+    }
+
+    impl Representable for PySimpleQueue {
+        fn repr(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<PyRef<PyStr>> {
+            Ok(vm.ctx.new_str(format!(
+                "<{} at {:#x}>",
+                Self::class(&vm.ctx).slot_name(),
+                zelf.get_id()
+            )))
+        }
+
+        fn repr_str(_zelf: &Py<Self>, _vm: &VirtualMachine) -> PyResult<String> {
+            unreachable!("repr() is overridden directly")
+        }
+    }
+}

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),