Issue #4892: multiprocessing Connections can now be transferred over multiprocessing Connections.

Patch by Richard Oudkerk (sbt).

Doc/library/multiprocessing.rst

@@ -832,6 +832,10 @@ Connection objects are usually created using :func:`Pipe` -- see also
       raised and the complete message is available as ``e.args[0]`` where ``e``
       is the exception instance.
 
+   .. versionchanged:: 3.3
+      Connection objects themselves can now be transferred between processes
+      using :meth:`Connection.send` and :meth:`Connection.recv`.
+
 
 For example:
 

Lib/multiprocessing/__init__.py

@@ -161,7 +161,9 @@ def allow_connection_pickling():
     '''
     Install support for sending connections and sockets between processes
     '''
-    from multiprocessing import reduction
+    # This is undocumented.  In previous versions of multiprocessing
+    # its only effect was to make socket objects inheritable on Windows.
+    import multiprocessing.connection
 
 #
 # Definitions depending on native semaphores

Lib/multiprocessing/connection.py

@@ -50,6 +50,7 @@ import _multiprocessing
 from multiprocessing import current_process, AuthenticationError, BufferTooShort
 from multiprocessing.util import (
     get_temp_dir, Finalize, sub_debug, debug, _eintr_retry)
+from multiprocessing.forking import ForkingPickler
 try:
     import _winapi
     from _winapi import WAIT_OBJECT_0, WAIT_TIMEOUT, INFINITE
@@ -227,8 +228,9 @@ class _ConnectionBase:
         """Send a (picklable) object"""
         self._check_closed()
         self._check_writable()
-        buf = pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)
-        self._send_bytes(memoryview(buf))
+        buf = io.BytesIO()
+        ForkingPickler(buf, pickle.HIGHEST_PROTOCOL).dump(obj)
+        self._send_bytes(buf.getbuffer())
 
     def recv_bytes(self, maxlength=None):
         """
@@ -880,3 +882,21 @@ else:
                     raise
             if timeout is not None:
                 timeout = deadline - time.time()
+
+#
+# Make connection and socket objects sharable if possible
+#
+
+if sys.platform == 'win32':
+    from . import reduction
+    ForkingPickler.register(socket.socket, reduction.reduce_socket)
+    ForkingPickler.register(Connection, reduction.reduce_connection)
+    ForkingPickler.register(PipeConnection, reduction.reduce_pipe_connection)
+else:
+    try:
+        from . import reduction
+    except ImportError:
+        pass
+    else:
+        ForkingPickler.register(socket.socket, reduction.reduce_socket)
+        ForkingPickler.register(Connection, reduction.reduce_connection)

Lib/multiprocessing/forking.py

@@ -407,25 +407,6 @@ else:
 
         return d
 
-    #
-    # Make (Pipe)Connection picklable
-    #
-
-    # Late import because of circular import
-    from .connection import Connection, PipeConnection
-
-    def reduce_connection(conn):
-        if not Popen.thread_is_spawning():
-            raise RuntimeError(
-                'By default %s objects can only be shared between processes\n'
-                'using inheritance' % type(conn).__name__
-                )
-        return type(conn), (Popen.duplicate_for_child(conn.fileno()),
-                            conn.readable, conn.writable)
-
-    ForkingPickler.register(Connection, reduce_connection)
-    ForkingPickler.register(PipeConnection, reduce_connection)
-
 #
 # Prepare current process
 #

Lib/test/test_multiprocessing.py

@@ -1959,35 +1959,40 @@ class _TestPoll(unittest.TestCase):
 #
 # Test of sending connection and socket objects between processes
 #
-"""
+
+@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
 class _TestPicklingConnections(BaseTestCase):
 
     ALLOWED_TYPES = ('processes',)
 
-    def _listener(self, conn, families):
+    @classmethod
+    def _listener(cls, conn, families):
         for fam in families:
-            l = self.connection.Listener(family=fam)
+            l = cls.connection.Listener(family=fam)
             conn.send(l.address)
             new_conn = l.accept()
             conn.send(new_conn)
+            new_conn.close()
+            l.close()
 
-        if self.TYPE == 'processes':
         l = socket.socket()
         l.bind(('localhost', 0))
         conn.send(l.getsockname())
         l.listen(1)
         new_conn, addr = l.accept()
         conn.send(new_conn)
+        new_conn.close()
+        l.close()
 
         conn.recv()
 
-    def _remote(self, conn):
+    @classmethod
+    def _remote(cls, conn):
         for (address, msg) in iter(conn.recv, None):
-            client = self.connection.Client(address)
+            client = cls.connection.Client(address)
             client.send(msg.upper())
             client.close()
 
-        if self.TYPE == 'processes':
         address, msg = conn.recv()
         client = socket.socket()
         client.connect(address)
@@ -1997,11 +2002,6 @@ class _TestPicklingConnections(BaseTestCase):
         conn.close()
 
     def test_pickling(self):
-        try:
-            multiprocessing.allow_connection_pickling()
-        except ImportError:
-            return
-
         families = self.connection.families
 
         lconn, lconn0 = self.Pipe()
@@ -2025,16 +2025,12 @@ class _TestPicklingConnections(BaseTestCase):
 
         rconn.send(None)
 
-        if self.TYPE == 'processes':
         msg = latin('This connection uses a normal socket')
         address = lconn.recv()
         rconn.send((address, msg))
-            if hasattr(socket, 'fromfd'):
         new_conn = lconn.recv()
         self.assertEqual(new_conn.recv(100), msg.upper())
-            else:
-                # XXX On Windows with Py2.6 need to backport fromfd()
-                discard = lconn.recv_bytes()
+        new_conn.close()
 
         lconn.send(None)
 
@@ -2043,7 +2039,46 @@ class _TestPicklingConnections(BaseTestCase):
 
         lp.join()
         rp.join()
-"""
+
+    @classmethod
+    def child_access(cls, conn):
+        w = conn.recv()
+        w.send('all is well')
+        w.close()
+
+        r = conn.recv()
+        msg = r.recv()
+        conn.send(msg*2)
+
+        conn.close()
+
+    def test_access(self):
+        # On Windows, if we do not specify a destination pid when
+        # using DupHandle then we need to be careful to use the
+        # correct access flags for DuplicateHandle(), or else
+        # DupHandle.detach() will raise PermissionError.  For example,
+        # for a read only pipe handle we should use
+        # access=FILE_GENERIC_READ.  (Unfortunately
+        # DUPLICATE_SAME_ACCESS does not work.)
+        conn, child_conn = self.Pipe()
+        p = self.Process(target=self.child_access, args=(child_conn,))
+        p.daemon = True
+        p.start()
+        child_conn.close()
+
+        r, w = self.Pipe(duplex=False)
+        conn.send(w)
+        w.close()
+        self.assertEqual(r.recv(), 'all is well')
+        r.close()
+
+        r, w = self.Pipe(duplex=False)
+        conn.send(r)
+        r.close()
+        w.send('foobar')
+        w.close()
+        self.assertEqual(conn.recv(), 'foobar'*2)
+
 #
 #
 #

Misc/NEWS

@@ -71,6 +71,9 @@ Core and Builtins
 Library
 -------
 
+- Issue #4892: multiprocessing Connections can now be transferred over
+  multiprocessing Connections.  Patch by Richard Oudkerk (sbt).
+
 - Issue #14160: TarFile.extractfile() failed to resolve symbolic links when
   the links were not located in an archive subdirectory.
 

Modules/_winapi.c

@@ -1280,6 +1280,7 @@ PyInit__winapi(void)
     WINAPI_CONSTANT(F_DWORD, CREATE_NEW_CONSOLE);
     WINAPI_CONSTANT(F_DWORD, CREATE_NEW_PROCESS_GROUP);
     WINAPI_CONSTANT(F_DWORD, DUPLICATE_SAME_ACCESS);
+    WINAPI_CONSTANT(F_DWORD, DUPLICATE_CLOSE_SOURCE);
     WINAPI_CONSTANT(F_DWORD, ERROR_ALREADY_EXISTS);
     WINAPI_CONSTANT(F_DWORD, ERROR_BROKEN_PIPE);
     WINAPI_CONSTANT(F_DWORD, ERROR_IO_PENDING);
@@ -1298,6 +1299,8 @@ PyInit__winapi(void)
     WINAPI_CONSTANT(F_DWORD, ERROR_SEM_TIMEOUT);
     WINAPI_CONSTANT(F_DWORD, FILE_FLAG_FIRST_PIPE_INSTANCE);
     WINAPI_CONSTANT(F_DWORD, FILE_FLAG_OVERLAPPED);
+    WINAPI_CONSTANT(F_DWORD, FILE_GENERIC_READ);
+    WINAPI_CONSTANT(F_DWORD, FILE_GENERIC_WRITE);
     WINAPI_CONSTANT(F_DWORD, GENERIC_READ);
     WINAPI_CONSTANT(F_DWORD, GENERIC_WRITE);
     WINAPI_CONSTANT(F_DWORD, INFINITE);
@@ -1310,6 +1313,7 @@ PyInit__winapi(void)
     WINAPI_CONSTANT(F_DWORD, PIPE_UNLIMITED_INSTANCES);
     WINAPI_CONSTANT(F_DWORD, PIPE_WAIT);
     WINAPI_CONSTANT(F_DWORD, PROCESS_ALL_ACCESS);
+    WINAPI_CONSTANT(F_DWORD, PROCESS_DUP_HANDLE);
     WINAPI_CONSTANT(F_DWORD, STARTF_USESHOWWINDOW);
     WINAPI_CONSTANT(F_DWORD, STARTF_USESTDHANDLES);
     WINAPI_CONSTANT(F_DWORD, STD_INPUT_HANDLE);