Skip to content

Commit 2875d1d

Browse files
gh-127049: fix race condition in asyncio signalling an unrelated process with ThreadedChildWatcher (#153810)
1 parent f389b03 commit 2875d1d

4 files changed

Lines changed: 121 additions & 8 deletions

File tree

Lib/asyncio/base_subprocess.py

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,9 @@ def kill(self):
165165
else:
166166
def send_signal(self, signal):
167167
self._check_proc()
168+
if self._returncode is not None:
169+
# The process already exited
170+
return
168171
try:
169172
os.kill(self._proc.pid, signal)
170173
except ProcessLookupError:

Lib/asyncio/unix_events.py

Lines changed: 45 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -219,7 +219,7 @@ async def _make_subprocess_transport(self, protocol, args, shell,
219219
return transp
220220

221221
def _child_watcher_callback(self, pid, returncode, transp):
222-
self.call_soon_threadsafe(transp._process_exited, returncode)
222+
transp._process_exited(returncode)
223223

224224
async def create_unix_connection(
225225
self, protocol_factory, path=None, *,
@@ -930,6 +930,49 @@ def add_child_handler(self, pid, callback, *args):
930930
def _do_waitpid(self, loop, expected_pid, callback, args):
931931
assert expected_pid > 0
932932

933+
if hasattr(os, 'waitid'):
934+
# Wait for the child process using waitid() on platforms which support it.
935+
# WNOWAIT is used to avoid reaping the child process, allowing the event loop to
936+
# reap the child process with waitpid() later in event loop thread.
937+
# This makes the reaping of the child and notification of the return code
938+
# atomic with respect to the event loop thread.
939+
try:
940+
os.waitid(os.P_PID, expected_pid, os.WEXITED | os.WNOWAIT)
941+
except ChildProcessError:
942+
# The child process is already reaped
943+
pass
944+
if loop.is_closed():
945+
# loop is already closed, reap the zombie here so that it is not leaked.
946+
pid, _ = self._reap(loop, expected_pid)
947+
logger.warning("Loop %r that handles pid %r is closed",
948+
loop, pid)
949+
else:
950+
try:
951+
loop.call_soon_threadsafe(
952+
self._reap_and_notify, loop, expected_pid,
953+
callback, args)
954+
except RuntimeError:
955+
# The event loop was closed concurrently.
956+
pid, _ = self._reap(loop, expected_pid)
957+
logger.warning("Loop %r that handles pid %r is closed",
958+
loop, pid)
959+
else:
960+
# Fallback for platforms that don't support waitid(): we have to
961+
# reap the child here, which is racy with respect to send_signal()
962+
pid, returncode = self._reap(loop, expected_pid)
963+
if loop.is_closed():
964+
logger.warning("Loop %r that handles pid %r is closed",
965+
loop, pid)
966+
else:
967+
loop.call_soon_threadsafe(callback, pid, returncode, *args)
968+
969+
self._threads.pop(expected_pid)
970+
971+
def _reap_and_notify(self, loop, expected_pid, callback, args):
972+
pid, returncode = self._reap(loop, expected_pid)
973+
callback(pid, returncode, *args)
974+
975+
def _reap(self, loop, expected_pid):
933976
try:
934977
pid, status = os.waitpid(expected_pid, 0)
935978
except ChildProcessError:
@@ -945,13 +988,7 @@ def _do_waitpid(self, loop, expected_pid, callback, args):
945988
if loop.get_debug():
946989
logger.debug('process %s exited with returncode %s',
947990
expected_pid, returncode)
948-
949-
if loop.is_closed():
950-
logger.warning("Loop %r that handles pid %r is closed", loop, pid)
951-
else:
952-
loop.call_soon_threadsafe(callback, pid, returncode, *args)
953-
954-
self._threads.pop(expected_pid)
991+
return pid, returncode
955992

956993
def can_use_pidfd():
957994
if not hasattr(os, 'pidfd_open'):

Lib/test/test_asyncio/test_subprocess.py

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -974,6 +974,45 @@ def test_watcher_implementation(self):
974974
else:
975975
self.assertIsInstance(watcher, unix_events._ThreadedChildWatcher)
976976

977+
@unittest.skipUnless(hasattr(os, 'waitid'), 'needs os.waitid()')
978+
def test_send_signal_never_targets_reaped_pid(self):
979+
# gh-127049: there must be no window between the child watcher
980+
# reaping the child and asyncio publishing the exit in which
981+
# send_signal() signals the freed (possibly recycled) PID.
982+
reaped_pids = set()
983+
stale_kills = []
984+
orig_waitpid = os.waitpid
985+
orig_kill = os.kill
986+
987+
def waitpid(pid, options):
988+
res = orig_waitpid(pid, options)
989+
if res[0] != 0:
990+
reaped_pids.add(res[0])
991+
return res
992+
993+
def kill(pid, sig):
994+
if pid in reaped_pids:
995+
stale_kills.append(pid)
996+
return
997+
orig_kill(pid, sig)
998+
999+
async def run():
1000+
with mock.patch('os.waitpid', waitpid), \
1001+
mock.patch('os.kill', kill):
1002+
proc = await asyncio.create_subprocess_exec(
1003+
*PROGRAM_BLOCKED)
1004+
proc.kill()
1005+
deadline = self.loop.time() + support.SHORT_TIMEOUT
1006+
while proc.returncode is None:
1007+
if self.loop.time() > deadline:
1008+
self.fail('child exit was not published in time')
1009+
proc.kill()
1010+
await asyncio.sleep(0)
1011+
await proc.wait()
1012+
self.assertEqual(stale_kills, [])
1013+
1014+
self.loop.run_until_complete(run())
1015+
9771016

9781017
class SubprocessThreadedWatcherTests(SubprocessWatcherMixin,
9791018
test_utils.TestCase):
@@ -987,6 +1026,35 @@ def tearDown(self):
9871026
unix_events.can_use_pidfd = self._original_can_use_pidfd
9881027
return super().tearDown()
9891028

1029+
@unittest.skipUnless(hasattr(os, 'waitid'), 'needs os.waitid()')
1030+
def test_pid_not_reaped_before_exit_published(self):
1031+
# gh-127049: the watcher thread must wait for the child
1032+
# without reaping it: the PID must stay reserved until the
1033+
# event loop thread reaps it and publishes the exit as one
1034+
# atomic step.
1035+
async def run():
1036+
proc = await asyncio.create_subprocess_exec(
1037+
*PROGRAM_BLOCKED)
1038+
thread = self.loop._watcher._threads.get(proc.pid)
1039+
self.assertIsNotNone(thread)
1040+
proc.kill()
1041+
# Wait for the watcher thread to observe the exit while
1042+
# the event loop cannot process the notification yet.
1043+
thread.join(support.SHORT_TIMEOUT)
1044+
self.assertFalse(thread.is_alive())
1045+
# The exit has not been published yet...
1046+
self.assertIsNone(proc.returncode)
1047+
# ...so the child must still be an unreaped zombie and
1048+
# signalling its PID must still be safe. waitid() raises
1049+
# ChildProcessError if the PID was already reaped (and
1050+
# possibly recycled by the kernel).
1051+
os.waitid(os.P_PID, proc.pid,
1052+
os.WEXITED | os.WNOWAIT | os.WNOHANG)
1053+
proc.kill()
1054+
self.assertEqual(await proc.wait(), -signal.SIGKILL)
1055+
1056+
self.loop.run_until_complete(run())
1057+
9901058
@unittest.skipUnless(
9911059
unix_events.can_use_pidfd(),
9921060
"operating system does not support pidfds",
Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
Fix a race condition in :mod:`asyncio` on Unix where
2+
:meth:`asyncio.subprocess.Process.send_signal`, :meth:`~asyncio.subprocess.Process.terminate`
3+
or :meth:`~asyncio.subprocess.Process.kill` could signal an unrelated process
4+
that was recycled onto the PID of the already-reaped child when ThreadedChildWatcher is used.
5+
Patch by Kumar Aditya.

0 commit comments

Comments
 (0)