diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py index 16d5c1b6f0a3e19..756d397bd9b14b0 100644 --- a/Lib/asyncio/selector_events.py +++ b/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 diff --git a/Lib/test/test_asyncio/test_selector_events.py b/Lib/test/test_asyncio/test_selector_events.py index a323084d262ebfe..953aa8ee5552d8f 100644 --- a/Lib/test/test_asyncio/test_selector_events.py +++ b/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(): diff --git a/Misc/NEWS.d/next/Library/2026-08-25-16-20-00.gh-issue-156344.Qm4vTz.rst b/Misc/NEWS.d/next/Library/2026-08-25-16-20-00.gh-issue-156344.Qm4vTz.rst new file mode 100644 index 000000000000000..b8c68e1cca3240f --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-08-25-16-20-00.gh-issue-156344.Qm4vTz.rst @@ -0,0 +1,8 @@ +Fix :class:`asyncio.SelectorEventLoop` 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 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.