[3.11] gh-110205: Fix asyncio ThreadedChildWatcher._join_threads() (GH-110884) (#111413)

- `ThreadedChildWatcher.close()` is now *officially* a no-op; `_join_threads()` never did anything.
- Threads created by that class are now named `asyncio-waitpid-NNN`.
- `test.test_asyncio.utils.TestCase.close_loop()` now waits for the child watcher's threads, but not forever; if a thread hangs, it raises `RuntimeError`.
(cherry picked from commit c3bb10c)

Co-authored-by: Guido van Rossum <[email protected]>

Author: miss-islington

Lib/asyncio/unix_events.py

@@ -1363,14 +1363,7 @@ def is_active(self):
         return True
 
     def close(self):
-        self._join_threads()
-
-    def _join_threads(self):
-        """Internal: Join all non-daemon threads"""
-        threads = [thread for thread in list(self._threads.values())
-                   if thread.is_alive() and not thread.daemon]
-        for thread in threads:
-            thread.join()
+        pass
 
     def __enter__(self):
         return self
@@ -1389,7 +1382,7 @@ def __del__(self, _warn=warnings.warn):
     def add_child_handler(self, pid, callback, *args):
         loop = events.get_running_loop()
         thread = threading.Thread(target=self._do_waitpid,
-                                  name=f"waitpid-{next(self._pid_counter)}",
+                                  name=f"asyncio-waitpid-{next(self._pid_counter)}",
                                   args=(loop, pid, callback, args),
                                   daemon=True)
         self._threads[pid] = thread

Lib/test/test_asyncio/utils.py

@@ -548,6 +548,7 @@ def close_loop(loop):
             else:
                 loop._default_executor.shutdown(wait=True)
         loop.close()
+
         policy = support.maybe_get_event_loop_policy()
         if policy is not None:
             try:
@@ -557,9 +558,13 @@ def close_loop(loop):
                 pass
             else:
                 if isinstance(watcher, asyncio.ThreadedChildWatcher):
-                    threads = list(watcher._threads.values())
-                    for thread in threads:
-                        thread.join()
+                    # Wait for subprocess to finish, but not forever
+                    for thread in list(watcher._threads.values()):
+                        thread.join(timeout=support.SHORT_TIMEOUT)
+                        if thread.is_alive():
+                            raise RuntimeError(f"thread {thread} still alive: "
+                                               "subprocess still running")
+
 
     def set_event_loop(self, loop, *, cleanup=True):
         if loop is None: