gh-156333: rebuild the proactor self-pipe on EOF instead of busy-looping

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)

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')

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

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.