gh-156344: rebuild the selector self-pipe on EOF instead of busy-looping

Lib/asyncio/selector_events.py

@@ -12,7 +12,9 @@
 import itertools
 import os
 import selectors
+import signal
 import socket
+import threading
 import warnings
 import weakref
 try:
@@ -124,6 +126,49 @@ def _make_self_pipe(self):
         self._internal_fds += 1
         self._add_reader(self._ssock.fileno(), self._read_from_self)
 
+    def _rebuild_self_pipe(self):
+        # gh-156344: the self-pipe socketpair reached EOF -- the OS tore the
+        # connection down (e.g. across a power/session state change on
+        # Windows).  A closed-for-read socket is permanently readable, so
+        # the registered reader would re-fire on every select() iteration,
+        # busy-looping the CPU.  Rebuild the pair instead: allocate the
+        # replacement before touching the old sockets, so an allocation
+        # failure leaves the previous state intact.
+        ssock, csock = socket.socketpair()
+        try:
+            ssock.setblocking(False)
+            csock.setblocking(False)
+            old_ssock = self._ssock
+            old_csock = self._csock
+            keep_old_csock = False
+            if getattr(self, '_signal_handlers', None):
+                # The Unix mixin registers the process-wide wakeup fd on
+                # _csock in add_signal_handler().  set_wakeup_fd() returns
+                # the previous fd: move ours to the new socket, but restore
+                # anything that belongs to someone else.  A signal arriving
+                # within this window can be lost -- the same unavoidable
+                # window close() has.
+                if threading.current_thread() is threading.main_thread():
+                    prev = signal.set_wakeup_fd(csock.fileno())
+                    if prev != old_csock.fileno():
+                        signal.set_wakeup_fd(prev)
+                else:
+                    # The wakeup fd cannot be moved from a worker thread;
+                    # keep the old write end open so signal delivery keeps
+                    # working (one leaked socket beats a process-wide wakeup
+                    # fd writing into whatever reuses the number).
+                    keep_old_csock = True
+        except BaseException:
+            ssock.close()
+            csock.close()
+            raise
+        self._remove_reader(old_ssock.fileno())
+        old_ssock.close()
+        if not keep_old_csock:
+            old_csock.close()
+        self._ssock, self._csock = ssock, csock
+        self._add_reader(self._ssock.fileno(), self._read_from_self)
+
     def _process_self_data(self, data):
         pass
 
@@ -132,7 +177,8 @@ def _read_from_self(self):
             try:
                 data = self._ssock.recv(4096)
                 if not data:
-                    break
+                    self._rebuild_self_pipe()
+                    return
                 self._process_self_data(data)
             except InterruptedError:
                 continue

Lib/test/test_asyncio/test_selector_events.py

@@ -3,6 +3,7 @@
 import collections
 import errno
 import selectors
+import signal
 import socket
 import sys
 import unittest
@@ -150,6 +151,119 @@ def test_read_from_self_exception(self):
         self.loop._ssock.recv.side_effect = OSError
         self.assertRaises(OSError, self.loop._read_from_self)
 
+    def test_read_from_self_eof_rebuilds_self_pipe(self):
+        # gh-156344: a clean EOF (recv returns b'') must rebuild the
+        # socketpair instead of leaving the reader registered on a socket
+        # that is readable forever, which would busy-loop the CPU at 100%.
+        loop = self.loop
+        old_ssock = loop._ssock
+        loop._ssock.recv.return_value = b''
+        loop._remove_reader = mock.Mock()
+        loop._add_reader = mock.Mock()
+        with mock.patch('asyncio.selector_events.socket.socketpair',
+                        return_value=(mock.Mock(), mock.Mock())) as socketpair:
+            self.assertIsNone(loop._read_from_self())
+            self.assertTrue(socketpair.called)
+        self.assertIsNot(loop._ssock, old_ssock)
+        loop._remove_reader.assert_called_with(old_ssock.fileno())
+        loop._add_reader.assert_called_with(loop._ssock.fileno(),
+                                            loop._read_from_self)
+
+    def test_read_from_self_blocking_is_not_eof(self):
+        # gh-156344: only a clean EOF triggers the rebuild -- a would-block
+        # read must not.
+        self.loop._ssock.recv.side_effect = BlockingIOError
+        with mock.patch('asyncio.selector_events.socket.socketpair') as sp:
+            self.assertIsNone(self.loop._read_from_self())
+            self.assertFalse(sp.called)
+
+    def test_self_pipe_eof_rebuild_functional(self):
+        # gh-156344 functional test on a real selector loop: kill the
+        # self-pipe with a graceful half-close and verify the pair is
+        # rebuilt, the reader lives only on the new fd, and wakeups keep
+        # working through the new pair.
+        loop = selector_events.BaseSelectorEventLoop()
+        self.addCleanup(loop.close)
+        old_ssock = loop._ssock
+        old_fd = old_ssock.fileno()
+        old_csock = loop._csock
+
+        old_csock.shutdown(socket.SHUT_WR)
+        loop._read_from_self()
+
+        # pair rebuilt and reader registered on the new fd only
+        self.assertIsNot(loop._ssock, old_ssock)
+        self.assertNotEqual(loop._ssock.fileno(), old_fd)
+        self.assertNotIn(old_fd, loop._selector.get_map())
+        self.assertIn(loop._ssock.fileno(), loop._selector.get_map())
+
+        # wakeups through the new pair still work
+        loop._write_to_self()
+        data = loop._ssock.recv(4096)
+        self.assertEqual(data, b'\0')
+
+    @mock.patch('asyncio.selector_events.socket.socketpair')
+    def test_rebuild_self_pipe_moves_wakeup_fd(self, socketpair):
+        # gh-156344: on Unix the wakeup fd registered by add_signal_handler()
+        # names _csock; a rebuild must move it to the new socket and must not
+        # touch a registration owned by someone else.
+        loop = self.loop
+        old_ssock, old_csock = loop._ssock, loop._csock
+        loop._remove_reader = mock.Mock()
+        loop._add_reader = mock.Mock()
+
+        new_ssock, new_csock = mock.Mock(), mock.Mock()
+        socketpair.return_value = (new_ssock, new_csock)
+
+        # Simulate the Unix mixin's signal state: _signal_handlers non-empty
+        # and the wakeup fd naming our _csock.
+        loop._signal_handlers = {signal.SIGINT: mock.Mock()}
+        with mock.patch('asyncio.selector_events.signal.set_wakeup_fd',
+                        return_value=old_csock.fileno()) as m_wakeup_fd:
+            loop._rebuild_self_pipe()
+        # moved: new fd registered, old one not re-registered
+        self.assertEqual(m_wakeup_fd.call_args_list,
+                         [mock.call(new_csock.fileno())])
+        # old reader removed, old sockets closed, reader re-armed on the new
+        loop._remove_reader.assert_called_with(old_ssock.fileno())
+        self.assertTrue(old_ssock.close.called)
+        self.assertTrue(old_csock.close.called)
+        self.assertIs(loop._ssock, new_ssock)
+        self.assertIs(loop._csock, new_csock)
+        loop._add_reader.assert_called_with(new_ssock.fileno(),
+                                            loop._read_from_self)
+
+    @mock.patch('asyncio.selector_events.socket.socketpair')
+    def test_rebuild_self_pipe_leaves_foreign_wakeup_fd(self, socketpair):
+        # set_wakeup_fd returned a fd that is not ours: it belongs to someone
+        # else and must be restored untouched.
+        loop = self.loop
+        loop._remove_reader = mock.Mock()
+        loop._add_reader = mock.Mock()
+        new_ssock, new_csock = mock.Mock(), mock.Mock()
+        socketpair.return_value = (new_ssock, new_csock)
+
+        loop._signal_handlers = {signal.SIGINT: mock.Mock()}
+        with mock.patch('asyncio.selector_events.signal.set_wakeup_fd',
+                        return_value=999) as m_wakeup_fd:
+            loop._rebuild_self_pipe()
+        self.assertEqual(m_wakeup_fd.call_args_list,
+                         [mock.call(new_csock.fileno()),
+                          mock.call(999)])
+
+    @mock.patch('asyncio.selector_events.socket.socketpair')
+    def test_rebuild_self_pipe_no_signals(self, socketpair):
+        # Without add_signal_handler() state the wakeup fd is untouched.
+        self.loop._remove_reader = mock.Mock()
+        self.loop._add_reader = mock.Mock()
+        new_ssock, new_csock = mock.Mock(), mock.Mock()
+        socketpair.return_value = (new_ssock, new_csock)
+
+        with mock.patch('asyncio.selector_events.signal.set_wakeup_fd',
+                        return_value=self.loop._csock.fileno()) as m_wakeup_fd:
+            self.loop._rebuild_self_pipe()
+        self.assertFalse(m_wakeup_fd.called)
+
     def test_write_to_self_tryagain(self):
         self.loop._csock.send.side_effect = BlockingIOError
         with test_utils.disable_logger():

Misc/NEWS.d/next/Library/2026-08-25-16-20-00.gh-issue-156344.Qm4vTz.rst

@@ -0,0 +1,6 @@
+Fix :class:`asyncio.SelectorEventLoop` spinning at 100% CPU forever when the
+event loop's self-pipe socketpair reaches EOF, which can happen when Windows
+tears down the idle loopback connection across a power or session state
+change.  The loop now rebuilds the self-pipe (moving any registered signal
+wakeup fd to the new socket) and re-registers the reader on the new socket
+instead of leaving it on a socket that is readable forever.