Coverage for Lib/asyncio/graph.py: 99%
116 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-25 03:29 +0000
1"""Introspection utils for tasks call graphs."""
3import dataclasses
4import io
5import sys
6import types
8from . import events
9from . import futures
10from . import tasks
12__all__ = (
13 'capture_call_graph',
14 'format_call_graph',
15 'print_call_graph',
16 'FrameCallGraphEntry',
17 'FutureCallGraph',
18)
20# Sadly, we can't re-use the traceback module's datastructures as those
21# are tailored for error reporting, whereas we need to represent an
22# async call graph.
23#
24# Going with pretty verbose names as we'd like to export them to the
25# top level asyncio namespace, and want to avoid future name clashes.
28@dataclasses.dataclass(frozen=True, slots=True)
29class FrameCallGraphEntry:
30 frame: types.FrameType
33@dataclasses.dataclass(frozen=True, slots=True)
34class FutureCallGraph:
35 future: futures.Future
36 call_stack: tuple["FrameCallGraphEntry", ...]
37 awaited_by: tuple["FutureCallGraph", ...]
40def _build_graph_for_future(
41 future: futures.Future,
42 *,
43 limit: int | None = None,
44) -> FutureCallGraph:
45 if not isinstance(future, futures.Future):
46 raise TypeError(
47 f"{future!r} object does not appear to be compatible "
48 f"with asyncio.Future"
49 )
51 coro = None
52 if get_coro := getattr(future, 'get_coro', None):
53 coro = get_coro() if limit != 0 else None
55 st: list[FrameCallGraphEntry] = []
56 awaited_by: list[FutureCallGraph] = []
58 while coro is not None:
59 if hasattr(coro, 'cr_await'):
60 # A native coroutine or duck-type compatible iterator
61 if coro.cr_frame is not None:
62 st.append(FrameCallGraphEntry(coro.cr_frame))
63 coro = coro.cr_await
64 elif hasattr(coro, 'ag_await'):
65 # A native async generator or duck-type compatible iterator
66 st.append(FrameCallGraphEntry(coro.ag_frame))
67 coro = coro.ag_await
68 else:
69 break
71 if future._asyncio_awaited_by:
72 for parent in future._asyncio_awaited_by:
73 awaited_by.append(_build_graph_for_future(parent, limit=limit))
75 if limit is not None:
76 if limit > 0:
77 st = st[:limit]
78 elif limit < 0: 78 ↛ 80line 78 didn't jump to line 80 because the condition on line 78 was always true
79 st = st[limit:]
80 st.reverse()
81 return FutureCallGraph(future, tuple(st), tuple(awaited_by))
84def capture_call_graph(
85 future: futures.Future | None = None,
86 /,
87 *,
88 depth: int = 1,
89 limit: int | None = None,
90) -> FutureCallGraph | None:
91 """Capture the async call graph for the current task or the provided Future.
93 The graph is represented with three data structures:
95 * FutureCallGraph(future, call_stack, awaited_by)
97 Where 'future' is an instance of asyncio.Future or asyncio.Task.
99 'call_stack' is a tuple of FrameGraphEntry objects.
101 'awaited_by' is a tuple of FutureCallGraph objects.
103 * FrameCallGraphEntry(frame)
105 Where 'frame' is a frame object of a regular Python function
106 in the call stack.
108 Receives an optional 'future' argument. If not passed,
109 the current task will be used. If there's no current task, the function
110 returns None.
112 If "capture_call_graph()" is introspecting *the current task*, the
113 optional keyword-only 'depth' argument can be used to skip the specified
114 number of frames from top of the stack.
116 If the optional keyword-only 'limit' argument is provided, each call
117 stack in the resulting graph is truncated to include at most
118 ``abs(limit)`` entries. If 'limit' is positive, the entries left are
119 the closest to the invocation point. If 'limit' is negative, the
120 topmost entries are left. If 'limit' is omitted or None, all entries
121 are present. If 'limit' is 0, the call stack is not captured at all,
122 only "awaited by" information is present.
123 """
125 loop = events._get_running_loop()
127 if future is not None:
128 # Check if we're in a context of a running event loop;
129 # if yes - check if the passed future is the currently
130 # running task or not.
131 if loop is None or future is not tasks.current_task(loop=loop):
132 return _build_graph_for_future(future, limit=limit)
133 # else: future is the current task, move on.
134 else:
135 if loop is None:
136 raise RuntimeError(
137 'capture_call_graph() is called outside of a running '
138 'event loop and no *future* to introspect was provided')
139 future = tasks.current_task(loop=loop)
141 if future is None:
142 # This isn't a generic call stack introspection utility. If we
143 # can't determine the current task and none was provided, we
144 # just return.
145 return None
147 if not isinstance(future, futures.Future):
148 raise TypeError(
149 f"{future!r} object does not appear to be compatible "
150 f"with asyncio.Future"
151 )
153 call_stack: list[FrameCallGraphEntry] = []
155 f = sys._getframe(depth) if limit != 0 else None
156 try:
157 while f is not None:
158 # gh-156988: sync gen should not clear the call chain
159 is_async = isinstance(
160 f.f_generator, (types.CoroutineType, types.AsyncGeneratorType))
161 call_stack.append(FrameCallGraphEntry(f))
163 if is_async:
164 if f.f_back is not None and f.f_back.f_generator is None:
165 # We've reached the bottom of the coroutine stack, which
166 # must be the Task that runs it.
167 break
169 f = f.f_back
170 finally:
171 del f
173 awaited_by = []
174 if future._asyncio_awaited_by:
175 for parent in future._asyncio_awaited_by:
176 awaited_by.append(_build_graph_for_future(parent, limit=limit))
178 if limit is not None:
179 limit *= -1
180 if limit > 0:
181 call_stack = call_stack[:limit]
182 elif limit < 0:
183 call_stack = call_stack[limit:]
185 return FutureCallGraph(future, tuple(call_stack), tuple(awaited_by))
188def format_call_graph(
189 future: futures.Future | None = None,
190 /,
191 *,
192 depth: int = 1,
193 limit: int | None = None,
194) -> str:
195 """Return the async call graph as a string for `future`.
197 If `future` is not provided, format the call graph for the current task.
198 """
200 def render_level(st: FutureCallGraph, buf: list[str], level: int) -> None:
201 def add_line(line: str) -> None:
202 buf.append(level * ' ' + line)
204 if isinstance(st.future, tasks.Task):
205 add_line(
206 f'* Task(name={st.future.get_name()!r}, id={id(st.future):#x})'
207 )
208 else:
209 add_line(
210 f'* Future(id={id(st.future):#x})'
211 )
213 if st.call_stack:
214 add_line(
215 f' + Call stack:'
216 )
217 for ste in st.call_stack:
218 f = ste.frame
220 if f.f_generator is None:
221 f = ste.frame
222 add_line(
223 f' | File {f.f_code.co_filename!r},'
224 f' line {f.f_lineno}, in'
225 f' {f.f_code.co_qualname}()'
226 )
227 else:
228 c = f.f_generator
230 try:
231 f = c.cr_frame
232 code = c.cr_code
233 tag = 'async'
234 except AttributeError:
235 try:
236 f = c.ag_frame
237 code = c.ag_code
238 tag = 'async generator'
239 except AttributeError:
240 f = c.gi_frame
241 code = c.gi_code
242 tag = 'generator'
244 add_line(
245 f' | File {f.f_code.co_filename!r},'
246 f' line {f.f_lineno}, in'
247 f' {tag} {code.co_qualname}()'
248 )
250 if st.awaited_by:
251 add_line(
252 f' + Awaited by:'
253 )
254 for fut in st.awaited_by:
255 render_level(fut, buf, level + 1)
257 graph = capture_call_graph(future, depth=depth + 1, limit=limit)
258 if graph is None:
259 return ""
261 buf: list[str] = []
262 try:
263 render_level(graph, buf, 0)
264 finally:
265 # 'graph' has references to frames so we should
266 # make sure it's GC'ed as soon as we don't need it.
267 del graph
268 return '\n'.join(buf)
270def print_call_graph(
271 future: futures.Future | None = None,
272 /,
273 *,
274 file: io.Writer[str] | None = None,
275 depth: int = 1,
276 limit: int | None = None,
277) -> None:
278 """Print the async call graph for the current task or the provided Future."""
279 # gh-156327: print_call_graph() must not report its own frame
280 print(format_call_graph(future, depth=depth + 1, limit=limit), file=file)