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

When the self-pipe socketpair of a BaseProactorEventLoop reaches a clean
EOF (e.g. the OS tears the loopback connection down across a power or
session state change on Windows), _loop_self_reading re-armed recv() on
the dead socket, which completed immediately and rescheduled the callback
forever, pinning one core at 100% CPU with nothing logged.

Detect the EOF via the empty recv result and rebuild the socketpair
instead: allocate the replacement first (so a failure leaves the previous
state untouched), re-register signal.set_wakeup_fd on the new socket
before closing the old one (mirroring close()), then arm the next read on
the new socket so cross-thread wakeups keep working.

Author: aidaodedjl

Lib/asyncio/proactor_events.py

@@ -783,10 +783,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
@@ -795,6 +815,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

Lib/test/test_asyncio/test_proactor_events.py

@@ -831,6 +831,34 @@ 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_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,7 @@
+Fix :class:`asyncio.ProactorEventLoop` spinning at 100% CPU forever when the
+event loop's self-pipe socketpair reaches EOF.  The connection can be torn
+down underneath the running process by a system power or session state
+change, but also by any in-path network filter driver re-applying its
+filters (a clean EOF, with no event logged anywhere).  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.