FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

gh-156333: rebuild the proactor self-pipe on EOF instead of busy-looping by aidaodedjl · Pull Request #156343 · python/cpython · GitHub

/ cpython Public
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension .py  (3) .rst  (1) All 2 file types selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
26 changes: 24 additions & 2 deletions Lib/asyncio/proactor_events.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
28 changes: 28 additions & 0 deletions Lib/test/test_asyncio/test_proactor_events.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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')
Expand Down
55 changes: 55 additions & 0 deletions Lib/test/test_asyncio/test_windows_events.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
Fix :class:`asyncio.ProactorEventLoop` 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 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.
Loading

Back | FazBrowse Home | New Git URL