From 08c0b0f03e1f450ce6abd78a639670f52cd73008 Mon Sep 17 00:00:00 2001 From: Ayush Anand Date: Tue, 10 Nov 2026 16:52:26 +0000 Subject: [PATCH] Keep watching an fd after a poll error on Linux On Linux, libuv reports POLLERR to the poll callback as UV_EBADF after it has stopped the watcher. UVPoll handled this as a fatal error: it ran the reader and the writer once, dropped them and closed the handle, which stayed in Loop._polls. The fd was then no longer watched, a later add_writer() raised RuntimeError and re-arming from the callback did nothing. Match asyncio (level-triggered epoll) instead: run the callbacks as for a ready fd and start the poll again, so they keep being called until they are removed. If the fd is gone or the restart fails, close the handle, remove it from Loop._polls so that a later add_reader() or add_writer() creates a new one, and report the error to the exception handler. Fixes #787. --- tests/test_base.py | 92 ++++++++++++++++++++++++++++++++++++++ uvloop/handles/poll.pxd | 3 ++ uvloop/handles/poll.pyx | 63 +++++++++++++++++++++++++- uvloop/includes/stdlib.pxi | 1 + 4 files changed, 158 insertions(+), 1 deletion(-) diff --git a/tests/test_base.py b/tests/test_base.py index b226a84fd..cbd924f65 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 c22054026..0d4390841 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 c905e9b0b..341b4e129 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 5fff4ad8e..6ea68b344 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