diff --git a/tests/test_base.py b/tests/test_base.py index b226a84f..cbd924f6 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -773,6 +773,98 @@ def scheduler(): thread.join() self.assertEqual(counter[0], ITERATIONS) + def _poll_error_pipe(self): + # A pipe whose read end is closed: on Linux, epoll reports + # EPOLLERR for the write end. + r, w = os.pipe() + os.set_blocking(w, False) + self.addCleanup(os.close, w) + return r, w + + def _run_briefly_for(self, delay): + self.loop.call_later(delay, self.loop.stop) + self.loop.run_forever() + + @unittest.skipUnless(sys.platform.startswith('linux'), 'Linux only') + def test_add_writer_keeps_watching_after_error(self): + # https://github.com/MagicStack/uvloop/issues/787 + errors = [] + self.loop.set_exception_handler(lambda loop, ctx: errors.append(ctx)) + r, w = self._poll_error_pipe() + calls = 0 + + def writer(): + nonlocal calls + calls += 1 + if calls == 1: + os.close(r) + elif calls == 100: + self.loop.remove_writer(w) + + self.loop.add_writer(w, writer) + self._run_briefly_for(0.2) + + self.assertEqual(calls, 100) + self.assertEqual(errors, []) + + def _test_add_writer_after_error(self, remove_first): + errors = [] + self.loop.set_exception_handler(lambda loop, ctx: errors.append(ctx)) + r, w = self._poll_error_pipe() + calls = [] + + def writer(): + calls.append('call') + if len(calls) == 1: + os.close(r) + + def rearm(): + calls.append('rearm') + if remove_first: + self.loop.remove_writer(w) + self.loop.add_writer(w, writer) + + self.loop.add_writer(w, writer) + self.loop.call_later(0.1, rearm) + self._run_briefly_for(0.2) + self.loop.remove_writer(w) + + i = calls.index('rearm') + self.assertGreater(i, 3) + self.assertGreater(len(calls) - i - 1, 3) + self.assertEqual(errors, []) + + @unittest.skipUnless(sys.platform.startswith('linux'), 'Linux only') + def test_add_writer_after_error(self): + self._test_add_writer_after_error(remove_first=False) + + @unittest.skipUnless(sys.platform.startswith('linux'), 'Linux only') + def test_remove_and_add_writer_after_error(self): + self._test_add_writer_after_error(remove_first=True) + + @unittest.skipUnless(sys.platform.startswith('linux'), 'Linux only') + def test_add_writer_rearm_in_callback_after_error(self): + errors = [] + self.loop.set_exception_handler(lambda loop, ctx: errors.append(ctx)) + r, w = self._poll_error_pipe() + calls = 0 + + def writer(): + nonlocal calls + calls += 1 + if calls == 1: + os.close(r) + elif calls == 2: + self.loop.add_writer(w, writer) + elif calls == 100: + self.loop.remove_writer(w) + + self.loop.add_writer(w, writer) + self._run_briefly_for(0.2) + + self.assertEqual(calls, 100) + self.assertEqual(errors, []) + def test_freethreading(self): if not hasattr(sys, "_is_gil_enabled"): raise unittest.SkipTest("No sys._is_gil_enabled()") diff --git a/uvloop/handles/poll.pxd b/uvloop/handles/poll.pxd index c2205402..0d439084 100644 --- a/uvloop/handles/poll.pxd +++ b/uvloop/handles/poll.pxd @@ -21,5 +21,8 @@ cdef class UVPoll(UVHandle): cdef stop_writing(self) cdef stop(self) + cdef _on_poll_error(self, exc) + cdef _close_on_error(self, exc) + @staticmethod cdef UVPoll new(Loop loop, int fd) diff --git a/uvloop/handles/poll.pyx b/uvloop/handles/poll.pyx index c905e9b0..341b4e12 100644 --- a/uvloop/handles/poll.pyx +++ b/uvloop/handles/poll.pyx @@ -189,6 +189,67 @@ cdef class UVPoll(UVHandle): finally: self._close() + cdef _on_poll_error(self, exc): + # On Linux, libuv reports POLLERR as UV_EBADF after it has + # stopped the watcher. To match asyncio (level-triggered + # epoll), run the callbacks as for a ready fd and keep + # watching the fd until the callbacks are removed. + cdef: + int mask = 0 + int err + + if not self.is_active(): + self._close_on_error(None) + return + + if self.reading_handle is not None: + try: + self.reading_handle._run() + except BaseException as ex: + self._error(ex, False) + + if self.writing_handle is not None: + try: + self.writing_handle._run() + except BaseException as ex: + self._error(ex, False) + + if not self._is_alive(): + # The callbacks removed the reader and the writer, + # and the loop closed the handle. + return + + if self.reading_handle is not None: + mask |= uv.UV_READABLE + if self.writing_handle is not None: + mask |= uv.UV_WRITABLE + if mask == 0: + return + + try: + # Restarting the watcher on a closed fd would make libuv + # abort() in epoll_ctl(), so check the fd first. + os_fstat(self.fd) + except OSError as ex: + self._close_on_error(ex) + return + + err = uv.uv_poll_start( + self._handle, + mask, + __on_uvpoll_event) + if err < 0: + self._close_on_error(convert_error(err)) + + cdef _close_on_error(self, exc): + # Remove the closed handle from the loop, so that a later + # add_reader() or add_writer() creates a new one. + if self._loop._polls.get(self.fd) is self: + del self._loop._polls[self.fd] + self._close() + if exc is not None: + self._loop._handle_exception(exc) + cdef void __on_uvpoll_event( uv.uv_poll_t* handle, @@ -204,7 +265,7 @@ cdef void __on_uvpoll_event( if status < 0: exc = convert_error(status) - poll._fatal_error(exc, False) + poll._on_poll_error(exc) return if ((events & (uv.UV_READABLE | uv.UV_DISCONNECT)) and diff --git a/uvloop/includes/stdlib.pxi b/uvloop/includes/stdlib.pxi index 5fff4ad8..6ea68b34 100644 --- a/uvloop/includes/stdlib.pxi +++ b/uvloop/includes/stdlib.pxi @@ -112,6 +112,7 @@ cdef os_pipe = os.pipe cdef os_read = os.read cdef os_remove = os.remove cdef os_stat = os.stat +cdef os_fstat = os.fstat cdef os_unlink = os.unlink cdef os_fspath = os.fspath