gh-158880: Fix asyncio ps showing tasks from only one interpreter

Lib/test/test_external_inspection.py

@@ -46,6 +46,7 @@
 
 try:
     from concurrent import interpreters
+    from concurrent.futures import InterpreterPoolExecutor
 except ImportError:
     interpreters = None
 
@@ -214,6 +215,15 @@ def _cleanup_sockets(*sockets):
 )
 
 
+def _asyncio_in_subinterpreter():
+    import asyncio
+
+    async def sub_worker():
+        await asyncio.sleep(2)
+
+    asyncio.run(sub_worker())
+
+
 def requires_subinterpreters(meth):
     """Decorator to skip a test if subinterpreters are not supported."""
     return unittest.skipIf(interpreters is None, "subinterpreters required")(
@@ -492,6 +502,36 @@ async def main():
         self.assertIn(main_name, names)
         self.assertEqual([len(n) for n in names if n.startswith("x")], [255])
 
+    @skip_if_not_supported
+    @requires_subinterpreters
+    def test_all_awaited_by_covers_every_interpreter(self):
+        # gh-158880
+        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):
+                    await asyncio.sleep(0)
+                    stacks = [
+                        [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
+
+        stacks = asyncio.run(main())
+        self.assertIn(["sleep", "sub_worker"], stacks)
+        self.assertIn(["sleep", "main_worker"], stacks)
+
     @skip_if_not_supported
     def test_recursive_coroutine_stack_is_not_truncated(self):
         # gh-158522

Misc/NEWS.d/next/Library/2026-10-05-23-40-53.gh-issue-158880.8imLqO.rst

@@ -0,0 +1,2 @@
+Fix ``python -m asyncio ps`` and ``pstree`` showing tasks from only one
+interpreter. Patch by Timofei Ivankov.

Modules/_remote_debugging/_remote_debugging.h

@@ -663,6 +663,7 @@ extern int collect_frames_with_cache(
 
 extern int iterate_threads(
     RemoteUnwinderObject *unwinder,
+    uintptr_t interpreter_addr,
     thread_processor_func processor,
     void *context
 );

Modules/_remote_debugging/module.c

@@ -958,7 +958,9 @@ _remote_debugging_RemoteUnwinder_get_all_awaited_by_impl(RemoteUnwinderObject *s
     if (ensure_async_debug_offsets(self) < 0) {
         return NULL;
     }
-    if (refresh_generation_caches_for_interpreter(self, self->interpreter_addr) < 0) {
+    PyObject *seen = PySet_New(NULL);
+    if (seen == NULL) {
+        set_exception_cause(self, PyExc_MemoryError, "Failed to create interpreter set");
         return NULL;
     }
 
@@ -968,30 +970,65 @@ _remote_debugging_RemoteUnwinder_get_all_awaited_by_impl(RemoteUnwinderObject *s
         goto result_err;
     }
 
-    // Process all threads
-    if (iterate_threads(self, process_thread_for_awaited_by, result) < 0) {
-        goto result_err;
-    }
+    // gh-158880: Tasks live in every interpreter, not only the one at the list head
+    for (uintptr_t interp = self->interpreter_addr; interp != 0; ) {
+        PyObject *addr = PyLong_FromUnsignedLongLong(interp);
+        if (addr == NULL) {
+            set_exception_cause(self, PyExc_MemoryError, "Failed to create interpreter address");
+            goto result_err;
+        }
+        Py_ssize_t seen_count = PySet_GET_SIZE(seen);
+        int marked = PySet_Add(seen, addr);
+        Py_DECREF(addr);
+        if (marked < 0) {
+            set_exception_cause(self, PyExc_RuntimeError, "Failed to mark interpreter as seen");
+            goto result_err;
+        }
+        if (PySet_GET_SIZE(seen) == seen_count) {
+            // already walked
+            break;
+        }
 
-    uintptr_t head_addr = self->interpreter_addr
-        + (uintptr_t)self->async_debug_offsets.asyncio_interpreter_state.asyncio_tasks_head;
+        if (refresh_generation_caches_for_interpreter(self, interp) < 0) {
+            goto result_err;
+        }
 
-    // On top of a per-thread task lists used by default by asyncio to avoid
-    // contention, there is also a fallback per-interpreter list of tasks;
-    // any tasks still pending when a thread is destroyed will be moved to the
-    // per-interpreter task list.  It's unlikely we'll find anything here, but
-    // interesting for debugging.
-    if (append_awaited_by(self, 0, head_addr, result))
-    {
-        set_exception_cause(self, PyExc_RuntimeError, "Failed to append interpreter awaited_by in get_all_awaited_by");
-        goto result_err;
+        // Process all threads
+        if (iterate_threads(self, interp, process_thread_for_awaited_by, result) < 0) {
+            goto result_err;
+        }
+
+        uintptr_t head_addr = interp
+            + (uintptr_t)self->async_debug_offsets.asyncio_interpreter_state.asyncio_tasks_head;
+
+        // On top of a per-thread task lists used by default by asyncio to avoid
+        // contention, there is also a fallback per-interpreter list of tasks;
+        // any tasks still pending when a thread is destroyed will be moved to
+        // the per-interpreter task list.  It's unlikely we'll find anything
+        // here, but interesting for debugging.
+        if (append_awaited_by(self, 0, head_addr, result))
+        {
+            set_exception_cause(self, PyExc_RuntimeError, "Failed to append interpreter awaited_by in get_all_awaited_by");
+            goto result_err;
+        }
+
+        if (_Py_RemoteDebug_PagedReadRemoteMemory(
+                &self->handle,
+                interp + (uintptr_t)self->debug_offsets.interpreter_state.next,
+                sizeof(void*),
+                &interp) < 0) {
+            set_exception_cause(self, PyExc_RuntimeError, "Failed to read next interpreter address");
+            goto result_err;
+        }
     }
 
     _Py_RemoteDebug_ClearCache(&self->handle);
+    Py_DECREF(seen);
     return result;
 
 result_err:
     _Py_RemoteDebug_ClearCache(&self->handle);
+    Py_DECREF(seen);
     Py_XDECREF(result);
     return NULL;
 }
@@ -1064,7 +1101,8 @@ _remote_debugging_RemoteUnwinder_get_async_stack_trace_impl(RemoteUnwinderObject
     }
 
     // Process all threads
-    if (iterate_threads(self, process_thread_for_async_stack_trace, result) < 0) {
+    if (iterate_threads(self, self->interpreter_addr,
+                        process_thread_for_async_stack_trace, result) < 0) {
         goto result_err;
     }
 

Modules/_remote_debugging/threads.c

@@ -27,6 +27,7 @@
 int
 iterate_threads(
     RemoteUnwinderObject *unwinder,
+    uintptr_t interpreter_addr,
     thread_processor_func processor,
     void *context
 ) {
@@ -37,7 +38,7 @@ iterate_threads(
 
     if (0 > _Py_RemoteDebug_PagedReadRemoteMemory(
                 &unwinder->handle,
-                unwinder->interpreter_addr + (uintptr_t)unwinder->debug_offsets.interpreter_state.threads_head,
+                interpreter_addr + (uintptr_t)unwinder->debug_offsets.interpreter_state.threads_head,
                 sizeof(void*),
                 &thread_state_addr))
     {