Merge 3.4: Issue #21119, fix ResourceWarning in asyncio

* Make sure that socketpair() close sockets on error. Close the listening
  socket if sock.bind() raises an exception.
* asyncio now closes sockets on errors. Fix ResourceWarning:
  create_connection(), create_datagram_endpoint() and create_unix_server()
  methods of event loop now close the newly created socket on error.

Lib/asyncio/base_events.py

@@ -412,6 +412,10 @@ class BaseEventLoop(events.AbstractEventLoop):
                     if sock is not None:
                         sock.close()
                     exceptions.append(exc)
+                except:
+                    if sock is not None:
+                        sock.close()
+                    raise
                 else:
                     break
             else:
@@ -512,6 +516,10 @@ class BaseEventLoop(events.AbstractEventLoop):
                 if sock is not None:
                     sock.close()
                 exceptions.append(exc)
+            except:
+                if sock is not None:
+                    sock.close()
+                raise
             else:
                 break
         else:

Lib/asyncio/unix_events.py

@@ -223,6 +223,9 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
                     raise OSError(errno.EADDRINUSE, msg) from None
                 else:
                     raise
+            except:
+                sock.close()
+                raise
         else:
             if sock is None:
                 raise ValueError(

Lib/asyncio/windows_utils.py

@@ -51,23 +51,25 @@ def socketpair(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0):
     # We create a connected TCP socket. Note the trick with setblocking(0)
     # that prevents us from having to create a thread.
     lsock = socket.socket(family, type, proto)
-    lsock.bind((host, 0))
-    lsock.listen(1)
-    # On IPv6, ignore flow_info and scope_id
-    addr, port = lsock.getsockname()[:2]
-    csock = socket.socket(family, type, proto)
-    csock.setblocking(False)
     try:
-        csock.connect((addr, port))
-    except (BlockingIOError, InterruptedError):
-        pass
-    except Exception:
+        lsock.bind((host, 0))
+        lsock.listen(1)
+        # On IPv6, ignore flow_info and scope_id
+        addr, port = lsock.getsockname()[:2]
+        csock = socket.socket(family, type, proto)
+        try:
+            csock.setblocking(False)
+            try:
+                csock.connect((addr, port))
+            except (BlockingIOError, InterruptedError):
+                pass
+            ssock, _ = lsock.accept()
+            csock.setblocking(True)
+        except:
+            csock.close()
+            raise
+    finally:
         lsock.close()
-        csock.close()
-        raise
-    ssock, _ = lsock.accept()
-    csock.setblocking(True)
-    lsock.close()
     return (ssock, csock)
 
 

Lib/test/test_asyncio/test_base_events.py

@@ -583,6 +583,27 @@ class BaseEventLoopWithSelectorTests(unittest.TestCase):
 
         self.assertEqual(str(cm.exception), 'Multiple exceptions: err1, err2')
 
+    @mock.patch('asyncio.base_events.socket')
+    def test_create_connection_timeout(self, m_socket):
+        # Ensure that the socket is closed on timeout
+        sock = mock.Mock()
+        m_socket.socket.return_value = sock
+
+        def getaddrinfo(*args, **kw):
+            fut = asyncio.Future(loop=self.loop)
+            addr = (socket.AF_INET, socket.SOCK_STREAM, 0, '',
+                    ('127.0.0.1', 80))
+            fut.set_result([addr])
+            return fut
+        self.loop.getaddrinfo = getaddrinfo
+
+        with mock.patch.object(self.loop, 'sock_connect',
+                               side_effect=asyncio.TimeoutError):
+            coro = self.loop.create_connection(MyProto, '127.0.0.1', 80)
+            with self.assertRaises(asyncio.TimeoutError) as cm:
+                self.loop.run_until_complete(coro)
+            self.assertTrue(sock.close.called)
+
     def test_create_connection_host_port_sock(self):
         coro = self.loop.create_connection(
             MyProto, 'example.com', 80, sock=object())

Lib/test/test_asyncio/test_unix_events.py

@@ -256,6 +256,24 @@ class SelectorEventLoopUnixSocketTests(unittest.TestCase):
                                         'A UNIX Domain Socket was expected'):
                 self.loop.run_until_complete(coro)
 
+    @mock.patch('asyncio.unix_events.socket')
+    def test_create_unix_server_bind_error(self, m_socket):
+        # Ensure that the socket is closed on any bind error
+        sock = mock.Mock()
+        m_socket.socket.return_value = sock
+
+        sock.bind.side_effect = OSError
+        coro = self.loop.create_unix_server(lambda: None, path="/test")
+        with self.assertRaises(OSError):
+            self.loop.run_until_complete(coro)
+        self.assertTrue(sock.close.called)
+
+        sock.bind.side_effect = MemoryError
+        coro = self.loop.create_unix_server(lambda: None, path="/test")
+        with self.assertRaises(MemoryError):
+            self.loop.run_until_complete(coro)
+        self.assertTrue(sock.close.called)
+
     def test_create_unix_connection_path_sock(self):
         coro = self.loop.create_unix_connection(
             lambda: None, '/dev/null', sock=object())

Lib/test/test_asyncio/test_windows_utils.py

@@ -51,6 +51,15 @@ class WinsocketpairTests(unittest.TestCase):
         self.assertRaises(ValueError,
                           windows_utils.socketpair, proto=1)
 
+    @mock.patch('asyncio.windows_utils.socket')
+    def test_winsocketpair_close(self, m_socket):
+        m_socket.AF_INET = socket.AF_INET
+        m_socket.SOCK_STREAM = socket.SOCK_STREAM
+        sock = mock.Mock()
+        m_socket.socket.return_value = sock
+        sock.bind.side_effect = OSError
+        self.assertRaises(OSError, windows_utils.socketpair)
+        self.assertTrue(sock.close.called)
 
 
 class PipeTests(unittest.TestCase):