Skip to content

Commit fb51040

Browse files
maurycymiss-islington
authored andcommitted
gh-153364: Make frame, coroutine, and task-waiter chain walks iterative and bounded (GH-153365)
* let me declare single limit * use our new limit in process_frame_chain() * add it in parse_async_frame_chain() * parse_coro_chain() * NEWS * async in the message? * test * no race * process_task_awaited_by * process_task_awaited_by limit test * NEWS * MAX_TASK_WAITER_CHAIN_DEPTH * TASK_WAITER_CHAIN_DEPTH in test * TASK_WAITER_CHAIN_DEPTH 256 * prevent the drift with the comment * better naming, better style * MAX_TASK_WAITER_CHAIN_DEPTH comment * task-waiter iterative bfs walk * iterative coro-walk * nicer news * 1 << 14 * comment * unused read_Py_ssize_t * fix tombstones * simplify * correct msg * better test * news for tombstones * left-over from when testing buggy version * redundant new line (cherry picked from commit e0861c6) Co-authored-by: Maurycy Pawłowski-Wieroński <maurycy@maurycy.com>
1 parent 442ad23 commit fb51040

6 files changed

Lines changed: 370 additions & 182 deletions

File tree

‎Lib/test/test_external_inspection.py‎

Lines changed: 257 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -340,6 +340,40 @@ def _run_script_and_get_trace(
340340
finally:
341341
_cleanup_sockets(client_socket, server_socket)
342342

343+
@contextmanager
344+
def _target_process(self, script_body):
345+
"""Context manager for running a target process with socket sync."""
346+
port = find_unused_port()
347+
script = f"""\
348+
import socket
349+
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
350+
sock.connect(('localhost', {port}))
351+
{textwrap.dedent(script_body)}
352+
"""
353+
354+
with os_helper.temp_dir() as work_dir:
355+
script_dir = os.path.join(work_dir, "script_pkg")
356+
os.mkdir(script_dir)
357+
358+
server_socket = _create_server_socket(port)
359+
script_name = _make_test_script(script_dir, "script", script)
360+
client_socket = None
361+
362+
try:
363+
with _managed_subprocess([sys.executable, script_name]) as p:
364+
client_socket, _ = server_socket.accept()
365+
server_socket.close()
366+
server_socket = None
367+
368+
def make_unwinder(cache_frames=True):
369+
return RemoteUnwinder(
370+
p.pid, all_threads=True, cache_frames=cache_frames
371+
)
372+
373+
yield p, client_socket, make_unwinder
374+
finally:
375+
_cleanup_sockets(client_socket, server_socket)
376+
343377
def _find_frame_in_trace(self, stack_trace, predicate):
344378
"""
345379
Find a frame matching predicate in stack trace.
@@ -1505,6 +1539,71 @@ def matches_awaited_by_pattern(task):
15051539
finally:
15061540
_cleanup_sockets(client_socket, server_socket)
15071541

1542+
@skip_if_not_supported
1543+
@unittest.skipIf(
1544+
sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
1545+
"Test only runs on Linux with process_vm_readv support",
1546+
)
1547+
def test_async_awaited_by_skips_set_tombstones(self):
1548+
script_body = """\
1549+
import asyncio
1550+
1551+
class RemovedTask(asyncio.Task):
1552+
def __hash__(self):
1553+
return 0
1554+
1555+
class RemainingTask(asyncio.Task):
1556+
def __hash__(self):
1557+
return 1
1558+
1559+
async def main():
1560+
victim = asyncio.current_task()
1561+
victim.set_name("victim")
1562+
removed = RemovedTask(
1563+
asyncio.sleep(10_000), name="removed"
1564+
)
1565+
remaining = RemainingTask(
1566+
asyncio.sleep(10_000), name="remaining"
1567+
)
1568+
1569+
asyncio.future_add_to_awaited_by(victim, removed)
1570+
asyncio.future_add_to_awaited_by(victim, remaining)
1571+
1572+
# Removing hash 0 leaves a dummy in slot 0 before the only
1573+
# active entry in slot 1. It must not count toward the set's
1574+
# used entries.
1575+
asyncio.future_discard_from_awaited_by(victim, removed)
1576+
1577+
sock.sendall(b"ready")
1578+
sock.recv(16)
1579+
1580+
asyncio.run(main())
1581+
"""
1582+
1583+
with self._target_process(script_body) as (
1584+
_,
1585+
client_socket,
1586+
make_unwinder,
1587+
):
1588+
_wait_for_signal(client_socket, b"ready")
1589+
1590+
for method_name in (
1591+
"get_async_stack_trace",
1592+
"get_all_awaited_by",
1593+
):
1594+
with self.subTest(method=method_name):
1595+
unwinder = make_unwinder(cache_frames=False)
1596+
stack_trace = getattr(unwinder, method_name)()
1597+
relationships = self._get_awaited_by_relationships(
1598+
stack_trace
1599+
)
1600+
self.assertEqual(
1601+
relationships["victim"],
1602+
{"remaining"},
1603+
)
1604+
1605+
client_socket.sendall(b"done")
1606+
15081607
@skip_if_not_supported
15091608
@unittest.skipIf(
15101609
sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
@@ -3116,40 +3215,6 @@ class TestFrameCaching(RemoteInspectionTestBase):
31163215
All tests verify cache reuse via object identity checks (assertIs).
31173216
"""
31183217

3119-
@contextmanager
3120-
def _target_process(self, script_body):
3121-
"""Context manager for running a target process with socket sync."""
3122-
port = find_unused_port()
3123-
script = f"""\
3124-
import socket
3125-
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
3126-
sock.connect(('localhost', {port}))
3127-
{textwrap.dedent(script_body)}
3128-
"""
3129-
3130-
with os_helper.temp_dir() as work_dir:
3131-
script_dir = os.path.join(work_dir, "script_pkg")
3132-
os.mkdir(script_dir)
3133-
3134-
server_socket = _create_server_socket(port)
3135-
script_name = _make_test_script(script_dir, "script", script)
3136-
client_socket = None
3137-
3138-
try:
3139-
with _managed_subprocess([sys.executable, script_name]) as p:
3140-
client_socket, _ = server_socket.accept()
3141-
server_socket.close()
3142-
server_socket = None
3143-
3144-
def make_unwinder(cache_frames=True):
3145-
return RemoteUnwinder(
3146-
p.pid, all_threads=True, cache_frames=cache_frames
3147-
)
3148-
3149-
yield p, client_socket, make_unwinder
3150-
finally:
3151-
_cleanup_sockets(client_socket, server_socket)
3152-
31533218
def _get_frames_with_retry(self, unwinder, required_funcs):
31543219
"""Get frames containing required_funcs, with retry for transient errors."""
31553220
for _ in range(MAX_TRIES):
@@ -3993,5 +4058,163 @@ def test_get_stats_disabled_raises(self):
39934058
client_socket.sendall(b"done")
39944059

39954060

4061+
@requires_remote_subprocess_debugging()
4062+
class TestFrameChainLimits(RemoteInspectionTestBase):
4063+
"""Frame chain walks abort instead of looping/overflowing on deep chains."""
4064+
4065+
# Limits plus one, to exceed them (must match MAX_FRAME_CHAIN_DEPTH /
4066+
# MAX_TASK_WAITER_WALK_TASKS from _remote_debugging.h)
4067+
FRAME_CHAIN_DEPTH = 1024 + 512 + 1
4068+
TASK_WAITER_WALK_TASKS = 2**14 + 1
4069+
4070+
def _assert_unwinder_limit_error(self, unwind, expected_substring):
4071+
"""Call unwind() until it raises the frame chain limit error.
4072+
4073+
unwind must construct the RemoteUnwinder and call it, so that
4074+
transient RuntimeErrors from either step are retried; a successful
4075+
call means the limit never triggered and fails immediately.
4076+
"""
4077+
last_error = None
4078+
for _ in busy_retry(SHORT_TIMEOUT, error=False):
4079+
try:
4080+
unwind()
4081+
except TRANSIENT_ERRORS as e:
4082+
if expected_substring in str(e):
4083+
return
4084+
last_error = e
4085+
continue
4086+
self.fail(
4087+
"frame chain limit did not trigger; call returned a result"
4088+
)
4089+
self.fail(
4090+
f"frame chain limit never raised; last transient error: "
4091+
f"{last_error!r}"
4092+
)
4093+
4094+
@skip_if_not_supported
4095+
@unittest.skipIf(
4096+
sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
4097+
"Test only runs on Linux with process_vm_readv support",
4098+
)
4099+
def test_get_stack_trace_deep_frame_chain_aborts(self):
4100+
"""Test that a frame chain deeper than the limit aborts the
4101+
synchronous stack walk instead of walking it indefinitely."""
4102+
script_body = f"""\
4103+
import sys
4104+
sys.setrecursionlimit({self.FRAME_CHAIN_DEPTH * 2})
4105+
4106+
def recurse(n):
4107+
if n <= 0:
4108+
sock.sendall(b"ready")
4109+
sock.recv(16)
4110+
return
4111+
recurse(n - 1)
4112+
4113+
recurse({self.FRAME_CHAIN_DEPTH})
4114+
"""
4115+
with self._target_process(script_body) as (p, client_socket, _):
4116+
_wait_for_signal(client_socket, b"ready")
4117+
self._assert_unwinder_limit_error(
4118+
lambda: RemoteUnwinder(p.pid).get_stack_trace(),
4119+
"Too many stack frames",
4120+
)
4121+
client_socket.sendall(b"done")
4122+
4123+
@skip_if_not_supported
4124+
@unittest.skipIf(
4125+
sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
4126+
"Test only runs on Linux with process_vm_readv support",
4127+
)
4128+
def test_get_async_stack_trace_deep_task_waiter_chain_aborts(self):
4129+
"""Test that a task waiter chain deeper than the limit aborts
4130+
the walk instead of overflowing the C stack."""
4131+
script_body = f"""\
4132+
import asyncio
4133+
4134+
async def chain(n):
4135+
if n <= 0:
4136+
sock.sendall(b"ready")
4137+
sock.recv(16)
4138+
return
4139+
4140+
task = asyncio.create_task(chain(n - 1))
4141+
await task
4142+
4143+
asyncio.run(chain({self.TASK_WAITER_WALK_TASKS}))
4144+
"""
4145+
with self._target_process(script_body) as (p, client_socket, _):
4146+
_wait_for_signal(client_socket, b"ready")
4147+
self._assert_unwinder_limit_error(
4148+
lambda: RemoteUnwinder(p.pid).get_async_stack_trace(),
4149+
"Too many task waiters",
4150+
)
4151+
client_socket.sendall(b"done")
4152+
4153+
@skip_if_not_supported
4154+
@unittest.skipIf(
4155+
sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
4156+
"Test only runs on Linux with process_vm_readv support",
4157+
)
4158+
def test_get_async_stack_trace_deep_frame_chain_aborts(self):
4159+
"""Test that a frame chain deeper than the limit aborts the async
4160+
stack walk instead of walking it indefinitely."""
4161+
script_body = f"""\
4162+
import sys, asyncio
4163+
sys.setrecursionlimit({self.FRAME_CHAIN_DEPTH * 2})
4164+
4165+
def recurse(n):
4166+
if n <= 0:
4167+
sock.sendall(b"ready")
4168+
sock.recv(16)
4169+
return
4170+
recurse(n - 1)
4171+
4172+
async def deep():
4173+
recurse({self.FRAME_CHAIN_DEPTH})
4174+
4175+
asyncio.run(deep())
4176+
"""
4177+
with self._target_process(script_body) as (p, client_socket, _):
4178+
_wait_for_signal(client_socket, b"ready")
4179+
self._assert_unwinder_limit_error(
4180+
lambda: RemoteUnwinder(p.pid).get_async_stack_trace(),
4181+
"Too many async stack frames",
4182+
)
4183+
client_socket.sendall(b"done")
4184+
4185+
@skip_if_not_supported
4186+
@unittest.skipIf(
4187+
sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
4188+
"Test only runs on Linux with process_vm_readv support",
4189+
)
4190+
def test_get_all_awaited_by_deep_coro_chain_aborts(self):
4191+
"""Test that a coroutine await chain deeper than the limit aborts
4192+
the walk instead of overflowing the C stack."""
4193+
script_body = f"""\
4194+
import sys, asyncio
4195+
sys.setrecursionlimit({self.FRAME_CHAIN_DEPTH * 2})
4196+
4197+
async def chain(n):
4198+
if n <= 0:
4199+
await asyncio.sleep(10_000)
4200+
return
4201+
await chain(n - 1)
4202+
4203+
async def main():
4204+
task = asyncio.create_task(chain({self.FRAME_CHAIN_DEPTH}))
4205+
await asyncio.sleep(0)
4206+
sock.sendall(b"ready")
4207+
await task
4208+
4209+
asyncio.run(main())
4210+
"""
4211+
with self._target_process(script_body) as (p, client_socket, _):
4212+
_wait_for_signal(client_socket, b"ready")
4213+
self._assert_unwinder_limit_error(
4214+
lambda: RemoteUnwinder(p.pid).get_all_awaited_by(),
4215+
"Too many coroutine frames",
4216+
)
4217+
4218+
39964219
if __name__ == "__main__":
39974220
unittest.main()
Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
Make frame, coroutine and task-waiter walks iterative and bounded, avoiding
2+
potential hangs and stack overflows. Fix asyncio task inspection when
3+
awaited-by sets contain removed entries. Patch by Maurycy Pawłowski-Wieroński.

0 commit comments

Comments
 (0)