diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py index 8350e0276046884..5c470cd7662816c 100644 --- a/Lib/asyncio/proactor_events.py +++ b/Lib/asyncio/proactor_events.py @@ -815,10 +815,30 @@ def _make_self_pipe(self): self._csock.setblocking(False) self._internal_fds += 1 + def _rebuild_self_pipe(self): + # gh-156333: the self-pipe socketpair reached EOF -- the OS tore the + # loopback connection down (e.g. across a power/session state change). + # Re-arming a read on the dead socket would busy-loop the CPU, so + # rebuild the pair instead. Build the replacement before touching the + # old sockets so a failure leaves the previous state intact, and + # re-register the wakeup fd before closing the old sockets, mirroring + # close(). + ssock, csock = socket.socketpair() + ssock.setblocking(False) + csock.setblocking(False) + if threading.current_thread() is threading.main_thread(): + # The wakeup fd was registered with the old socket. + signal.set_wakeup_fd(csock.fileno()) + self._ssock.close() + self._csock.close() + self._ssock, self._csock = ssock, csock + def _loop_self_reading(self, f=None): try: - if f is not None: - f.result() # may raise + if f is None: + data = None + else: + data = f.result() # may raise if self._self_reading_future is not f: # When we scheduled this Future, we assigned it to # _self_reading_future. If it's not there now, something has @@ -827,6 +847,8 @@ def _loop_self_reading(self, f=None): # that case stop here instead of continuing to schedule a new # iteration. return + if f is not None and not data: + self._rebuild_self_pipe() f = self._proactor.recv(self._ssock, 4096) except exceptions.CancelledError: # _close_self_pipe() has been called, stop waiting for data @@ -839,6 +861,36 @@ def _loop_self_reading(self, f=None): 'exception': exc, 'loop': self, }) + if f is not None and self._self_reading_future is f: + # The read itself failed (e.g. an aborted or reset + # overlapped operation on Windows), but the loop is still + # running and nothing else will re-arm it: report, then + # rebuild the pipe and arm a fresh read, or cross-thread + # wakeups would be lost silently for the life of the loop. + # A recovery that fails in turn is reported the same way + # and leaves the loop un-wakeable, like the original code; + # the field is reset so a later run_forever() can arm a + # read again, though that arm is not itself recovered if + # the pipe is still dead. Rebuilding gives the next read + # a fresh socketpair, so a repeat failure needs a new + # cause -- sustained exhaustion of sockets or handles + # fails the rebuild itself and ends the cycle above. + try: + self._rebuild_self_pipe() + f = self._proactor.recv(self._ssock, 4096) + except (SystemExit, KeyboardInterrupt): + raise + except BaseException as exc: + self._self_reading_future = None + self.call_exception_handler({ + 'message': ('Error on re-arming the event loop ' + 'self pipe read'), + 'exception': exc, + 'loop': self, + }) + else: + self._self_reading_future = f + f.add_done_callback(self._loop_self_reading) else: self._self_reading_future = f f.add_done_callback(self._loop_self_reading) diff --git a/Lib/test/test_asyncio/test_proactor_events.py b/Lib/test/test_asyncio/test_proactor_events.py index 2f229887ad62cc8..dd76fda1c52f51f 100644 --- a/Lib/test/test_asyncio/test_proactor_events.py +++ b/Lib/test/test_asyncio/test_proactor_events.py @@ -831,6 +831,106 @@ def test_loop_self_reading_exception(self): self.loop._loop_self_reading() self.assertTrue(self.loop.call_exception_handler.called) + def test_loop_self_reading_error_rebuilds_and_rearms(self): + # gh-156333: if the pending read on the self-pipe fails (e.g. an + # aborted overlapped operation on Windows), the loop must rebuild + # the pipe and arm a fresh read, or every later cross-thread + # wakeup would be silently dropped for the life of the loop. + fut = mock.Mock() + fut.result.side_effect = ConnectionResetError() + self.loop._self_reading_future = fut + self.loop.call_exception_handler = mock.Mock() + + new_ssock, new_csock = mock.Mock(), mock.Mock() + with mock.patch('asyncio.proactor_events.socket.socketpair', + return_value=(new_ssock, new_csock)): + with mock.patch('signal.set_wakeup_fd'): + self.loop._loop_self_reading(fut) + + self.loop.call_exception_handler.assert_called_once() + self.assertTrue(self.ssock.close.called) + self.assertIs(self.loop._ssock, new_ssock) + self.proactor.recv.assert_called_with(new_ssock, 4096) + self.assertIs(self.loop._self_reading_future, + self.proactor.recv.return_value) + (self.proactor.recv.return_value.add_done_callback + .assert_called_with(self.loop._loop_self_reading)) + + def test_loop_self_reading_error_while_closing_stops(self): + # A failed read for a future that is no longer the current one + # means the loop is closing (gh-39010): report the error but do + # not touch the sockets or arm anything. + fut = mock.Mock() + fut.result.side_effect = ConnectionResetError() + self.proactor.recv.side_effect = OSError() + self.loop.call_exception_handler = mock.Mock() + + # Patched defensively: the path under test should never reach the + # rebuild, but if the guard regresses the real signal wakeup fd + # and a real socketpair must not be touched from a test. + new_ssock, new_csock = mock.Mock(), mock.Mock() + with mock.patch('asyncio.proactor_events.socket.socketpair', + return_value=(new_ssock, new_csock)): + with mock.patch('signal.set_wakeup_fd'): + self.loop._loop_self_reading(fut) + + self.loop.call_exception_handler.assert_called_once() + self.assertFalse(self.ssock.close.called) + self.proactor.recv.assert_not_called() + + def test_loop_self_reading_recover_failure_reported_once(self): + # If rebuilding or re-arming after a failed read fails in turn, + # the failure is reported once and the field is reset so a later + # run_forever() can arm a read again. + fut = mock.Mock() + fut.result.side_effect = ConnectionResetError() + self.loop._self_reading_future = fut + self.proactor.recv.side_effect = OSError() + self.loop.call_exception_handler = mock.Mock() + + new_ssock, new_csock = mock.Mock(), mock.Mock() + with mock.patch('asyncio.proactor_events.socket.socketpair', + return_value=(new_ssock, new_csock)): + with mock.patch('signal.set_wakeup_fd'): + self.loop._loop_self_reading(fut) + + self.assertEqual(self.loop.call_exception_handler.call_count, 2) + messages = [c.args[0]['message'] + for c in self.loop.call_exception_handler.call_args_list] + self.assertIn('Error on reading from the event loop self pipe', + messages) + self.assertIn('Error on re-arming the event loop self pipe read', + messages) + self.assertIsNone(self.loop._self_reading_future) + + def test_loop_self_reading_eof_rebuilds_self_pipe(self): + # gh-156333: a clean EOF on the self-pipe (recv returns b'') must + # rebuild the socketpair instead of re-arming a read that completes + # immediately, which would busy-loop the CPU at 100%. + fut = mock.Mock() + fut.result.return_value = b'' + self.loop._self_reading_future = fut + + new_ssock, new_csock = mock.Mock(), mock.Mock() + with mock.patch('asyncio.proactor_events.socket.socketpair', + return_value=(new_ssock, new_csock)): + with mock.patch('signal.set_wakeup_fd') as m_wakeup_fd: + self.loop._loop_self_reading(fut) + + # the dead pipe is closed and replaced + self.assertTrue(self.ssock.close.called) + self.assertTrue(self.csock.close.called) + self.assertIs(self.loop._ssock, new_ssock) + self.assertIs(self.loop._csock, new_csock) + self.assertEqual(self.loop._internal_fds, 1) + # the wakeup fd is re-registered to the new socket before the old + # sockets are closed + self.assertEqual(m_wakeup_fd.call_args.args, (new_csock.fileno(),)) + # a new read is armed on the NEW socket, not the dead one + self.proactor.recv.assert_called_with(new_ssock, 4096) + self.assertIs(self.loop._self_reading_future, + self.proactor.recv.return_value) + def test_write_to_self(self): self.loop._write_to_self() self.csock.send.assert_called_with(b'\0') diff --git a/Lib/test/test_asyncio/test_windows_events.py b/Lib/test/test_asyncio/test_windows_events.py index bb4ba74f19a17f0..18bc6c609bf5bd2 100644 --- a/Lib/test/test_asyncio/test_windows_events.py +++ b/Lib/test/test_asyncio/test_windows_events.py @@ -252,6 +252,61 @@ def test_read_self_pipe_restart(self): self.close_loop(self.loop) self.assertFalse(self.loop.call_exception_handler.called) + def test_read_self_pipe_eof_rebuild(self): + # Regression test for gh-156333: if the self-pipe socketpair + # reaches a clean EOF (e.g. the OS tears down the loopback + # connection across a power/session state change), re-arming + # recv() on the dead socket completes immediately and reschedules + # _loop_self_reading forever, pinning a CPU core. The loop must + # instead rebuild the pipe. + loop = self.loop + calls = 0 + orig = loop._loop_self_reading + def counting(f=None): + nonlocal calls + calls += 1 + return orig(f) + loop._loop_self_reading = counting + + old_ssock = loop._ssock + + async def main(): + # Let the loop arm its self-pipe read first. + await asyncio.sleep(0.1) + # Graceful half-close: the read half sees a clean EOF, which + # is what an OS teardown of the loopback connection looks like. + loop._csock.shutdown(socket.SHUT_WR) + # Wait (bounded) for the rebuild instead of assuming a fixed + # delay, so a slow machine cannot fail the test spuriously. + deadline = time.monotonic() + support.LOOPBACK_TIMEOUT + while (loop._ssock is old_ssock + and time.monotonic() < deadline): + await asyncio.sleep(0.01) + # Let any (buggy) busy-loop rescheduling surface. + await asyncio.sleep(0.3) + + loop.run_until_complete(main()) + + # Without the fix, _loop_self_reading is rescheduled hundreds of + # thousands of times here; with the fix, the pipe is rebuilt and + # the loop goes back to sleep. + self.assertIsNot(loop._ssock, old_ssock) + self.assertLess(calls, 100) + + # The rebuilt pipe must still deliver cross-thread wakeups. + woke = [] + async def main2(): + threading.Thread( + target=lambda: loop.call_soon_threadsafe(woke.append, True) + ).start() + for _ in range(200): + if woke: + break + await asyncio.sleep(0.01) + loop.run_until_complete(main2()) + self.assertEqual(woke, [True]) + self.close_loop(self.loop) + def test_address_argument_type_error(self): # Regression test for https://github.com/python/cpython/issues/98793 proactor = self.loop._proactor diff --git a/Misc/NEWS.d/next/Library/2026-08-25-14-30-00.gh-issue-156333.prxNm7.rst b/Misc/NEWS.d/next/Library/2026-08-25-14-30-00.gh-issue-156333.prxNm7.rst new file mode 100644 index 000000000000000..070e37e959cdda3 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-08-25-14-30-00.gh-issue-156333.prxNm7.rst @@ -0,0 +1,10 @@ +Fix :class:`asyncio.ProactorEventLoop` spinning at 100% CPU forever when the +event loop's self-pipe socketpair reaches EOF. The loopback connection can +be torn down underneath the running process by a system power or session +state change and by other unlogged events, and nothing on the sockets +themselves reports the teardown. The loop now detects the EOF, rebuilds the +self-pipe, and re-arms the read on the new socket instead of re-arming a +read that completes immediately. A failed read on the self-pipe (for +example an aborted or reset overlapped operation on Windows) is recovered +the same way, instead of leaving the loop running with a permanently dead +wakeup channel.