Skip to navigation

aitune.torch.performance.runtime_profile

View as Markdown

PyTorch Profiler capture helpers for performance profiles.

Module Contents

Classes

NameDescription
_ProfileWarningStructured warning emitted in PerformanceProfile.data["warnings"].
_RegionAggregateSum of one region’s profiler events within one profiled run. Shares are derived at to_dict() time.
_ResidualPortion of one profiled run not attributed to any topmost-owned region, per time domain.
_RunTimingPer-run timing block. device_time_us is None when the profiler reported no device time.
_RuntimeProfileData captured from one PyTorch Profiler run plus its derived analysis.

Functions

NameDescription
_aot_region_idReturn the stable region id for an AOT-managed module name.
_device_time_usReturn positive PyTorch Profiler device time in microseconds when present.
_event_floatRead a profiler event numeric attribute as a float.
_is_cpu_profiler_eventReturn whether a profiler event is a CPU-side event with parent ancestry.
_profiler_activitiesReturn profiler activities supported by the current runtime.
_profiler_device_typeReturn the profiler event device type label.
_untuned_region_idReturn the stable region id for an untuned module path.

Data

PROFILER_KEY_AVERAGES_ROW_LIMIT

API

class aitune.torch.performance.runtime_profile._ProfileWarning

Bases: typing.TypedDict

Structured warning emitted in PerformanceProfile.data["warnings"].

code
str
message
str
source
str
class aitune.torch.performance.runtime_profile._RegionAggregate(
region_id: str,
calls: int = 0,
cpu_time_us: float = 0.0,
device_time_us: float | None = None
)
Dataclass

Sum of one region’s profiler events within one profiled run. Shares are derived at to_dict() time.

calls
int = 0
cpu_time_us
float = 0.0
device_time_us
float | None = None
region_id
str
aitune.torch.performance.runtime_profile._RegionAggregate.accumulate(
event: typing.Any
) -> None

Fold one profiler event into this aggregate.

aitune.torch.performance.runtime_profile._RegionAggregate.to_dict(
) -> dict[str, typing.Any]

Serialize to the JSON-facing dict shape, computing shares relative to the run timing.

class aitune.torch.performance.runtime_profile._Residual(
cpu_time_us: float,
cpu_time_fraction: float,
device_time_us: float | None,
device_time_fraction: float | None
)
Dataclass

Portion of one profiled run not attributed to any topmost-owned region, per time domain.

cpu_time_fraction
float
cpu_time_us
float
device_time_fraction
float | None
device_time_us
float | None
classmethod
aitune.torch.performance.runtime_profile._Residual.to_dict() -> dict[str, typing.Any]

Serialize to the JSON-facing dict shape, omitting unavailable device timing.

class aitune.torch.performance.runtime_profile._RunTiming(
wall_time_us: float,
cpu_time_us: float,
device_time_us: float | None = None
)
Dataclass

Per-run timing block. device_time_us is None when the profiler reported no device time.

cpu_time_us
float
device_time_us
float | None = None
wall_time_us
float
aitune.torch.performance.runtime_profile._RunTiming.to_dict() -> dict[str, typing.Any]

Serialize to the JSON-facing dict shape, omitting unavailable device timing.

class aitune.torch.performance.runtime_profile._RuntimeProfile(
run_wall_times_us: list[float],
events: torch.autograd.profiler_util.EventList,
key_averages: torch.autograd.profiler_util.EventList,
activities: list[str],
untuned_module_types: dict[str, str] = dict()
)
Dataclass

Data captured from one PyTorch Profiler run plus its derived analysis.

_marker_events
list[Any] = field(init=False, repr=False)
_observed_aot_region_names
list[str] = field(init=False, repr=False)
_observed_untuned_region_paths
list[str] = field(init=False, repr=False)
_run_region_aggregates
dict[int, dict[str, _RegionAggregate]] = field(init=False, repr=False)
_unmapped_aot_event_count
int = field(init=False, repr=False)
activities
list[str]
events
EventList
key_averages
EventList
run_wall_times_us
list[float]
untuned_module_types
dict[str, str] = field(default_factory=dict)
aitune.torch.performance.runtime_profile._RuntimeProfile.__post_init__() -> None
aitune.torch.performance.runtime_profile._RuntimeProfile._aot_event_is_nested(
event: typing.Any,
marker_indices: dict[int, int]
) -> bool
staticmethod

Return True if a strict CPU ancestor of event is another AOT-region event.

Walks cpu_parent up to (but not past) the profiled-run marker. The first ancestor matching :data:AOT_MODULE_REGION_PREFIX means this event is already enclosed by another AOT region in this run and its time would be double-counted if aggregated independently.

aitune.torch.performance.runtime_profile._RuntimeProfile._classify_aot_event(
event: typing.Any,
name: str,
marker_indices: dict[int, int],
observed_owned: set[str],
observed_names: list[str]
) -> tuple[int, str] | str | None

Decide what to do with one AOT-prefixed event.

Returns None if the event is nested under another AOT event (skip), the string "unmapped" if it has no profiled-run ancestor, or (run_index, region_id) for aggregation.

aitune.torch.performance.runtime_profile._RuntimeProfile._classify_region_events() -> None

Walk events once and bucket each region event (AOT or untuned) into its run.

Events matching AOT_MODULE_REGION_PREFIX are tracked as AOT regions; events matching UNTUNED_MODULE_REGION_PREFIX are tracked as untuned regions. AOT events that cannot be mapped to a profiled-run ancestor are counted as unmapped; untuned events without a profiled-run ancestor are silently dropped (they fire outside the measured window, e.g. during warmup setup).

Nested AOT events — an AOT-region event whose nearest AOT ancestor (before the profiled-run marker) is also an AOT-region event — are skipped from aggregation: the enclosing AOT region already owns that work, and counting both would double-count and break sum(fractions) + residual ≈ 1.0. The wrapper’s name is still recorded in _observed_aot_region_names only if at least one non-nested event was seen, so a wrapper that ever owned time appears in the regions metadata; one that only fired nested under another does not.

aitune.torch.performance.runtime_profile._RuntimeProfile._classify_untuned_event(
event: typing.Any,
name: str,
marker_indices: dict[int, int],
observed_seen: set[str],
observed_paths: list[str]
) -> tuple[int, str] | None

Decide what to do with one untuned-prefixed event.

aitune.torch.performance.runtime_profile._RuntimeProfile._key_averages_summary(
sort_by: str
) -> dict[str, typing.Any]

Return one bounded view over profiler key averages.

aitune.torch.performance.runtime_profile._RuntimeProfile._owning_run_index(
event: typing.Any,
marker_indices: dict[int, int]
) -> int | None
staticmethod

Return the profiled run index that owns an event, walking CPU parents.

aitune.torch.performance.runtime_profile._RuntimeProfile._run_timing(
index: int,
wall_time_us: float

Build the timing record for one profiled run.

aitune.torch.performance.runtime_profile._RuntimeProfile._serialize_key_average_row(
event: typing.Any
) -> dict[str, typing.Any]
staticmethod

Serialize one PyTorch Profiler key-average row.

aitune.torch.performance.runtime_profile._RuntimeProfile.aot_region_names() -> list[str]

Return observed AOT-managed module region names in profiler event order.

inference_function: collections.abc.Callable,
args: tuple[typing.Any, ...],
kwargs: dict[str, typing.Any],
trace_file: pathlib.Path | None,
warmup_runs: int,
measured_runs: int,
discovery_root: typing.Any | None = None
classmethod

Run warmup + measured iterations under PyTorch Profiler.

PyTorch Profiler drives the iteration phases via torch.profiler.schedule: warmup_runs user-requested warmup iterations during which the profiler is idle (no event capture, no overhead), followed by 1 profiler-internal warmup iteration that absorbs the profiler’s buffer-allocation cost, followed by measured_runs recorded iterations. Only the recorded iterations contribute events to the captured trace and analysis.

discovery_root selects the object from which to discover top-level untuned module candidates for forward-hook instrumentation. Pass None to skip untuned instrumentation entirely (used for pure-callable targets and tests).

aitune.torch.performance.runtime_profile._RuntimeProfile.profiler_summary() -> dict[str, typing.Any]

Return bounded summaries from PyTorch Profiler’s own aggregate events.

aitune.torch.performance.runtime_profile._RuntimeProfile.runs() -> list[dict[str, typing.Any]]

Return per-run timing and aggregated region measurements (AOT and untuned) plus residual.

aitune.torch.performance.runtime_profile._RuntimeProfile.untuned_region_paths() -> list[str]

Return observed untuned module region paths in profiler event order.

aitune.torch.performance.runtime_profile._RuntimeProfile.warnings() -> list[aitune.torch.performance.runtime_profile._ProfileWarning]

Return structured warnings for report integrity issues.

aitune.torch.performance.runtime_profile._aot_region_id(
module_name: str
) -> str

Return the stable region id for an AOT-managed module name.

aitune.torch.performance.runtime_profile._device_time_us(
event: typing.Any
) -> float | None

Return positive PyTorch Profiler device time in microseconds when present.

aitune.torch.performance.runtime_profile._event_float(
event: typing.Any,
attr: str
) -> float

Read a profiler event numeric attribute as a float.

aitune.torch.performance.runtime_profile._is_cpu_profiler_event(
event: typing.Any
) -> bool

Return whether a profiler event is a CPU-side event with parent ancestry.

aitune.torch.performance.runtime_profile._profiler_activities() -> list[torch.profiler.ProfilerActivity]

Return profiler activities supported by the current runtime.

aitune.torch.performance.runtime_profile._profiler_device_type(
event: typing.Any
) -> str

Return the profiler event device type label.

aitune.torch.performance.runtime_profile._untuned_region_id(
module_path: str
) -> str

Return the stable region id for an untuned module path.

aitune.torch.performance.runtime_profile.PROFILER_KEY_AVERAGES_ROW_LIMIT = 20