rewrite test with queues

Author: deadlovelll

Lib/test/test_external_inspection.py

@@ -215,11 +215,17 @@ def _cleanup_sockets(*sockets):
 )
 
 
-def _asyncio_in_subinterpreter():
+def _asyncio_in_subinterpreter(ready, release):
+    """Park an asyncio task in a subinterpreter until released."""
     import asyncio
 
+    def wait(loop):
+        # Signal via the loop, so the task has already suspended
+        loop.call_soon_threadsafe(ready.put, None)
+        release.get()
+
     async def sub_worker():
-        await asyncio.sleep(2)
+        await asyncio.to_thread(wait, asyncio.get_running_loop())
 
     asyncio.run(sub_worker())
 
@@ -506,30 +512,36 @@ async def main():
     @requires_subinterpreters
     def test_all_awaited_by_covers_every_interpreter(self):
         # gh-158880
+        ready = interpreters.create_queue()
+        release = interpreters.create_queue()
+
         async def main_worker():
             await asyncio.sleep(SHORT_TIMEOUT)
 
         async def main():
             with InterpreterPoolExecutor() as pool:
-                loop = asyncio.get_running_loop()
-                loop.run_in_executor(pool, _asyncio_in_subinterpreter)
-                task = asyncio.create_task(main_worker(), name="main_worker")
-                self.addCleanup(task.cancel)
-                for _ in busy_retry(SHORT_TIMEOUT):
+                try:
+                    loop = asyncio.get_running_loop()
+                    loop.run_in_executor(pool, _asyncio_in_subinterpreter,
+                                         ready, release)
+                    task = asyncio.create_task(main_worker(),
+                                               name="main_worker")
+                    self.addCleanup(task.cancel)
                     await asyncio.sleep(0)
-                    stacks = [
+                    ready.get(timeout=SHORT_TIMEOUT)
+                    return [
                         [frame.funcname.rpartition(".")[2]
                          for frame in coro.call_stack]
                         for info in RemoteUnwinder(
                             os.getpid()).get_all_awaited_by()
                         for task in info.awaited_by
                         for coro in task.coroutine_stack
                     ]
-                    if ["sleep", "sub_worker"] in stacks:
-                        return stacks
+                finally:
+                    release.put(None)
 
         stacks = asyncio.run(main())
-        self.assertIn(["sleep", "sub_worker"], stacks)
+        self.assertIn(["to_thread", "sub_worker"], stacks)
         self.assertIn(["sleep", "main_worker"], stacks)
 
     @skip_if_not_supported