Update multiprocessing from CPython 3.10.5

Author: CPython developers

Lib/multiprocessing/__init__.py

@@ -19,9 +19,8 @@
 # Copy stuff from default context
 #
 
-globals().update((name, getattr(context._default_context, name))
-                 for name in context._default_context.__all__)
-__all__ = context._default_context.__all__
+__all__ = [x for x in dir(context._default_context) if not x.startswith('_')]
+globals().update((name, getattr(context._default_context, name)) for name in __all__)
 
 #
 # XXX These should not really be documented or public.

Lib/multiprocessing/connection.py

@@ -57,10 +57,10 @@
 
 
 def _init_timeout(timeout=CONNECTION_TIMEOUT):
-    return time.time() + timeout
+    return time.monotonic() + timeout
 
 def _check_timeout(t):
-    return time.time() > t
+    return time.monotonic() > t
 
 #
 #
@@ -73,6 +73,11 @@ def arbitrary_address(family):
     if family == 'AF_INET':
         return ('localhost', 0)
     elif family == 'AF_UNIX':
+        # Prefer abstract sockets if possible to avoid problems with the address
+        # size.  When coding portable applications, some implementations have
+        # sun_path as short as 92 bytes in the sockaddr_un struct.
+        if util.abstract_sockets_supported:
+            return f"\0listener-{os.getpid()}-{next(_mmap_counter)}"
         return tempfile.mktemp(prefix='listener-', dir=util.get_temp_dir())
     elif family == 'AF_PIPE':
         return tempfile.mktemp(prefix=r'\\.\pipe\pyc-%d-%d-' %
@@ -102,7 +107,7 @@ def address_type(address):
         return 'AF_INET'
     elif type(address) is str and address.startswith('\\\\'):
         return 'AF_PIPE'
-    elif type(address) is str:
+    elif type(address) is str or util.is_abstract_socket_namespace(address):
         return 'AF_UNIX'
     else:
         raise ValueError('address type of %r unrecognized' % address)
@@ -389,23 +394,33 @@ def _recv(self, size, read=_read):
 
     def _send_bytes(self, buf):
         n = len(buf)
-        # For wire compatibility with 3.2 and lower
-        header = struct.pack("!i", n)
-        if n > 16384:
-            # The payload is large so Nagle's algorithm won't be triggered
-            # and we'd better avoid the cost of concatenation.
+        if n > 0x7fffffff:
+            pre_header = struct.pack("!i", -1)
+            header = struct.pack("!Q", n)
+            self._send(pre_header)
             self._send(header)
             self._send(buf)
         else:
-            # Issue #20540: concatenate before sending, to avoid delays due
-            # to Nagle's algorithm on a TCP socket.
-            # Also note we want to avoid sending a 0-length buffer separately,
-            # to avoid "broken pipe" errors if the other end closed the pipe.
-            self._send(header + buf)
+            # For wire compatibility with 3.7 and lower
+            header = struct.pack("!i", n)
+            if n > 16384:
+                # The payload is large so Nagle's algorithm won't be triggered
+                # and we'd better avoid the cost of concatenation.
+                self._send(header)
+                self._send(buf)
+            else:
+                # Issue #20540: concatenate before sending, to avoid delays due
+                # to Nagle's algorithm on a TCP socket.
+                # Also note we want to avoid sending a 0-length buffer separately,
+                # to avoid "broken pipe" errors if the other end closed the pipe.
+                self._send(header + buf)
 
     def _recv_bytes(self, maxsize=None):
         buf = self._recv(4)
         size, = struct.unpack("!i", buf.getvalue())
+        if size == -1:
+            buf = self._recv(8)
+            size, = struct.unpack("!Q", buf.getvalue())
         if maxsize is not None and size > maxsize:
             return None
         return self._recv(size)
@@ -465,8 +480,13 @@ def close(self):
             self._listener = None
             listener.close()
 
-    address = property(lambda self: self._listener._address)
-    last_accepted = property(lambda self: self._listener._last_accepted)
+    @property
+    def address(self):
+        return self._listener._address
+
+    @property
+    def last_accepted(self):
+        return self._listener._last_accepted
 
     def __enter__(self):
         return self
@@ -582,7 +602,8 @@ def __init__(self, address, family, backlog=1):
         self._family = family
         self._last_accepted = None
 
-        if family == 'AF_UNIX':
+        if family == 'AF_UNIX' and not util.is_abstract_socket_namespace(address):
+            # Linux abstract socket namespaces do not need to be explicitly unlinked
             self._unlink = util.Finalize(
                 self, os.unlink, args=(address,), exitpriority=0
                 )
@@ -715,7 +736,9 @@ def PipeClient(address):
 
 def deliver_challenge(connection, authkey):
     import hmac
-    assert isinstance(authkey, bytes)
+    if not isinstance(authkey, bytes):
+        raise ValueError(
+            "Authkey must be bytes, not {0!s}".format(type(authkey)))
     message = os.urandom(MESSAGE_LENGTH)
     connection.send_bytes(CHALLENGE + message)
     digest = hmac.new(authkey, message, 'md5').digest()
@@ -728,7 +751,9 @@ def deliver_challenge(connection, authkey):
 
 def answer_challenge(connection, authkey):
     import hmac
-    assert isinstance(authkey, bytes)
+    if not isinstance(authkey, bytes):
+        raise ValueError(
+            "Authkey must be bytes, not {0!s}".format(type(authkey)))
     message = connection.recv_bytes(256)         # reject large message
     assert message[:len(CHALLENGE)] == CHALLENGE, 'message = %r' % message
     message = message[len(CHALLENGE):]
@@ -780,8 +805,7 @@ def XmlClient(*args, **kwds):
 # Wait
 #
 
-# XXX RustPython TODO: implement all the functions in this block
-if sys.platform == 'win32' and False:
+if sys.platform == 'win32':
 
     def _exhaustive_wait(handles, timeout):
         # Return ALL handles which are currently signalled.  (Only
@@ -906,15 +930,15 @@ def wait(object_list, timeout=None):
                 selector.register(obj, selectors.EVENT_READ)
 
             if timeout is not None:
-                deadline = time.time() + timeout
+                deadline = time.monotonic() + timeout
 
             while True:
                 ready = selector.select(timeout)
                 if ready:
                     return [key.fileobj for (key, events) in ready]
                 else:
                     if timeout is not None:
-                        timeout = deadline - time.time()
+                        timeout = deadline - time.monotonic()
                         if timeout < 0:
                             return ready
 

Lib/multiprocessing/context.py

@@ -5,7 +5,7 @@
 from . import process
 from . import reduction
 
-__all__ = []            # things are copied from here to __init__.py
+__all__ = ()
 
 #
 # Exceptions
@@ -24,7 +24,7 @@ class AuthenticationError(ProcessError):
     pass
 
 #
-# Base type for contexts
+# Base type for contexts. Bound methods of an instance of this type are included in __all__ of __init__.py
 #
 
 class BaseContext(object):
@@ -35,6 +35,7 @@ class BaseContext(object):
     AuthenticationError = AuthenticationError
 
     current_process = staticmethod(process.current_process)
+    parent_process = staticmethod(process.parent_process)
     active_children = staticmethod(process.active_children)
 
     def cpu_count(self):
@@ -189,14 +190,14 @@ def get_context(self, method=None):
         try:
             ctx = _concrete_contexts[method]
         except KeyError:
-            raise ValueError('cannot find context for %r' % method)
+            raise ValueError('cannot find context for %r' % method) from None
         ctx._check_available()
         return ctx
 
     def get_start_method(self, allow_none=False):
         return self._name
 
-    def set_start_method(self, method=None):
+    def set_start_method(self, method, force=False):
         raise ValueError('cannot set start method of concrete context')
 
     @property
@@ -256,12 +257,11 @@ def get_all_start_methods(self):
         if sys.platform == 'win32':
             return ['spawn']
         else:
+            methods = ['spawn', 'fork'] if sys.platform == 'darwin' else ['fork', 'spawn']
             if reduction.HAVE_SEND_HANDLE:
-                return ['fork', 'spawn', 'forkserver']
-            else:
-                return ['fork', 'spawn']
+                methods.append('forkserver')
+            return methods
 
-DefaultContext.__all__ = list(x for x in dir(DefaultContext) if x[0] != '_')
 
 #
 # Context types for fixed start method
@@ -310,7 +310,12 @@ def _check_available(self):
         'spawn': SpawnContext(),
         'forkserver': ForkServerContext(),
     }
-    _default_context = DefaultContext(_concrete_contexts['fork'])
+    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'])
+    else:
+        _default_context = DefaultContext(_concrete_contexts['fork'])
 
 else:
 

Lib/multiprocessing/dummy/__init__.py

@@ -41,7 +41,10 @@ def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
         self._parent = current_process()
 
     def start(self):
-        assert self._parent is current_process()
+        if self._parent is not current_process():
+            raise RuntimeError(
+                "Parent is {0!r} but current_process is {1!r}".format(
+                    self._parent, current_process()))
         self._start_called = True
         if hasattr(self._parent, '_children'):
             self._parent._children[self] = None
@@ -77,7 +80,7 @@ def freeze_support():
 #
 
 class Namespace(object):
-    def __init__(self, **kwds):
+    def __init__(self, /, **kwds):
         self.__dict__.update(kwds)
     def __repr__(self):
         items = list(self.__dict__.items())
@@ -98,11 +101,15 @@ class Value(object):
     def __init__(self, typecode, value, lock=True):
         self._typecode = typecode
         self._value = value
-    def _get(self):
+
+    @property
+    def value(self):
         return self._value
-    def _set(self, value):
+
+    @value.setter
+    def value(self, value):
         self._value = value
-    value = property(_get, _set)
+
     def __repr__(self):
         return '<%s(%r, %r)>'%(type(self).__name__,self._typecode,self._value)
 

Lib/multiprocessing/dummy/connection.py

@@ -26,7 +26,9 @@ def accept(self):
     def close(self):
         self._backlog_queue = None
 
-    address = property(lambda self: self._backlog_queue)
+    @property
+    def address(self):
+        return self._backlog_queue
 
     def __enter__(self):
         return self