Fix sock_* methods to support already closed sockets on cancellation

Author: 1st1

tests/test_sockets.py

@@ -505,7 +505,7 @@ def test_socket_close_remove_writer(self):
             s.close()
             self.assertEqual(s.fileno(), -1)
 
-    def test_socket_cancel_sock_recv(self):
+    def test_socket_cancel_sock_recv_1(self):
         def srv_gen(sock):
             time.sleep(1.2)
             sock.send(b'helo')
@@ -534,6 +534,46 @@ async def client(sock, addr):
                 w = asyncio.wait_for(c, timeout=5.0, loop=self.loop)
                 self.loop.run_until_complete(w)
 
+    def test_socket_cancel_sock_recv_2(self):
+        def srv_gen(sock):
+            time.sleep(1.2)
+            sock.send(b'helo')
+
+        async def kill(fut):
+            await asyncio.sleep(0.5, loop=self.loop)
+            fut.cancel()
+
+        async def recv(sock):
+            fut = self.loop.sock_recv(sock, 10)
+            await asyncio.sleep(0.1, loop=self.loop)
+            self.loop.remove_reader(sock)
+            sock.close()
+            try:
+                await fut
+            except asyncio.CancelledError:
+                raise
+            finally:
+                sock.close()
+
+        async def client(sock, addr):
+            await self.loop.sock_connect(sock, addr)
+
+            f = asyncio.ensure_future(recv(sock), loop=self.loop)
+            self.loop.create_task(kill(f))
+            with self.assertRaises(asyncio.CancelledError):
+                await f
+            sock.close()
+            self.assertEqual(sock.fileno(), -1)
+
+        with self.tcp_server(srv_gen) as srv:
+
+            sock = socket.socket()
+            with sock:
+                sock.setblocking(False)
+                c = client(sock, srv.addr)
+                w = asyncio.wait_for(c, timeout=5.0, loop=self.loop)
+                self.loop.run_until_complete(w)
+
     def test_socket_cancel_sock_sendall(self):
         def srv_gen(sock):
             time.sleep(1.2)

uvloop/loop.pyx

@@ -804,7 +804,15 @@ cdef class Loop:
 
     cdef _new_reader_future(self, sock):
         def _on_cancel(fut):
-            if fut.cancelled():
+            # Check if the future was cancelled and if the socket
+            # is still open, i.e.
+            #
+            #    loop.remove_reader(sock)
+            #    sock.close()
+            #    fut.cancel()
+            #
+            # wasn't called by the user.
+            if fut.cancelled() and sock.fileno() != -1:
                 self._remove_reader(sock)
 
         fut = self._new_future()
@@ -813,7 +821,7 @@ cdef class Loop:
 
     cdef _new_writer_future(self, sock):
         def _on_cancel(fut):
-            if fut.cancelled():
+            if fut.cancelled() and sock.fileno() != -1:
                 self._remove_writer(sock)
 
         fut = self._new_future()