gh-127049: fix race condition in asyncio signalling an unrelated process with ThreadedChildWatcher

Lib/asyncio/base_subprocess.py

@@ -166,6 +166,9 @@ def kill(self):
     else:
         def send_signal(self, signal):
             self._check_proc()
+            if self._returncode is not None:
+                # The process already exited
+                return
             try:
                 os.kill(self._proc.pid, signal)
             except ProcessLookupError:

Lib/asyncio/unix_events.py

@@ -219,7 +219,7 @@ async def _make_subprocess_transport(self, protocol, args, shell,
         return transp
 
     def _child_watcher_callback(self, pid, returncode, transp):
-        self.call_soon_threadsafe(transp._process_exited, returncode)
+        transp._process_exited(returncode)
 
     async def create_unix_connection(
             self, protocol_factory, path=None, *,
@@ -930,6 +930,49 @@ def add_child_handler(self, pid, callback, *args):
     def _do_waitpid(self, loop, expected_pid, callback, args):
         assert expected_pid > 0
 
+        if hasattr(os, 'waitid'):
+            # Wait for the child process using waitid() on platforms which support it.
+            # WNOWAIT is used to avoid reaping the child process, allowing the event loop to
+            # reap the child process with waitpid() later in event loop thread.
+            # This makes the reaping of the child and notification of the return code
+            # atomic with respect to the event loop thread.
+            try:
+                os.waitid(os.P_PID, expected_pid, os.WEXITED | os.WNOWAIT)
+            except ChildProcessError:
+                # The child process is already reaped
+                pass
+            if loop.is_closed():
+                # loop is already closed, reap the zombie here so that it is not leaked.
+                pid, _ = self._reap(loop, expected_pid)
+                logger.warning("Loop %r that handles pid %r is closed",
+                               loop, pid)
+            else:
+                try:
+                    loop.call_soon_threadsafe(
+                        self._reap_and_notify, loop, expected_pid,
+                        callback, args)
+                except RuntimeError:
+                    # The event loop was closed concurrently.
+                    pid, _ = self._reap(loop, expected_pid)
+                    logger.warning("Loop %r that handles pid %r is closed",
+                                   loop, pid)
+        else:
+            # Fallback for platforms that don't support waitid(): we have to
+            # reap the child here, which is racy with respect to send_signal()
+            pid, returncode = self._reap(loop, expected_pid)
+            if loop.is_closed():
+                logger.warning("Loop %r that handles pid %r is closed",
+                               loop, pid)
+            else:
+                loop.call_soon_threadsafe(callback, pid, returncode, *args)
+
+        self._threads.pop(expected_pid)
+
+    def _reap_and_notify(self, loop, expected_pid, callback, args):
+        pid, returncode = self._reap(loop, expected_pid)
+        callback(pid, returncode, *args)
+
+    def _reap(self, loop, expected_pid):
         try:
             pid, status = os.waitpid(expected_pid, 0)
         except ChildProcessError:
@@ -945,13 +988,7 @@ def _do_waitpid(self, loop, expected_pid, callback, args):
             if loop.get_debug():
                 logger.debug('process %s exited with returncode %s',
                              expected_pid, returncode)
-
-        if loop.is_closed():
-            logger.warning("Loop %r that handles pid %r is closed", loop, pid)
-        else:
-            loop.call_soon_threadsafe(callback, pid, returncode, *args)
-
-        self._threads.pop(expected_pid)
+        return pid, returncode
 
 def can_use_pidfd():
     if not hasattr(os, 'pidfd_open'):

Lib/test/test_asyncio/test_subprocess.py

@@ -965,6 +965,45 @@ def test_watcher_implementation(self):
             else:
                 self.assertIsInstance(watcher, unix_events._ThreadedChildWatcher)
 
+        @unittest.skipUnless(hasattr(os, 'waitid'), 'needs os.waitid()')
+        def test_send_signal_never_targets_reaped_pid(self):
+            # gh-127049: there must be no window between the child watcher
+            # reaping the child and asyncio publishing the exit in which
+            # send_signal() signals the freed (possibly recycled) PID.
+            reaped_pids = set()
+            stale_kills = []
+            orig_waitpid = os.waitpid
+            orig_kill = os.kill
+
+            def waitpid(pid, options):
+                res = orig_waitpid(pid, options)
+                if res[0] != 0:
+                    reaped_pids.add(res[0])
+                return res
+
+            def kill(pid, sig):
+                if pid in reaped_pids:
+                    stale_kills.append(pid)
+                    return
+                orig_kill(pid, sig)
+
+            async def run():
+                with mock.patch('os.waitpid', waitpid), \
+                        mock.patch('os.kill', kill):
+                    proc = await asyncio.create_subprocess_exec(
+                        *PROGRAM_BLOCKED)
+                    proc.kill()
+                    deadline = self.loop.time() + support.SHORT_TIMEOUT
+                    while proc.returncode is None:
+                        if self.loop.time() > deadline:
+                            self.fail('child exit was not published in time')
+                        proc.kill()
+                        await asyncio.sleep(0)
+                    await proc.wait()
+                self.assertEqual(stale_kills, [])
+
+            self.loop.run_until_complete(run())
+
 
     class SubprocessThreadedWatcherTests(SubprocessWatcherMixin,
                                          test_utils.TestCase):
@@ -978,6 +1017,35 @@ def tearDown(self):
             unix_events.can_use_pidfd = self._original_can_use_pidfd
             return super().tearDown()
 
+        @unittest.skipUnless(hasattr(os, 'waitid'), 'needs os.waitid()')
+        def test_pid_not_reaped_before_exit_published(self):
+            # gh-127049: the watcher thread must wait for the child
+            # without reaping it: the PID must stay reserved until the
+            # event loop thread reaps it and publishes the exit as one
+            # atomic step.
+            async def run():
+                proc = await asyncio.create_subprocess_exec(
+                    *PROGRAM_BLOCKED)
+                thread = self.loop._watcher._threads.get(proc.pid)
+                self.assertIsNotNone(thread)
+                proc.kill()
+                # Wait for the watcher thread to observe the exit while
+                # the event loop cannot process the notification yet.
+                thread.join(support.SHORT_TIMEOUT)
+                self.assertFalse(thread.is_alive())
+                # The exit has not been published yet...
+                self.assertIsNone(proc.returncode)
+                # ...so the child must still be an unreaped zombie and
+                # signalling its PID must still be safe.  waitid() raises
+                # ChildProcessError if the PID was already reaped (and
+                # possibly recycled by the kernel).
+                os.waitid(os.P_PID, proc.pid,
+                          os.WEXITED | os.WNOWAIT | os.WNOHANG)
+                proc.kill()
+                self.assertEqual(await proc.wait(), -signal.SIGKILL)
+
+            self.loop.run_until_complete(run())
+
     @unittest.skipUnless(
         unix_events.can_use_pidfd(),
         "operating system does not support pidfds",

Misc/NEWS.d/next/Library/2026-07-14-15-00-00.gh-issue-127049.pQw7Rk.rst

@@ -0,0 +1,5 @@
+Fix a race condition in :mod:`asyncio` on Unix where
+:meth:`asyncio.subprocess.Process.send_signal`, :meth:`~asyncio.subprocess.Process.terminate`
+or :meth:`~asyncio.subprocess.Process.kill` could signal an unrelated process
+that was recycled onto the PID of the already-reaped child when ThreadedChildWatcher is used.
+Patch by Kumar Aditya.