Repository navigation
Expand file tree
/
Copy pathspeedscope.py
More file actions
239 lines (211 loc) · 8.09 KB
/
Copy pathspeedscope.py
File metadata and controls
239 lines (211 loc) · 8.09 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
"""Convert dotnet-trace speedscope exports to the trace interchange format.
``dotnet-trace collect`` samples managed stacks over EventPipe;
``dotnet-trace convert --format speedscope`` exports them as speedscope JSON
whose samples are full stacks, root first. Adjacent in-scope frames within a
stack are caller/callee relationships the sampler actually observed, so
dispatch through interfaces, DI containers, and reflection appears whenever
samples landed there. Counts are sample weights, not call counts.
.NET frames carry no source paths or lines, only
``Assembly!Namespace.Class.Method(arg types)``. Scoping therefore works on
name prefixes (the ``include`` namespaces) rather than file paths, and
frames from runtime assemblies are walked through to the nearest in-scope
ancestor, mirroring the JVM agent's stack walk.
"""
from __future__ import annotations
import json
import math
from typing import TYPE_CHECKING, TypeGuard, cast
from .. import constants as cs
from .records import (
CallRecord,
FramePoint,
TraceFormatError,
TraceHeader,
write_trace_file,
)
if TYPE_CHECKING:
from collections.abc import Sequence
from pathlib import Path
def _scoped_name(name: object, include: Sequence[str]) -> str | None:
"""The frame's assembly-stripped, argument-stripped dotted name, if in scope."""
if not isinstance(name, str):
return None
_assembly, separator, rest = name.partition(cs.TRACE_DOTNET_ASSEMBLY_SEPARATOR)
if not separator:
return None
args_start = rest.find("(")
if args_start >= 0:
rest = rest[:args_start]
for prefix in include:
if rest == prefix or rest.startswith(prefix + cs.SEPARATOR_DOT):
return rest
return None
def _valid_index(value: object, count: int) -> TypeGuard[int]:
"""A plain (non-bool) int that indexes the frame table."""
return isinstance(value, int) and not isinstance(value, bool) and 0 <= value < count
def _sample_weight(weights: list[object], position: int) -> float:
"""The sample's weight, defaulting non-positive or non-numeric values to 1."""
weight = weights[position] if position < len(weights) else 1
if (
not isinstance(weight, int | float)
or isinstance(weight, bool)
or not math.isfinite(weight)
or weight <= 0
):
return 1.0
return float(weight)
def _accumulate_sampled(
profile: dict[str, object],
names: list[str | None],
edges: dict[tuple[str, str], float],
) -> bool:
"""Adjacent in-scope frames in each sampled stack, weighted by sample."""
samples = profile.get("samples", [])
weights = profile.get("weights", [])
if not isinstance(samples, list) or not isinstance(weights, list):
return False
for position, stack in enumerate(samples):
if not isinstance(stack, list):
return False
weight = _sample_weight(cast("list[object]", weights), position)
if not _accumulate_stack(stack, names, weight, edges):
return False
return True
def _accumulate_stack(
stack: Sequence[object],
names: list[str | None],
weight: float,
edges: dict[tuple[str, str], float],
) -> bool:
"""One sampled stack, root first: each in-scope frame under the nearest
in-scope ancestor adds `weight` to that edge. False on a bad frame index."""
ancestor: str | None = None
for frame_index in stack:
if not _valid_index(frame_index, len(names)):
return False
current = names[frame_index]
if current is None:
continue
if ancestor is not None:
key = (ancestor, current)
edges[key] = edges.get(key, 0) + weight
ancestor = current
return True
def _open_event(
event: dict[str, object],
names: list[str | None],
stack: list[str | None],
edges: dict[tuple[str, str], float],
) -> bool:
"""Replay one ``O`` event: validate the frame and count the in-scope edge."""
frame_index = event.get("frame")
if not _valid_index(frame_index, len(names)):
return False
current = names[frame_index]
if current is not None:
ancestor = next((name for name in reversed(stack) if name is not None), None)
if ancestor is not None:
key = (ancestor, current)
edges[key] = edges.get(key, 0) + 1
stack.append(current)
return True
def _accumulate_evented(
profile: dict[str, object],
names: list[str | None],
edges: dict[tuple[str, str], float],
) -> bool:
"""Frame open/close events replayed as a stack; each in-scope open under
an in-scope ancestor counts one observed activation."""
events = profile.get("events")
if not isinstance(events, list):
return False
stack: list[str | None] = []
for raw_event in cast("list[object]", events):
if not isinstance(raw_event, dict):
return False
event = cast("dict[str, object]", raw_event)
kind = event.get("type")
if kind == "O":
if not _open_event(event, names, stack, edges):
return False
elif kind == "C":
if not stack:
return False
stack.pop()
else:
return False
return True
def _accumulate_profiles(
profiles: list[object],
names: list[str | None],
edges: dict[tuple[str, str], float],
profile_path: Path,
) -> bool:
"""Accumulate every recognised profile; raise if none was usable.
Returns True when any accumulated profile was ``sampled`` (its edges are
approximate). An ``evented`` profile records explicit frame open/close
observations, so its edges are exact; a file mixing both is approximate.
"""
converted = False
sampled = False
for raw_profile in profiles:
if not isinstance(raw_profile, dict):
continue
profile = cast("dict[str, object]", raw_profile)
kind = profile.get("type")
if kind == "sampled":
ok = _accumulate_sampled(profile, names, edges)
sampled = sampled or ok
elif kind == "evented":
ok = _accumulate_evented(profile, names, edges)
else:
continue
if not ok:
raise TraceFormatError(
cs.TRACE_ERR_BAD_SPEEDSCOPE.format(path=profile_path)
)
converted = True
if not converted:
raise TraceFormatError(cs.TRACE_ERR_BAD_SPEEDSCOPE.format(path=profile_path))
return sampled
def convert_speedscope(
profile_path: Path,
output: Path,
include: Sequence[str],
workload: str | None = None,
) -> int:
"""Write ``profile_path``'s in-scope call edges to ``output``; returns count."""
raw = json.loads(profile_path.read_text(encoding="utf-8"))
shared = raw.get("shared") if isinstance(raw, dict) else None
frames = shared.get("frames") if isinstance(shared, dict) else None
profiles = raw.get("profiles") if isinstance(raw, dict) else None
if not isinstance(frames, list) or not isinstance(profiles, list):
raise TraceFormatError(cs.TRACE_ERR_BAD_SPEEDSCOPE.format(path=profile_path))
names: list[str | None] = [
_scoped_name(f.get("name") if isinstance(f, dict) else None, include)
for f in frames
]
# Fractional sample weights accumulate as-is and round half-up only at
# emission, so their combined contribution is never truncated away.
edges: dict[tuple[str, str], float] = {}
sampled = _accumulate_profiles(profiles, names, edges, profile_path)
workloads = (workload,) if workload else ()
records = [
CallRecord(
caller=FramePoint(path="", qualname=caller, line=0),
callee=FramePoint(path="", qualname=callee, line=0),
count=max(int(count + 0.5), 1),
workloads=workloads,
receiver_types=(),
)
for (caller, callee), count in edges.items()
]
header = TraceHeader(
version=cs.TRACE_FORMAT_VERSION,
language=cs.TRACE_LANGUAGE_DOTNET,
repo_root="",
tracer=cs.TRACE_TOOL_NAME_SPEEDSCOPE,
sampled=sampled,
)
write_trace_file(output, header, records)
return len(records)