Detach socket on create_connection/create_unix_connection cancellation

When create_connection(sock=sock) or create_unix_connection(sock=sock) is
cancelled or raises, tr._close() closes the fd via libuv but the Python
socket object still believes it owns that fd number. Its __del__ later
closes whatever fd the OS recycled into that slot, corrupting unrelated
transports.

Call sock.detach() on the error path so the Python socket sets its
internal fd to -1, matching the semantics of standard asyncio where the
transport always takes full ownership of the socket.

Fixes #738

Author: junjzhang

tests/test_tcp.py

@@ -1181,6 +1181,32 @@ def client():
         # let it close
         self.loop.run_until_complete(asyncio.sleep(0.1))
 
+    def test_create_connection_sock_cancel_detaches(self):
+        async def client(addr):
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.setblocking(False)
+            try:
+                sock.connect(addr)
+            except BlockingIOError:
+                pass
+            await asyncio.sleep(0.01)
+
+            task = asyncio.ensure_future(
+                self.loop.create_connection(asyncio.Protocol, sock=sock))
+            await asyncio.sleep(0)
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await task
+
+            # After cancellation the socket must be detached (fd == -1)
+            # so that its __del__ won't close a recycled fd.
+            self.assertEqual(sock.fileno(), -1)
+
+        with self.tcp_server(lambda sock: sock.recv_all(1),
+                             max_clients=1,
+                             backlog=1) as srv:
+            self.loop.run_until_complete(client(srv.addr))
+
     @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'no Unix sockets')
     def test_create_connection_wrong_sock(self):
         sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

uvloop/loop.pyx

@@ -2065,6 +2065,7 @@ cdef class Loop:
                 # up in `Transport._call_connection_made()`, and calling
                 # `_close()` before it is fine.
                 tr._close()
+                sock.detach()
                 raise
 
             tr._attach_fileobj(sock)
@@ -2307,6 +2308,7 @@ cdef class Loop:
                 raise
             except BaseException:
                 tr._close()
+                sock.detach()
                 raise
 
             tr._attach_fileobj(sock)