Skip to content

Commit 03b736b

Browse files
committed
gh-158880: Fix asyncio ps showing tasks from only one interpreter
1 parent dc0add3 commit 03b736b

5 files changed

Lines changed: 97 additions & 18 deletions

File tree

‎Lib/test/test_external_inspection.py‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646

4747
try:
4848
from concurrent import interpreters
49+
from concurrent.futures import InterpreterPoolExecutor
4950
except ImportError:
5051
interpreters = None
5152

@@ -214,6 +215,15 @@ def _cleanup_sockets(*sockets):
214215
)
215216

216217

218+
def _asyncio_in_subinterpreter():
219+
import asyncio
220+
221+
async def sub_worker():
222+
await asyncio.sleep(2)
223+
224+
asyncio.run(sub_worker())
225+
226+
217227
def requires_subinterpreters(meth):
218228
"""Decorator to skip a test if subinterpreters are not supported."""
219229
return unittest.skipIf(interpreters is None, "subinterpreters required")(
@@ -492,6 +502,33 @@ async def main():
492502
self.assertIn(main_name, names)
493503
self.assertEqual([len(n) for n in names if n.startswith("x")], [255])
494504

505+
@skip_if_not_supported
506+
@requires_subinterpreters
507+
def test_all_awaited_by_covers_every_interpreter(self):
508+
# gh-158880
509+
async def main_worker():
510+
await asyncio.sleep(2)
511+
512+
async def main():
513+
with InterpreterPoolExecutor() as pool:
514+
loop = asyncio.get_running_loop()
515+
loop.run_in_executor(pool, _asyncio_in_subinterpreter)
516+
task = asyncio.create_task(main_worker(), name="main_worker")
517+
self.addCleanup(task.cancel)
518+
await asyncio.sleep(1)
519+
return [
520+
[frame.funcname.rpartition(".")[2]
521+
for frame in coro.call_stack]
522+
for info in RemoteUnwinder(
523+
os.getpid()).get_all_awaited_by()
524+
for task in info.awaited_by
525+
for coro in task.coroutine_stack
526+
]
527+
528+
stacks = asyncio.run(main())
529+
self.assertIn(["sleep", "sub_worker"], stacks)
530+
self.assertIn(["sleep", "main_worker"], stacks)
531+
495532
@skip_if_not_supported
496533
def test_recursive_coroutine_stack_is_not_truncated(self):
497534
# gh-158522
Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
Fix ``python -m asyncio ps`` and ``pstree`` showing tasks from only one
2+
interpreter. Patch by Timofei Ivankov.

‎Modules/_remote_debugging/_remote_debugging.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -663,6 +663,7 @@ extern int collect_frames_with_cache(
663663

664664
extern int iterate_threads(
665665
RemoteUnwinderObject *unwinder,
666+
uintptr_t interpreter_addr,
666667
thread_processor_func processor,
667668
void *context
668669
);

‎Modules/_remote_debugging/module.c‎

Lines changed: 55 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -958,7 +958,9 @@ _remote_debugging_RemoteUnwinder_get_all_awaited_by_impl(RemoteUnwinderObject *s
958958
if (ensure_async_debug_offsets(self) < 0) {
959959
return NULL;
960960
}
961-
if (refresh_generation_caches_for_interpreter(self, self->interpreter_addr) < 0) {
961+
PyObject *seen = PySet_New(NULL);
962+
if (seen == NULL) {
963+
set_exception_cause(self, PyExc_MemoryError, "Failed to create interpreter set");
962964
return NULL;
963965
}
964966

@@ -968,30 +970,65 @@ _remote_debugging_RemoteUnwinder_get_all_awaited_by_impl(RemoteUnwinderObject *s
968970
goto result_err;
969971
}
970972

971-
// Process all threads
972-
if (iterate_threads(self, process_thread_for_awaited_by, result) < 0) {
973-
goto result_err;
974-
}
973+
// gh-158880: Tasks live in every interpreter, not only the one at the list head
974+
for (uintptr_t interp = self->interpreter_addr; interp != 0; ) {
975+
PyObject *addr = PyLong_FromUnsignedLongLong(interp);
976+
if (addr == NULL) {
977+
set_exception_cause(self, PyExc_MemoryError, "Failed to create interpreter address");
978+
goto result_err;
979+
}
980+
Py_ssize_t seen_count = PySet_GET_SIZE(seen);
981+
int marked = PySet_Add(seen, addr);
982+
Py_DECREF(addr);
983+
if (marked < 0) {
984+
set_exception_cause(self, PyExc_RuntimeError, "Failed to mark interpreter as seen");
985+
goto result_err;
986+
}
987+
if (PySet_GET_SIZE(seen) == seen_count) {
988+
// already walked
989+
break;
990+
}
975991

976-
uintptr_t head_addr = self->interpreter_addr
977-
+ (uintptr_t)self->async_debug_offsets.asyncio_interpreter_state.asyncio_tasks_head;
992+
if (refresh_generation_caches_for_interpreter(self, interp) < 0) {
993+
goto result_err;
994+
}
978995

979-
// On top of a per-thread task lists used by default by asyncio to avoid
980-
// contention, there is also a fallback per-interpreter list of tasks;
981-
// any tasks still pending when a thread is destroyed will be moved to the
982-
// per-interpreter task list. It's unlikely we'll find anything here, but
983-
// interesting for debugging.
984-
if (append_awaited_by(self, 0, head_addr, result))
985-
{
986-
set_exception_cause(self, PyExc_RuntimeError, "Failed to append interpreter awaited_by in get_all_awaited_by");
987-
goto result_err;
996+
// Process all threads
997+
if (iterate_threads(self, interp, process_thread_for_awaited_by, result) < 0) {
998+
goto result_err;
999+
}
1000+
1001+
uintptr_t head_addr = interp
1002+
+ (uintptr_t)self->async_debug_offsets.asyncio_interpreter_state.asyncio_tasks_head;
1003+
1004+
// On top of a per-thread task lists used by default by asyncio to avoid
1005+
// contention, there is also a fallback per-interpreter list of tasks;
1006+
// any tasks still pending when a thread is destroyed will be moved to
1007+
// the per-interpreter task list. It's unlikely we'll find anything
1008+
// here, but interesting for debugging.
1009+
if (append_awaited_by(self, 0, head_addr, result))
1010+
{
1011+
set_exception_cause(self, PyExc_RuntimeError, "Failed to append interpreter awaited_by in get_all_awaited_by");
1012+
goto result_err;
1013+
}
1014+
1015+
if (_Py_RemoteDebug_PagedReadRemoteMemory(
1016+
&self->handle,
1017+
interp + (uintptr_t)self->debug_offsets.interpreter_state.next,
1018+
sizeof(void*),
1019+
&interp) < 0) {
1020+
set_exception_cause(self, PyExc_RuntimeError, "Failed to read next interpreter address");
1021+
goto result_err;
1022+
}
9881023
}
9891024

9901025
_Py_RemoteDebug_ClearCache(&self->handle);
1026+
Py_DECREF(seen);
9911027
return result;
9921028

9931029
result_err:
9941030
_Py_RemoteDebug_ClearCache(&self->handle);
1031+
Py_DECREF(seen);
9951032
Py_XDECREF(result);
9961033
return NULL;
9971034
}
@@ -1064,7 +1101,8 @@ _remote_debugging_RemoteUnwinder_get_async_stack_trace_impl(RemoteUnwinderObject
10641101
}
10651102

10661103
// Process all threads
1067-
if (iterate_threads(self, process_thread_for_async_stack_trace, result) < 0) {
1104+
if (iterate_threads(self, self->interpreter_addr,
1105+
process_thread_for_async_stack_trace, result) < 0) {
10681106
goto result_err;
10691107
}
10701108

‎Modules/_remote_debugging/threads.c‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
int
2828
iterate_threads(
2929
RemoteUnwinderObject *unwinder,
30+
uintptr_t interpreter_addr,
3031
thread_processor_func processor,
3132
void *context
3233
) {
@@ -37,7 +38,7 @@ iterate_threads(
3738

3839
if (0 > _Py_RemoteDebug_PagedReadRemoteMemory(
3940
&unwinder->handle,
40-
unwinder->interpreter_addr + (uintptr_t)unwinder->debug_offsets.interpreter_state.threads_head,
41+
interpreter_addr + (uintptr_t)unwinder->debug_offsets.interpreter_state.threads_head,
4142
sizeof(void*),
4243
&thread_state_addr))
4344
{

0 commit comments

Comments
 (0)