Issue #34: Fix UVPoll handler crash

Author: 1st1

tests/test_sockets.py

@@ -111,6 +111,48 @@ def test_socket_blocking_error(self):
                 self.loop.run_until_complete(
                     self.loop.sock_connect(sock, (b'', 0)))
 
+    def test_socket_handler_cleanup(self):
+        # This tests recreates a rare condition where we have a socket
+        # with an attached reader.  We then remove the reader, and close the
+        # socket.  If the libuv Poll handler is still cached when we open
+        # a new TCP connection, it might so happen that the new TCP connection
+        # will receive a fileno that our previous socket was registered on.
+        # In this case, when the cached Poll handle is finally closed,
+        # we have a failed assertion in uv_poll_stop.
+        # See also https://github.com/MagicStack/uvloop/issues/34
+        # for details.
+
+        srv_sock = socket.socket()
+        with srv_sock:
+            srv_sock.bind(('127.0.0.1', 0))
+            srv_sock.listen(100)
+
+            srv = self.loop.run_until_complete(
+                self.loop.create_server(
+                    lambda: None, host='127.0.0.1', port=0))
+            key_fileno = srv.sockets[0].fileno()
+            srv.close()
+            self.loop.run_until_complete(srv.wait_closed())
+
+            # Schedule create_connection task's callbacks
+            tsk = self.loop.create_task(
+                self.loop.create_connection(
+                    asyncio.Protocol, *srv_sock.getsockname()))
+
+            sock = socket.socket()
+            with sock:
+                # Add/remove readers
+                if sock.fileno() != key_fileno:
+                    raise unittest.SkipTest()
+                self.loop.add_reader(sock.fileno(), lambda: None)
+                self.loop.remove_reader(sock.fileno())
+
+            tr, pr = self.loop.run_until_complete(
+                asyncio.wait_for(tsk, loop=self.loop, timeout=0.1))
+            tr.close()
+            # Let the transport close
+            self.loop.run_until_complete(asyncio.sleep(0, loop=self.loop))
+
 
 class TestUVSockets(_TestSockets, tb.UVTestCase):
     pass

uvloop/loop.pxd

@@ -58,7 +58,6 @@ cdef class Loop:
 
         set _timers
         dict _polls
-        dict _polls_gc
 
         dict _signal_handlers
         bint _custom_sigint

uvloop/loop.pyx

@@ -70,7 +70,6 @@ cdef class Loop:
 
         self._timers = set()
         self._polls = dict()
-        self._polls_gc = dict()
 
         if MAIN_THREAD_ID == PyThread_get_thread_ident():  # XXX
             self.py_signals = SignalsStack()
@@ -243,14 +242,6 @@ cdef class Loop:
         if len(self._queued_streams):
             self._exec_queued_writes()
 
-        if len(self._polls_gc):
-            for fd in tuple(self._polls_gc):
-                poll = <UVPoll> self._polls_gc[fd]
-                if not poll.is_active():
-                    poll._close()
-                    self._polls.pop(fd)
-                self._polls_gc.pop(fd)
-
         self._ready_len = len(self._ready)
         if self._ready_len == 0 and self.handler_idle.running:
             self.handler_idle.stop()
@@ -353,7 +344,6 @@ cdef class Loop:
                 (<UVHandle>poll_handle)._close()
 
             self._polls.clear()
-            self._polls_gc.clear()
 
         if self._timers:
             for timer_cbhandle in tuple(self._timers):
@@ -518,7 +508,8 @@ cdef class Loop:
 
         result = poll.stop_reading()
         if not poll.is_active():
-            self._polls_gc[fd] = poll
+            del self._polls[fd]
+            poll._close()
         return result
 
     cdef inline _add_writer(self, fd, Handle handle):
@@ -549,7 +540,8 @@ cdef class Loop:
 
         result = poll.stop_writing()
         if not poll.is_active():
-            self._polls_gc[fd] = poll
+            del self._polls[fd]
+            poll._close()
         return result
 
     cdef _getaddrinfo(self, object host, object port,