gh-105288: wake up exit waiters after sub-process exits

Lib/asyncio/base_subprocess.py

@@ -235,6 +235,14 @@ def _process_exited(self, returncode):
         self._call(self._protocol.process_exited)
 
         self._try_finish()
+        self._wakeup_exit_waiters()
+
+    def _wakeup_exit_waiters(self):
+        # wake up futures waiting for wait()
+        for waiter in self._exit_waiters:
+            if not waiter.cancelled():
+                waiter.set_result(self._returncode)
+        self._exit_waiters = []
 
     async def _wait(self):
         """Wait until the process exit and return the process return code.
@@ -260,11 +268,7 @@ def _call_connection_lost(self, exc):
         try:
             self._protocol.connection_lost(exc)
         finally:
-            # wake up futures waiting for wait()
-            for waiter in self._exit_waiters:
-                if not waiter.cancelled():
-                    waiter.set_result(self._returncode)
-            self._exit_waiters = None
+            self._wakeup_exit_waiters()
             self._loop = None
             self._proc = None
             self._protocol = None

Lib/test/test_asyncio/test_subprocess.py

@@ -1,6 +1,8 @@
 import os
 import signal
 import sys
+import threading
+import time
 import textwrap
 import unittest
 import warnings
@@ -847,6 +849,35 @@ async def main() -> None:
         for fd in fds:
             self.assertEqual(per_fd_events[fd], per_fd_expected, (fd, events))
 
+    def test_subprocess_wait_not_hang_gh105288(self):
+        # See https://github.com/python/cpython/issues/105288
+        TIMEOUT = 5
+        finished = False
+
+        async def whoami():
+            proc = await subprocess.create_subprocess_exec('whoami')
+            await proc.communicate()
+
+        async def main():
+            t = asyncio.create_task(whoami())
+            await asyncio.wait([0, t])
+
+        def _main():
+            nonlocal finished
+            try:
+                # We have to run the event loop in a separate thread,
+                # to make sure we can fail in the main thread
+                # and successfully notify the test driver.
+                self.loop.run_until_complete(main())
+            finally:
+                finished = True
+
+
+        threading.Thread(target=_main).start()
+        time.sleep(TIMEOUT)
+        if not finished:
+            self.fail('watchdog timeout, event loop is probably sleeping infinitely')
+
     def test_subprocess_communicate_stdout(self):
         # See https://github.com/python/cpython/issues/100133
         async def get_command_stdout(cmd, *args):

Misc/NEWS.d/next/Library/2023-06-05-02-48-08.gh-issue-105288.Ewnyas.rst

@@ -0,0 +1,4 @@
+Exit waiters are not resolved after subprocess handled by :class:`asyncio.Task`
+exits. This will causes a recent :func:`asyncio.wait` call hang forever if
+there are objects that are not :class:`typing.Awaitable` yield from its first
+parameter.