Skip to content

Commit 1bf1bb1

Browse files
author
aidaodedjl
committed
gh-156344: rebuild the selector self-pipe on EOF instead of busy-looping
When the self-pipe socketpair of a BaseSelectorEventLoop reaches a clean EOF (e.g. the OS tears the connection down across a power or session state change on Windows), _read_from_self broke out of its read loop but left the reader registered on the dead socket. A closed-for-read socket is permanently readable, so every select() iteration re-fired the callback: one core pinned at 100% CPU with nothing logged, measured at 582k callback invocations during a 3-second idle sleep. Rebuild the pair instead: allocate the replacement before touching the old sockets so an allocation failure leaves the previous state intact, move the process-wide signal wakeup fd to the new socket when (and only when) it is registered on our _csock -- restoring foreign registrations untouched, and keeping the old write end open when it cannot be moved from a worker thread -- then remove the old reader, close the old sockets, and register the reader on the new socket.
1 parent 7b4364d commit 1bf1bb1

3 files changed

Lines changed: 167 additions & 1 deletion

File tree

Lib/asyncio/selector_events.py

Lines changed: 47 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,9 @@
1212
import itertools
1313
import os
1414
import selectors
15+
import signal
1516
import socket
17+
import threading
1618
import warnings
1719
import weakref
1820
try:
@@ -124,6 +126,49 @@ def _make_self_pipe(self):
124126
self._internal_fds += 1
125127
self._add_reader(self._ssock.fileno(), self._read_from_self)
126128

129+
def _rebuild_self_pipe(self):
130+
# gh-156344: the self-pipe socketpair reached EOF -- the OS tore the
131+
# connection down (e.g. across a power/session state change on
132+
# Windows). A closed-for-read socket is permanently readable, so
133+
# the registered reader would re-fire on every select() iteration,
134+
# busy-looping the CPU. Rebuild the pair instead: allocate the
135+
# replacement before touching the old sockets, so an allocation
136+
# failure leaves the previous state intact.
137+
ssock, csock = socket.socketpair()
138+
try:
139+
ssock.setblocking(False)
140+
csock.setblocking(False)
141+
old_ssock = self._ssock
142+
old_csock = self._csock
143+
keep_old_csock = False
144+
if getattr(self, '_signal_handlers', None):
145+
# The Unix mixin registers the process-wide wakeup fd on
146+
# _csock in add_signal_handler(). set_wakeup_fd() returns
147+
# the previous fd: move ours to the new socket, but restore
148+
# anything that belongs to someone else. A signal arriving
149+
# within this window can be lost -- the same unavoidable
150+
# window close() has.
151+
if threading.current_thread() is threading.main_thread():
152+
prev = signal.set_wakeup_fd(csock.fileno())
153+
if prev != old_csock.fileno():
154+
signal.set_wakeup_fd(prev)
155+
else:
156+
# The wakeup fd cannot be moved from a worker thread;
157+
# keep the old write end open so signal delivery keeps
158+
# working (one leaked socket beats a process-wide wakeup
159+
# fd writing into whatever reuses the number).
160+
keep_old_csock = True
161+
except BaseException:
162+
ssock.close()
163+
csock.close()
164+
raise
165+
self._remove_reader(old_ssock.fileno())
166+
old_ssock.close()
167+
if not keep_old_csock:
168+
old_csock.close()
169+
self._ssock, self._csock = ssock, csock
170+
self._add_reader(self._ssock.fileno(), self._read_from_self)
171+
127172
def _process_self_data(self, data):
128173
pass
129174

@@ -132,7 +177,8 @@ def _read_from_self(self):
132177
try:
133178
data = self._ssock.recv(4096)
134179
if not data:
135-
break
180+
self._rebuild_self_pipe()
181+
return
136182
self._process_self_data(data)
137183
except InterruptedError:
138184
continue

Lib/test/test_asyncio/test_selector_events.py

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import collections
44
import errno
55
import selectors
6+
import signal
67
import socket
78
import sys
89
import unittest
@@ -150,6 +151,119 @@ def test_read_from_self_exception(self):
150151
self.loop._ssock.recv.side_effect = OSError
151152
self.assertRaises(OSError, self.loop._read_from_self)
152153

154+
def test_read_from_self_eof_rebuilds_self_pipe(self):
155+
# gh-156344: a clean EOF (recv returns b'') must rebuild the
156+
# socketpair instead of leaving the reader registered on a socket
157+
# that is readable forever, which would busy-loop the CPU at 100%.
158+
loop = self.loop
159+
old_ssock = loop._ssock
160+
loop._ssock.recv.return_value = b''
161+
loop._remove_reader = mock.Mock()
162+
loop._add_reader = mock.Mock()
163+
with mock.patch('asyncio.selector_events.socket.socketpair',
164+
return_value=(mock.Mock(), mock.Mock())) as socketpair:
165+
self.assertIsNone(loop._read_from_self())
166+
self.assertTrue(socketpair.called)
167+
self.assertIsNot(loop._ssock, old_ssock)
168+
loop._remove_reader.assert_called_with(old_ssock.fileno())
169+
loop._add_reader.assert_called_with(loop._ssock.fileno(),
170+
loop._read_from_self)
171+
172+
def test_read_from_self_blocking_is_not_eof(self):
173+
# gh-156344: only a clean EOF triggers the rebuild -- a would-block
174+
# read must not.
175+
self.loop._ssock.recv.side_effect = BlockingIOError
176+
with mock.patch('asyncio.selector_events.socket.socketpair') as sp:
177+
self.assertIsNone(self.loop._read_from_self())
178+
self.assertFalse(sp.called)
179+
180+
def test_self_pipe_eof_rebuild_functional(self):
181+
# gh-156344 functional test on a real selector loop: kill the
182+
# self-pipe with a graceful half-close and verify the pair is
183+
# rebuilt, the reader lives only on the new fd, and wakeups keep
184+
# working through the new pair.
185+
loop = selector_events.BaseSelectorEventLoop()
186+
self.addCleanup(loop.close)
187+
old_ssock = loop._ssock
188+
old_fd = old_ssock.fileno()
189+
old_csock = loop._csock
190+
191+
old_csock.shutdown(socket.SHUT_WR)
192+
loop._read_from_self()
193+
194+
# pair rebuilt and reader registered on the new fd only
195+
self.assertIsNot(loop._ssock, old_ssock)
196+
self.assertNotEqual(loop._ssock.fileno(), old_fd)
197+
self.assertNotIn(old_fd, loop._selector.get_map())
198+
self.assertIn(loop._ssock.fileno(), loop._selector.get_map())
199+
200+
# wakeups through the new pair still work
201+
loop._write_to_self()
202+
data = loop._ssock.recv(4096)
203+
self.assertEqual(data, b'\0')
204+
205+
@mock.patch('asyncio.selector_events.socket.socketpair')
206+
def test_rebuild_self_pipe_moves_wakeup_fd(self, socketpair):
207+
# gh-156344: on Unix the wakeup fd registered by add_signal_handler()
208+
# names _csock; a rebuild must move it to the new socket and must not
209+
# touch a registration owned by someone else.
210+
loop = self.loop
211+
old_ssock, old_csock = loop._ssock, loop._csock
212+
loop._remove_reader = mock.Mock()
213+
loop._add_reader = mock.Mock()
214+
215+
new_ssock, new_csock = mock.Mock(), mock.Mock()
216+
socketpair.return_value = (new_ssock, new_csock)
217+
218+
# Simulate the Unix mixin's signal state: _signal_handlers non-empty
219+
# and the wakeup fd naming our _csock.
220+
loop._signal_handlers = {signal.SIGINT: mock.Mock()}
221+
with mock.patch('asyncio.selector_events.signal.set_wakeup_fd',
222+
return_value=old_csock.fileno()) as m_wakeup_fd:
223+
loop._rebuild_self_pipe()
224+
# moved: new fd registered, old one not re-registered
225+
self.assertEqual(m_wakeup_fd.call_args_list,
226+
[mock.call(new_csock.fileno())])
227+
# old reader removed, old sockets closed, reader re-armed on the new
228+
loop._remove_reader.assert_called_with(old_ssock.fileno())
229+
self.assertTrue(old_ssock.close.called)
230+
self.assertTrue(old_csock.close.called)
231+
self.assertIs(loop._ssock, new_ssock)
232+
self.assertIs(loop._csock, new_csock)
233+
loop._add_reader.assert_called_with(new_ssock.fileno(),
234+
loop._read_from_self)
235+
236+
@mock.patch('asyncio.selector_events.socket.socketpair')
237+
def test_rebuild_self_pipe_leaves_foreign_wakeup_fd(self, socketpair):
238+
# set_wakeup_fd returned a fd that is not ours: it belongs to someone
239+
# else and must be restored untouched.
240+
loop = self.loop
241+
loop._remove_reader = mock.Mock()
242+
loop._add_reader = mock.Mock()
243+
new_ssock, new_csock = mock.Mock(), mock.Mock()
244+
socketpair.return_value = (new_ssock, new_csock)
245+
246+
loop._signal_handlers = {signal.SIGINT: mock.Mock()}
247+
with mock.patch('asyncio.selector_events.signal.set_wakeup_fd',
248+
return_value=999) as m_wakeup_fd:
249+
loop._rebuild_self_pipe()
250+
self.assertEqual(m_wakeup_fd.call_args_list,
251+
[mock.call(new_csock.fileno()),
252+
mock.call(999)])
253+
254+
@mock.patch('asyncio.selector_events.socket.socketpair')
255+
def test_rebuild_self_pipe_no_signals(self, socketpair):
256+
# Without add_signal_handler() state the wakeup fd is untouched.
257+
self.loop._remove_reader = mock.Mock()
258+
self.loop._add_reader = mock.Mock()
259+
new_ssock, new_csock = mock.Mock(), mock.Mock()
260+
socketpair.return_value = (new_ssock, new_csock)
261+
262+
with mock.patch('asyncio.selector_events.signal.set_wakeup_fd',
263+
return_value=self.loop._csock.fileno()) as m_wakeup_fd:
264+
self.loop._rebuild_self_pipe()
265+
self.assertFalse(m_wakeup_fd.called)
266+
153267
def test_write_to_self_tryagain(self):
154268
self.loop._csock.send.side_effect = BlockingIOError
155269
with test_utils.disable_logger():
Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
Fix :class:`asyncio.SelectorEventLoop` spinning at 100% CPU forever when the
2+
event loop's self-pipe socketpair reaches EOF, which can happen when Windows
3+
tears down the idle loopback connection across a power or session state
4+
change. The loop now rebuilds the self-pipe (moving any registered signal
5+
wakeup fd to the new socket) and re-registers the reader on the new socket
6+
instead of leaving it on a socket that is readable forever.

0 commit comments

Comments
 (0)