pythongh-110205: Fix asyncio ThreadedChildWatcher._join_threads()

vstinner · vstinner · commit 33e6569e4ae9 · 2023-10-13T02:28:27.000+02:00
ThreadedChildWatcher._join_threads() now clears references to
completed threads.

test_asyncio.utils.TestCase now calls _join_threads() of the watcher,
uses SHORT_TIMEOUT to join a thread, and then raises an exception if
there are still running threads.
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
@@ -1373,12 +1373,16 @@ def is_active(self):
     def close(self):
         self._join_threads()
 
-    def _join_threads(self):
+    def _join_threads(self, timeout=None):
         """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()
+            thread.join(timeout)
+
+        # Clear references to terminated threads
+        self.threads = [thread for thread in list(self._threads.values())
+                        if thread.is_alive() and not thread.daemon]
 
     def __enter__(self):
         return self
@@ -1397,7 +1401,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
diff --git a/Lib/test/test_asyncio/utils.py b/Lib/test/test_asyncio/utils.py
@@ -546,6 +546,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,11 @@ def close_loop(loop):
                 pass
             else:
                 if isinstance(watcher, asyncio.ThreadedChildWatcher):
-                    threads = list(watcher._threads.values())
-                    for thread in threads:
-                        thread.join()
+                    watcher._join_threads(timeout=support.SHORT_TIMEOUT)
+                    threads = watcher._threads
+                    if threads:
+                        self.fail(f"watcher still has running threads: "
+                                  f"{threads}")
 
     def set_event_loop(self, loop, *, cleanup=True):
         if loop is None:
