Coverage for src/footman/profile.py: 92%
136 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +0000
1"""Write the run as a profiler trace — `fm --profile … check`.
3Open the file at ui.perfetto.dev (chrome://tracing and speedscope read it
4too).
6`plugin("footman.profile")` in a tasks file switches it on; unmounted, it is
7inert metadata like any other plugin. Mounted, `--profile` writes
8`fm-profile.json` in the invocation's directory and `--profile=FILE` chooses —
9Chrome Trace Event Format, stdlib `json` only, whole-file at `post_tasks`
10when every row is in.
12What the trace shows, from what the run already records: one track per
13worker, a slice per task (queue wait in its args), lane waits at the head of
14the slot, every `run()` step nested inside its task, the task's own
15`section()`/`stream()`/`mark()` records, and a flow arrow per dependency
16edge. Anything that genuinely overlaps on one track — a `parallel()` child's
17steps, a named stream's windows — renders as an async span instead, because
18slices on a track must nest and the trace never lies to make a prettier
19picture. The writer times itself: the last slice is `profile: write`, the
20serialisation cost, appended just before the dump — everything but the file
21write is in the profile.
23Children may add their own detail. A profiled run exports `FM_PROFILE_DIR`
24to every task's environment (so every child inherits it); any process may
25drop Chrome-trace fragments there — `*.json`, a `{"traceEvents": […]}`
26object or a bare event array, `ts` in **epoch microseconds**, `pid` its own
27— and the writer sweeps the directory, shifts each fragment onto the run
28clock, and embeds it as its own process group. footman's pytest plugin
29speaks the convention out of the box: a profiled `run("pytest …")` puts
30every test's setup/call/teardown on the timeline, xdist workers as named
31tracks, under a `pytest` process beside `fm`'s own.
32"""
34from __future__ import annotations
36import atexit
37import json
38import os
39import shutil
40import sys
41import tempfile
42import threading
43import time
44from pathlib import Path
45from typing import Annotated, Any
47import footman
48from footman import GlobalOption, context, matching
49from footman._executor import reported_state
51PROFILE = GlobalOption(
52 "profile",
53 # A trace is JSON, and Tab offering every file in the tree helped nobody.
54 Annotated[Path, matching("*.json")],
55 default=Path("fm-profile.json"),
56 help="write the run as a trace for ui.perfetto.dev",
57)
59_PID = 1
61_child_dir: str | None = None
62"""Where this run's children may drop trace fragments — created at
63`pre_tasks` when the line asked for a profile, swept and removed by the
64writer. Module state, not invocation state: the two hooks are the only
65readers and they bracket one run."""
68@footman.pre_tasks
69def arm(inv: footman.Invocation) -> None:
70 """Open the fragment drop for a profiled run's children.
72 Reads `inv.cli`, not `PROFILE.value`: the manifest refresh child runs
73 this hook with no command line at all, and there the answer must be
74 "not profiling" rather than an unbound-value error.
76 Asks whether the option was *mentioned*, not whether its value is truthy.
77 Those coincided only while a bare mention arrived as `True`; the question
78 was always presence, and a value can be empty and still have been asked
79 for."""
80 global _child_dir
81 _child_dir = None
82 if "profile" not in inv.cli:
83 return
84 _child_dir = tempfile.mkdtemp(prefix="fm-profile-")
85 # The embed pass consumes this directory — but a run that mentions
86 # --profile and never reaches it (a refusal after this hook, an
87 # interrupt) leaked it forever, one directory per mention. The atexit
88 # sweep is the backstop: the consume path clears `_child_dir` first,
89 # so a healthy run's handler finds nothing to do.
90 atexit.register(_sweep_orphan, _child_dir)
91 # The single-threaded moment: what lands in the environment here is in
92 # every task's copy, and so in every child any task spawns.
93 os.environ["FM_PROFILE_DIR"] = _child_dir
96def _sweep_orphan(path: str) -> None:
97 if _child_dir == path:
98 shutil.rmtree(path, ignore_errors=True)
101def _nests(
102 spans: list[dict[str, Any]],
103) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
104 """Split candidate spans for one track into (nesting, overlapping).
106 Chrome `X` events on a tid render by containment, so a span may share a
107 track only when it nests inside whatever is open; a partial overlap — a
108 `parallel()` child's step against its sibling's — goes to the async lane
109 instead. Spans arrive as dicts with `ts`/`dur` already set (µs)."""
110 nested: list[dict[str, Any]] = []
111 async_: list[dict[str, Any]] = []
112 stack: list[float] = [] # open ends, µs
113 for span in sorted(spans, key=lambda s: (s["ts"], -s["dur"])):
114 start, end = span["ts"], span["ts"] + span["dur"]
115 while stack and stack[-1] <= start + 0.001:
116 stack.pop()
117 if not stack or end <= stack[-1] + 0.001:
118 nested.append(span)
119 stack.append(end)
120 else:
121 async_.append(span)
122 return nested, async_
125def _events(results: tuple[Any, ...]) -> tuple[list[dict[str, Any]], float]:
126 rows = [r for r in results if r.started is not None]
127 moments = [r.started for r in rows]
128 for r in rows:
129 moments += [s.started for s in r.steps if s.started is not None]
130 moments += [s.started for s in r.sections]
131 zero = min(moments, default=0.0)
133 def us(clock: float) -> float:
134 return round((clock - zero) * 1e6, 1)
136 events: list[dict[str, Any]] = [
137 {
138 "ph": "M",
139 "name": "process_name",
140 "pid": _PID,
141 "args": {"name": "fm"},
142 }
143 ]
144 named: set[int] = set()
145 flow = 0
146 by_address = {r.address: r for r in rows if r.address}
147 for r in rows:
148 tid = r.thread_id
149 if tid and tid not in named:
150 named.add(tid)
151 events.append(
152 {
153 "ph": "M",
154 "name": "thread_name",
155 "pid": _PID,
156 "tid": tid,
157 "args": {"name": r.thread},
158 }
159 )
160 label = r.address or r.task
161 args: dict[str, Any] = {"state": reported_state(r), "code": r.code}
162 if r.eligible is not None:
163 args["queue_ms"] = round(max(r.started - r.eligible, 0.0) * 1000, 3)
164 events.append(
165 {
166 "ph": "X",
167 "cat": "task",
168 "name": label,
169 "pid": _PID,
170 "tid": tid,
171 "ts": us(r.started),
172 "dur": round(r.duration * 1e6, 1),
173 "args": args,
174 }
175 )
176 cursor = r.started
177 for lane, seconds in r.lane_waits: 177 ↛ 178line 177 didn't jump to line 178 because the loop on line 177 never started
178 events.append(
179 {
180 "ph": "X",
181 "cat": "lane",
182 "name": f"lane: {lane}",
183 "pid": _PID,
184 "tid": tid,
185 "ts": us(cursor),
186 "dur": round(seconds * 1e6, 1),
187 }
188 )
189 cursor += seconds
190 spans = [
191 {
192 "ph": "X",
193 "cat": "step",
194 # A trace is a file that gets attached to tickets and dropped
195 # into ui.perfetto.dev — the shown line, never the record's.
196 "name": s.shown,
197 "pid": _PID,
198 "tid": tid,
199 "ts": us(s.started),
200 "dur": round(s.duration * 1e6, 1),
201 }
202 for s in r.steps
203 if s.started is not None
204 ]
205 for s in r.sections:
206 if s.stream:
207 continue # a named stream is async by design, below
208 if s.duration == 0.0:
209 events.append(
210 {
211 "ph": "i",
212 "s": "t", # thread scope: a tick on this track
213 "cat": "mark",
214 "name": s.name,
215 "pid": _PID,
216 "tid": tid,
217 "ts": us(s.started),
218 }
219 )
220 continue
221 spans.append(
222 {
223 "ph": "X",
224 "cat": "section",
225 "name": s.name,
226 "pid": _PID,
227 "tid": tid,
228 "ts": us(s.started),
229 "dur": round(s.duration * 1e6, 1),
230 }
231 )
232 nested, overlapping = _nests(spans)
233 events += nested
234 streamed = [
235 {
236 "cat": f"stream: {s.stream}",
237 "name": s.name,
238 "ts": us(s.started),
239 "dur": round(s.duration * 1e6, 1),
240 }
241 for s in r.sections
242 if s.stream
243 ]
244 for n, span in enumerate(
245 overlapping + [{**s, "tid": tid} for s in streamed],
246 ):
247 ident = f"{label}#{n}"
248 begin = {
249 "ph": "b",
250 "cat": span.get("cat", "step"),
251 # Process-local scope: a plain `id` is global, and Perfetto
252 # would file the pair under "Global Legacy Events" instead
253 # of with fm's own tracks.
254 "id2": {"local": ident},
255 "name": span["name"],
256 "pid": _PID,
257 "tid": span.get("tid", tid),
258 "ts": span["ts"],
259 }
260 events.append(begin)
261 events.append({**begin, "ph": "e", "ts": span["ts"] + span["dur"]})
262 for dep_addr in r.after:
263 dep = by_address.get(dep_addr)
264 if dep is None or dep.started is None: 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 continue
266 flow += 1
267 done = dep.started + dep.duration
268 events.append(
269 {
270 "ph": "s",
271 "cat": "dep",
272 "id": flow,
273 "name": "after",
274 "pid": _PID,
275 "tid": dep.thread_id,
276 # A hair inside the finishing slice, so the arrow binds
277 # to it rather than to whatever came next on the track.
278 "ts": max(us(dep.started), us(done) - 1.0),
279 }
280 )
281 events.append(
282 {
283 "ph": "f",
284 "bp": "e",
285 "cat": "dep",
286 "id": flow,
287 "name": "after",
288 "pid": _PID,
289 "tid": tid,
290 "ts": us(r.started),
291 }
292 )
293 return events, zero
296def _sweep_children(zero: float) -> list[dict[str, Any]]:
297 """Embed what the run's children dropped in `FM_PROFILE_DIR`.
299 Fragments carry `ts` in epoch microseconds and their own `pid`; the
300 shift onto the run clock goes through the same two-clock anchor the
301 retroactive stream sections use, so a child's timeline lands beside the
302 parent's exactly where it happened. The drop directory is consumed:
303 swept, embedded, removed."""
304 global _child_dir
305 sink, _child_dir = _child_dir, None
306 if sink is None: 306 ↛ 307line 306 didn't jump to line 307 because the condition on line 306 was never true
307 return []
308 os.environ.pop("FM_PROFILE_DIR", None)
309 anchor_wall, anchor_clock = context._WALL_ANCHOR
310 zero_wall_us = (anchor_wall + (zero - anchor_clock)) * 1e6
311 embedded: list[dict[str, Any]] = []
312 for fragment in sorted(Path(sink).glob("*.json")):
313 try:
314 payload = json.loads(fragment.read_text(encoding="utf-8"))
315 except (OSError, ValueError) as exc:
316 print(f"profile: skipped fragment {fragment.name}: {exc}", file=sys.stderr)
317 continue
318 found = payload.get("traceEvents") if isinstance(payload, dict) else payload
319 if not isinstance(found, list): 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true
320 continue
321 for event in found:
322 if not isinstance(event, dict): 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true
323 continue
324 if isinstance(ts := event.get("ts"), (int, float)):
325 event = {**event, "ts": round(ts - zero_wall_us, 1)}
326 embedded.append(event)
327 shutil.rmtree(sink, ignore_errors=True)
328 return embedded
331@footman.post_tasks
332def write(inv: footman.Invocation) -> None:
333 # Three outcomes from one declared value: `.given` says whether a profile
334 # was asked for at all, `.value` says where it goes — the declared default
335 # when `--profile` was named bare, the attached path when it was not.
336 if not PROFILE.given:
337 return
338 target = PROFILE.value
339 begin = time.perf_counter()
340 events, zero = _events(inv.results)
341 events += _sweep_children(zero)
342 path = Path(inv.cwd or ".") / target # an absolute target wins the join
343 tid = threading.get_native_id()
344 events.append(
345 {
346 "ph": "M",
347 "name": "thread_name",
348 "pid": _PID,
349 "tid": tid,
350 "args": {"name": "fm (report)"},
351 }
352 )
353 events.append(
354 {
355 # The writer's own receipt: serialisation, timed to just before
356 # the dump. The file write is the one thing a closed file cannot
357 # contain.
358 "ph": "X",
359 "cat": "profile",
360 "name": "profile: write",
361 "pid": _PID,
362 "tid": tid,
363 "ts": _self_ts(events),
364 "dur": round((time.perf_counter() - begin) * 1e6, 1),
365 }
366 )
367 path.write_text(
368 json.dumps({"traceEvents": events, "displayTimeUnit": "ms"}), encoding="utf-8"
369 )
370 print(f"profile: {path}", file=sys.stderr)
373def _self_ts(events: list[dict[str, Any]]) -> float:
374 """The writer slice's `ts`: past everything already in the trace."""
375 latest = max(
376 (e["ts"] + e.get("dur", 0.0) for e in events if "ts" in e), default=0.0
377 )
378 return round(latest + 10.0, 1)