Skip to content

Commit e0861c6

Browse files
authored
gh-153364: Make frame, coroutine, and task-waiter chain walks iterative and bounded (#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
1 parent 587b7a5 commit e0861c6

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,
@@ -3190,40 +3289,6 @@ class TestFrameCaching(RemoteInspectionTestBase):
31903289
All tests verify cache reuse via object identity checks (assertIs).
31913290
"""
31923291

3193-
@contextmanager
3194-
def _target_process(self, script_body):
3195-
"""Context manager for running a target process with socket sync."""
3196-
port = find_unused_port()
3197-
script = f"""\
3198-
import socket
3199-
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
3200-
sock.connect(('localhost', {port}))
3201-
{textwrap.dedent(script_body)}
3202-
"""
3203-
3204-
with os_helper.temp_dir() as work_dir:
3205-
script_dir = os.path.join(work_dir, "script_pkg")
3206-
os.mkdir(script_dir)
3207-
3208-
server_socket = _create_server_socket(port)
3209-
script_name = _make_test_script(script_dir, "script", script)
3210-
client_socket = None
3211-
3212-
try:
3213-
with _managed_subprocess([sys.executable, script_name]) as p:
3214-
client_socket, _ = server_socket.accept()
3215-
server_socket.close()
3216-
server_socket = None
3217-
3218-
def make_unwinder(cache_frames=True):
3219-
return RemoteUnwinder(
3220-
p.pid, all_threads=True, cache_frames=cache_frames
3221-
)
3222-
3223-
yield p, client_socket, make_unwinder
3224-
finally:
3225-
_cleanup_sockets(client_socket, server_socket)
3226-
32273292
def _get_frames_with_retry(self, unwinder, required_funcs):
32283293
"""Get frames containing required_funcs, with retry for transient errors."""
32293294
for _ in range(MAX_TRIES):
@@ -4067,5 +4132,163 @@ def test_get_stats_disabled_raises(self):
40674132
client_socket.sendall(b"done")
40684133

40694134

4135+
@requires_remote_subprocess_debugging()
4136+
class TestFrameChainLimits(RemoteInspectionTestBase):
4137+
"""Frame chain walks abort instead of looping/overflowing on deep chains."""
4138+
4139+
# Limits plus one, to exceed them (must match MAX_FRAME_CHAIN_DEPTH /
4140+
# MAX_TASK_WAITER_WALK_TASKS from _remote_debugging.h)
4141+
FRAME_CHAIN_DEPTH = 1024 + 512 + 1
4142+
TASK_WAITER_WALK_TASKS = 2**14 + 1
4143+
4144+
def _assert_unwinder_limit_error(self, unwind, expected_substring):
4145+
"""Call unwind() until it raises the frame chain limit error.
4146+
4147+
unwind must construct the RemoteUnwinder and call it, so that
4148+
transient RuntimeErrors from either step are retried; a successful
4149+
call means the limit never triggered and fails immediately.
4150+
"""
4151+
last_error = None
4152+
for _ in busy_retry(SHORT_TIMEOUT, error=False):
4153+
try:
4154+
unwind()
4155+
except TRANSIENT_ERRORS as e:
4156+
if expected_substring in str(e):
4157+
return
4158+
last_error = e
4159+
continue
4160+
self.fail(
4161+
"frame chain limit did not trigger; call returned a result"
4162+
)
4163+
self.fail(
4164+
f"frame chain limit never raised; last transient error: "
4165+
f"{last_error!r}"
4166+
)
4167+
4168+
@skip_if_not_supported
4169+
@unittest.skipIf(
4170+
sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
4171+
"Test only runs on Linux with process_vm_readv support",
4172+
)
4173+
def test_get_stack_trace_deep_frame_chain_aborts(self):
4174+
"""Test that a frame chain deeper than the limit aborts the
4175+
synchronous stack walk instead of walking it indefinitely."""
4176+
script_body = f"""\
4177+
import sys
4178+
sys.setrecursionlimit({self.FRAME_CHAIN_DEPTH * 2})
4179+
4180+
def recurse(n):
4181+
if n <= 0:
4182+
sock.sendall(b"ready")
4183+
sock.recv(16)
4184+
return
4185+
recurse(n - 1)
4186+
4187+
recurse({self.FRAME_CHAIN_DEPTH})
4188+
"""
4189+
with self._target_process(script_body) as (p, client_socket, _):
4190+
_wait_for_signal(client_socket, b"ready")
4191+
self._assert_unwinder_limit_error(
4192+
lambda: RemoteUnwinder(p.pid).get_stack_trace(),
4193+
"Too many stack frames",
4194+
)
4195+
client_socket.sendall(b"done")
4196+
4197+
@skip_if_not_supported
4198+
@unittest.skipIf(
4199+
sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
4200+
"Test only runs on Linux with process_vm_readv support",
4201+
)
4202+
def test_get_async_stack_trace_deep_task_waiter_chain_aborts(self):
4203+
"""Test that a task waiter chain deeper than the limit aborts
4204+
the walk instead of overflowing the C stack."""
4205+
script_body = f"""\
4206+
import asyncio
4207+
4208+
async def chain(n):
4209+
if n <= 0:
4210+
sock.sendall(b"ready")
4211+
sock.recv(16)
4212+
return
4213+
4214+
task = asyncio.create_task(chain(n - 1))
4215+
await task
4216+
4217+
asyncio.run(chain({self.TASK_WAITER_WALK_TASKS}))
4218+
"""
4219+
with self._target_process(script_body) as (p, client_socket, _):
4220+
_wait_for_signal(client_socket, b"ready")
4221+
self._assert_unwinder_limit_error(
4222+
lambda: RemoteUnwinder(p.pid).get_async_stack_trace(),
4223+
"Too many task waiters",
4224+
)
4225+
client_socket.sendall(b"done")
4226+
4227+
@skip_if_not_supported
4228+
@unittest.skipIf(
4229+
sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
4230+
"Test only runs on Linux with process_vm_readv support",
4231+
)
4232+
def test_get_async_stack_trace_deep_frame_chain_aborts(self):
4233+
"""Test that a frame chain deeper than the limit aborts the async
4234+
stack walk instead of walking it indefinitely."""
4235+
script_body = f"""\
4236+
import sys, asyncio
4237+
sys.setrecursionlimit({self.FRAME_CHAIN_DEPTH * 2})
4238+
4239+
def recurse(n):
4240+
if n <= 0:
4241+
sock.sendall(b"ready")
4242+
sock.recv(16)
4243+
return
4244+
recurse(n - 1)
4245+
4246+
async def deep():
4247+
recurse({self.FRAME_CHAIN_DEPTH})
4248+
4249+
asyncio.run(deep())
4250+
"""
4251+
with self._target_process(script_body) as (p, client_socket, _):
4252+
_wait_for_signal(client_socket, b"ready")
4253+
self._assert_unwinder_limit_error(
4254+
lambda: RemoteUnwinder(p.pid).get_async_stack_trace(),
4255+
"Too many async stack frames",
4256+
)
4257+
client_socket.sendall(b"done")
4258+
4259+
@skip_if_not_supported
4260+
@unittest.skipIf(
4261+
sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
4262+
"Test only runs on Linux with process_vm_readv support",
4263+
)
4264+
def test_get_all_awaited_by_deep_coro_chain_aborts(self):
4265+
"""Test that a coroutine await chain deeper than the limit aborts
4266+
the walk instead of overflowing the C stack."""
4267+
script_body = f"""\
4268+
import sys, asyncio
4269+
sys.setrecursionlimit({self.FRAME_CHAIN_DEPTH * 2})
4270+
4271+
async def chain(n):
4272+
if n <= 0:
4273+
await asyncio.sleep(10_000)
4274+
return
4275+
await chain(n - 1)
4276+
4277+
async def main():
4278+
task = asyncio.create_task(chain({self.FRAME_CHAIN_DEPTH}))
4279+
await asyncio.sleep(0)
4280+
sock.sendall(b"ready")
4281+
await task
4282+
4283+
asyncio.run(main())
4284+
"""
4285+
with self._target_process(script_body) as (p, client_socket, _):
4286+
_wait_for_signal(client_socket, b"ready")
4287+
self._assert_unwinder_limit_error(
4288+
lambda: RemoteUnwinder(p.pid).get_all_awaited_by(),
4289+
"Too many coroutine frames",
4290+
)
4291+
4292+
40704293
if __name__ == "__main__":
40714294
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)