Add AF_INET6 support to multiprocessing IPC.

Also makes the test_multiprocessing suites work on IPv6-only hosts.

Lib/multiprocessing/connection.py

@@ -51,6 +51,9 @@
     default_family = 'AF_UNIX'
     families += ['AF_UNIX']
 
+if hasattr(socket, 'AF_INET6') and socket.has_ipv6:
+    families.append('AF_INET6')
+
 if sys.platform == 'win32':
     default_family = 'AF_PIPE'
     families += ['AF_PIPE']
@@ -70,7 +73,7 @@ def arbitrary_address(family):
     '''
     Return an arbitrary free address for the given family
     '''
-    if family == 'AF_INET':
+    if family in {'AF_INET', 'AF_INET6'}:
         return ('localhost', 0)
     elif family == 'AF_UNIX':
         # Prefer abstract sockets if possible to avoid problems with the address
@@ -101,9 +104,16 @@ def address_type(address):
     '''
     Return the types of the address
 
-    This can be 'AF_INET', 'AF_UNIX', or 'AF_PIPE'
+    This can be 'AF_INET', 'AF_INET6', 'AF_UNIX', or 'AF_PIPE'
     '''
     if type(address) == tuple:
+        if '.' in address[0]:
+            return 'AF_INET'
+        if ':' in address[0]:
+            return 'AF_INET6'
+        addr_info = socket.getaddrinfo(*address[:2])
+        if addr_info:
+            return addr_info[0][0].name
         return 'AF_INET'
     elif type(address) is str and address.startswith('\\\\'):
         return 'AF_PIPE'

Lib/test/_test_multiprocessing.py

@@ -184,6 +184,14 @@ class BaseTestCase(object):
 
     ALLOWED_TYPES = ('processes', 'manager', 'threads')
 
+    def get_families(self):
+        fams = set(self.connection.families)
+        if not socket_helper.IPV6_ENABLED:
+            fams -= {'AF_INET6'}
+        if not socket_helper.IPV4_ENABLED:
+            fams -= {'AF_INET'}
+        return fams
+
     def assertTimingAlmostEqual(self, a, b):
         if CHECK_TIMINGS:
             self.assertAlmostEqual(a, b, 1)
@@ -3284,7 +3292,7 @@ class _TestListener(BaseTestCase):
     ALLOWED_TYPES = ('processes',)
 
     def test_multiple_bind(self):
-        for family in self.connection.families:
+        for family in self.get_families():
             l = self.connection.Listener(family=family)
             self.addCleanup(l.close)
             self.assertRaises(OSError, self.connection.Listener,
@@ -3324,7 +3332,7 @@ def _test(cls, address):
         conn.close()
 
     def test_listener_client(self):
-        for family in self.connection.families:
+        for family in self.get_families():
             l = self.connection.Listener(family=family)
             p = self.Process(target=self._test, args=(l.address,))
             p.daemon = True
@@ -3351,7 +3359,7 @@ def test_issue14725(self):
         l.close()
 
     def test_issue16955(self):
-        for fam in self.connection.families:
+        for fam in self.get_families():
             l = self.connection.Listener(family=fam)
             c = self.connection.Client(l.address)
             a = l.accept()
@@ -3464,7 +3472,8 @@ def _listener(cls, conn, families):
             new_conn.close()
             l.close()
 
-        l = socket.create_server((socket_helper.HOST, 0))
+        l = socket.create_server((socket_helper.HOST, 0),
+                                 family=socket_helper.get_family())
         conn.send(l.getsockname())
         new_conn, addr = l.accept()
         conn.send(new_conn)
@@ -3481,15 +3490,15 @@ def _remote(cls, conn):
             client.close()
 
         address, msg = conn.recv()
-        client = socket.socket()
+        client = socket_helper.tcp_socket()
         client.connect(address)
         client.sendall(msg.upper())
         client.close()
 
         conn.close()
 
     def test_pickling(self):
-        families = self.connection.families
+        families = self.get_families()
 
         lconn, lconn0 = self.Pipe()
         lp = self.Process(target=self._listener, args=(lconn0, families))
@@ -4638,7 +4647,7 @@ def test_wait(self, slow=False):
 
     @classmethod
     def _child_test_wait_socket(cls, address, slow):
-        s = socket.socket()
+        s = socket_helper.tcp_socket()
         s.connect(address)
         for i in range(10):
             if slow:
@@ -4648,7 +4657,8 @@ def _child_test_wait_socket(cls, address, slow):
 
     def test_wait_socket(self, slow=False):
         from multiprocessing.connection import wait
-        l = socket.create_server((socket_helper.HOST, 0))
+        l = socket.create_server((socket_helper.HOST, 0),
+                                 family=socket_helper.get_family())
         addr = l.getsockname()
         readers = []
         procs = []
@@ -4836,7 +4846,8 @@ def test_timeout(self):
         try:
             socket.setdefaulttimeout(0.1)
             parent, child = multiprocessing.Pipe(duplex=True)
-            l = multiprocessing.connection.Listener(family='AF_INET')
+            l = multiprocessing.connection.Listener(
+                    family=socket_helper.get_family().name)
             p = multiprocessing.Process(target=self._test_timeout,
                                         args=(child, l.address))
             p.start()
@@ -4910,11 +4921,11 @@ def get_high_socket_fd(self):
             # The child process will not have any socket handles, so
             # calling socket.fromfd() should produce WSAENOTSOCK even
             # if there is a handle of the same number.
-            return socket.socket().detach()
+            return socket_helper.tcp_socket().detach()
         else:
             # We want to produce a socket with an fd high enough that a
             # freshly created child process will not have any fds as high.
-            fd = socket.socket().detach()
+            fd = socket_helper.tcp_socket().detach()
             to_close = []
             while fd < 50:
                 to_close.append(fd)
@@ -4925,7 +4936,7 @@ def get_high_socket_fd(self):
 
     def close(self, fd):
         if WIN32:
-            socket.socket(socket.AF_INET, socket.SOCK_STREAM, fileno=fd).close()
+            socket.socket(socket_helper.get_family(), socket.SOCK_STREAM, fileno=fd).close()
         else:
             os.close(fd)